From 1bc96cda274bd784eef26cc27b480f6523dd94b1 Mon Sep 17 00:00:00 2001 From: Kazuhiro Sera Date: Fri, 30 Jan 2026 09:18:06 +0900 Subject: [PATCH] feat: #1788 add configurable MCP tool failure handlers (#2378) --- src/agents/agent.py | 15 +++++- src/agents/mcp/server.py | 48 +++++++++++++++++++- src/agents/mcp/util.py | 81 +++++++++++++++++++++++---------- tests/mcp/helpers.py | 5 +- tests/mcp/test_mcp_util.py | 93 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 215 insertions(+), 27 deletions(-) diff --git a/src/agents/agent.py b/src/agents/agent.py index 492776e2..2136c981 100644 --- a/src/agents/agent.py +++ b/src/agents/agent.py @@ -102,6 +102,12 @@ class MCPConfig(TypedDict): best-effort conversion, so some schemas may not be convertible. Defaults to False. """ + failure_error_function: NotRequired[ToolErrorFunction | None] + """Optional function to convert MCP tool failures into model-visible messages. If explicitly + set to None, tool errors will be raised instead. If unset, defaults to + default_tool_error_function. + """ + @dataclass class AgentBase(Generic[TContext]): @@ -135,8 +141,15 @@ class AgentBase(Generic[TContext]): async def get_mcp_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]: """Fetches the available tools from the MCP servers.""" convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False) + failure_error_function = self.mcp_config.get( + "failure_error_function", default_tool_error_function + ) return await MCPUtil.get_all_function_tools( - self.mcp_servers, convert_schemas_to_strict, run_context, self + self.mcp_servers, + convert_schemas_to_strict, + run_context, + self, + failure_error_function=failure_error_function, ) async def get_all_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]: diff --git a/src/agents/mcp/server.py b/src/agents/mcp/server.py index 2d2c6314..22ecbb53 100644 --- a/src/agents/mcp/server.py +++ b/src/agents/mcp/server.py @@ -8,7 +8,7 @@ from collections.abc import Awaitable from contextlib import AbstractAsyncContextManager, AsyncExitStack from datetime import timedelta from pathlib import Path -from typing import TYPE_CHECKING, Any, Callable, Literal, TypeVar +from typing import TYPE_CHECKING, Any, Callable, Literal, TypeVar, cast import httpx @@ -26,6 +26,7 @@ from typing_extensions import NotRequired, TypedDict from ..exceptions import UserError from ..logger import logger from ..run_context import RunContextWrapper +from ..tool import ToolErrorFunction from ..util._types import MaybeAwaitable from .util import HttpClientFactory, ToolFilter, ToolFilterContext, ToolFilterStatic @@ -48,6 +49,13 @@ RequireApprovalSetting = ( T = TypeVar("T") + +class _UnsetType: + pass + + +_UNSET = _UnsetType() + if TYPE_CHECKING: from ..agent import AgentBase @@ -59,6 +67,7 @@ class MCPServer(abc.ABC): self, use_structured_content: bool = False, require_approval: RequireApprovalSetting = None, + failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET, ): """ Args: @@ -70,11 +79,16 @@ class MCPServer(abc.ABC): require_approval: Approval policy for tools on this server. Accepts "always"/"never", a dict of tool names to those values, a boolean, or an object with always/never tool lists (mirroring TS requireApproval). Normalized into a needs_approval policy. + failure_error_function: Optional function used to convert MCP tool failures into + a model-visible error message. If explicitly set to None, tool errors will be + raised instead of converted. If left unset, the agent-level configuration (or + SDK default) will be used. """ self.use_structured_content = use_structured_content self._needs_approval_policy = self._normalize_needs_approval( require_approval=require_approval ) + self._failure_error_function = failure_error_function @abc.abstractmethod async def connect(self): @@ -207,6 +221,14 @@ class MCPServer(abc.ABC): return bool(policy) + def _get_failure_error_function( + self, agent_failure_error_function: ToolErrorFunction | None + ) -> ToolErrorFunction | None: + """Return the effective error handler for MCP tool failures.""" + if self._failure_error_function is _UNSET: + return agent_failure_error_function + return cast(ToolErrorFunction | None, self._failure_error_function) + class _MCPServerWithClientSession(MCPServer, abc.ABC): """Base class for MCP servers that use a `ClientSession` to communicate with the server.""" @@ -221,6 +243,7 @@ class _MCPServerWithClientSession(MCPServer, abc.ABC): retry_backoff_seconds_base: float = 1.0, message_handler: MessageHandlerFnT | None = None, require_approval: RequireApprovalSetting = None, + failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET, ): """ Args: @@ -247,10 +270,15 @@ class _MCPServerWithClientSession(MCPServer, abc.ABC): require_approval: Approval policy for tools on this server. Accepts "always"/"never", a dict of tool names to those values, a boolean, or an object with always/never tool lists. + failure_error_function: Optional function used to convert MCP tool failures into + a model-visible error message. If explicitly set to None, tool errors will be + raised instead of converted. If left unset, the agent-level configuration (or + SDK default) will be used. """ super().__init__( use_structured_content=use_structured_content, require_approval=require_approval, + failure_error_function=failure_error_function, ) self.session: ClientSession | None = None self.exit_stack: AsyncExitStack = AsyncExitStack() @@ -682,6 +710,7 @@ class MCPServerStdio(_MCPServerWithClientSession): retry_backoff_seconds_base: float = 1.0, message_handler: MessageHandlerFnT | None = None, require_approval: RequireApprovalSetting = None, + failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET, ): """Create a new MCP server based on the stdio transport. @@ -713,6 +742,10 @@ class MCPServerStdio(_MCPServerWithClientSession): ClientSession. require_approval: Approval policy for tools on this server. Accepts "always"/"never", a dict of tool names to those values, or an object with always/never tool lists. + failure_error_function: Optional function used to convert MCP tool failures into + a model-visible error message. If explicitly set to None, tool errors will be + raised instead of converted. If left unset, the agent-level configuration (or + SDK default) will be used. """ super().__init__( cache_tools_list, @@ -723,6 +756,7 @@ class MCPServerStdio(_MCPServerWithClientSession): retry_backoff_seconds_base, message_handler=message_handler, require_approval=require_approval, + failure_error_function=failure_error_function, ) self.params = StdioServerParameters( @@ -788,6 +822,7 @@ class MCPServerSse(_MCPServerWithClientSession): retry_backoff_seconds_base: float = 1.0, message_handler: MessageHandlerFnT | None = None, require_approval: RequireApprovalSetting = None, + failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET, ): """Create a new MCP server based on the HTTP with SSE transport. @@ -821,6 +856,10 @@ class MCPServerSse(_MCPServerWithClientSession): ClientSession. require_approval: Approval policy for tools on this server. Accepts "always"/"never", a dict of tool names to those values, or an object with always/never tool lists. + failure_error_function: Optional function used to convert MCP tool failures into + a model-visible error message. If explicitly set to None, tool errors will be + raised instead of converted. If left unset, the agent-level configuration (or + SDK default) will be used. """ super().__init__( cache_tools_list, @@ -831,6 +870,7 @@ class MCPServerSse(_MCPServerWithClientSession): retry_backoff_seconds_base, message_handler=message_handler, require_approval=require_approval, + failure_error_function=failure_error_function, ) self.params = params @@ -899,6 +939,7 @@ class MCPServerStreamableHttp(_MCPServerWithClientSession): retry_backoff_seconds_base: float = 1.0, message_handler: MessageHandlerFnT | None = None, require_approval: RequireApprovalSetting = None, + failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET, ): """Create a new MCP server based on the Streamable HTTP transport. @@ -933,6 +974,10 @@ class MCPServerStreamableHttp(_MCPServerWithClientSession): ClientSession. require_approval: Approval policy for tools on this server. Accepts "always"/"never", a dict of tool names to those values, or an object with always/never tool lists. + failure_error_function: Optional function used to convert MCP tool failures into + a model-visible error message. If explicitly set to None, tool errors will be + raised instead of converted. If left unset, the agent-level configuration (or + SDK default) will be used. """ super().__init__( cache_tools_list, @@ -943,6 +988,7 @@ class MCPServerStreamableHttp(_MCPServerWithClientSession): retry_backoff_seconds_base, message_handler=message_handler, require_approval=require_approval, + failure_error_function=failure_error_function, ) self.params = params diff --git a/src/agents/mcp/util.py b/src/agents/mcp/util.py index bbcd8159..411c64ef 100644 --- a/src/agents/mcp/util.py +++ b/src/agents/mcp/util.py @@ -1,8 +1,10 @@ +from __future__ import annotations + import functools import inspect import json from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Callable, Optional, Protocol, Union +from typing import TYPE_CHECKING, Any, Callable, Protocol, Union import httpx from typing_extensions import NotRequired, TypedDict @@ -15,6 +17,7 @@ from ..strict_schema import ensure_strict_json_schema from ..tool import ( FunctionTool, Tool, + ToolErrorFunction, ToolOutputImageDict, ToolOutputTextDict, default_tool_error_function, @@ -24,8 +27,12 @@ from ..tracing import FunctionSpanData, SpanError, get_current_span, mcp_tools_s from ..util import _error_tracing from ..util._types import MaybeAwaitable -ToolOutputItem = Union[ToolOutputTextDict, ToolOutputImageDict] -ToolOutput = Union[str, ToolOutputItem, list[ToolOutputItem]] +if TYPE_CHECKING: + ToolOutputItem = ToolOutputTextDict | ToolOutputImageDict + ToolOutput = str | ToolOutputItem | list[ToolOutputItem] +else: + ToolOutputItem = Union[ToolOutputTextDict, ToolOutputImageDict] # noqa: UP007 + ToolOutput = Union[str, ToolOutputItem, list[ToolOutputItem]] # noqa: UP007 if TYPE_CHECKING: from mcp.types import Tool as MCPTool @@ -43,9 +50,9 @@ class HttpClientFactory(Protocol): def __call__( self, - headers: Optional[dict[str, str]] = None, - timeout: Optional[httpx.Timeout] = None, - auth: Optional[httpx.Auth] = None, + headers: dict[str, str] | None = None, + timeout: httpx.Timeout | None = None, + auth: httpx.Auth | None = None, ) -> httpx.AsyncClient: ... @@ -56,14 +63,17 @@ class ToolFilterContext: run_context: RunContextWrapper[Any] """The current run context.""" - agent: "AgentBase" + agent: AgentBase """The agent that is requesting the tool list.""" server_name: str """The name of the MCP server.""" -ToolFilterCallable = Callable[["ToolFilterContext", "MCPTool"], MaybeAwaitable[bool]] +if TYPE_CHECKING: + ToolFilterCallable = Callable[[ToolFilterContext, MCPTool], MaybeAwaitable[bool]] +else: + ToolFilterCallable = Callable[[ToolFilterContext, Any], MaybeAwaitable[bool]] """A function that determines whether a tool should be available. Args: @@ -87,14 +97,17 @@ class ToolFilterStatic(TypedDict): If set, these tools will be filtered out.""" -ToolFilter = Union[ToolFilterCallable, ToolFilterStatic, None] +if TYPE_CHECKING: + ToolFilter = ToolFilterCallable | ToolFilterStatic | None +else: + ToolFilter = Union[ToolFilterCallable, ToolFilterStatic, None] # noqa: UP007 """A tool filter that can be either a function, static configuration, or None (no filtering).""" def create_static_tool_filter( - allowed_tool_names: Optional[list[str]] = None, - blocked_tool_names: Optional[list[str]] = None, -) -> Optional[ToolFilterStatic]: + allowed_tool_names: list[str] | None = None, + blocked_tool_names: list[str] | None = None, +) -> ToolFilterStatic | None: """Create a static tool filter from allowlist and blocklist parameters. This is a convenience function for creating a ToolFilterStatic. @@ -124,17 +137,22 @@ class MCPUtil: @classmethod async def get_all_function_tools( cls, - servers: list["MCPServer"], + servers: list[MCPServer], convert_schemas_to_strict: bool, run_context: RunContextWrapper[Any], - agent: "AgentBase", + agent: AgentBase, + failure_error_function: ToolErrorFunction | None = default_tool_error_function, ) -> list[Tool]: """Get all function tools from a list of MCP servers.""" tools = [] tool_names: set[str] = set() for server in servers: server_tools = await cls.get_function_tools( - server, convert_schemas_to_strict, run_context, agent + server, + convert_schemas_to_strict, + run_context, + agent, + failure_error_function=failure_error_function, ) server_tool_names = {tool.name for tool in server_tools} if len(server_tool_names & tool_names) > 0: @@ -150,10 +168,11 @@ class MCPUtil: @classmethod async def get_function_tools( cls, - server: "MCPServer", + server: MCPServer, convert_schemas_to_strict: bool, run_context: RunContextWrapper[Any], - agent: "AgentBase", + agent: AgentBase, + failure_error_function: ToolErrorFunction | None = default_tool_error_function, ) -> list[Tool]: """Get all function tools from a single MCP server.""" @@ -162,19 +181,30 @@ class MCPUtil: span.span_data.result = [tool.name for tool in tools] return [ - cls.to_function_tool(tool, server, convert_schemas_to_strict, agent) for tool in tools + cls.to_function_tool( + tool, + server, + convert_schemas_to_strict, + agent, + failure_error_function=failure_error_function, + ) + for tool in tools ] @classmethod def to_function_tool( cls, - tool: "MCPTool", - server: "MCPServer", + tool: MCPTool, + server: MCPServer, convert_schemas_to_strict: bool, - agent: "AgentBase", + agent: AgentBase, + failure_error_function: ToolErrorFunction | None = default_tool_error_function, ) -> FunctionTool: """Convert an MCP tool to an Agents SDK function tool.""" invoke_func_impl = functools.partial(cls.invoke_mcp_tool, server, tool) + effective_failure_error_function = server._get_failure_error_function( + failure_error_function + ) schema, is_strict = tool.inputSchema, False # MCP spec doesn't require the inputSchema to have `properties`, but OpenAI spec does. @@ -195,8 +225,11 @@ class MCPUtil: try: return await invoke_func_impl(ctx, input_json) except Exception as e: - # Use default error handling function to convert exception to error message. - result = default_tool_error_function(ctx, e) + if effective_failure_error_function is None: + raise + + # Use configured error handling function to convert exception to error message. + result = effective_failure_error_function(ctx, e) if inspect.isawaitable(result): result = await result @@ -233,7 +266,7 @@ class MCPUtil: @classmethod async def invoke_mcp_tool( - cls, server: "MCPServer", tool: "MCPTool", context: RunContextWrapper[Any], input_json: str + cls, server: MCPServer, tool: MCPTool, context: RunContextWrapper[Any], input_json: str ) -> ToolOutput: """Invoke an MCP tool and return the result as a string.""" try: diff --git a/tests/mcp/helpers.py b/tests/mcp/helpers.py index 6eced1e9..5318800b 100644 --- a/tests/mcp/helpers.py +++ b/tests/mcp/helpers.py @@ -16,8 +16,9 @@ from mcp.types import ( ) from agents.mcp import MCPServer -from agents.mcp.server import _MCPServerWithClientSession +from agents.mcp.server import _UNSET, _MCPServerWithClientSession, _UnsetType from agents.mcp.util import ToolFilter +from agents.tool import ToolErrorFunction tee = shutil.which("tee") or "" assert tee, "tee not found" @@ -70,10 +71,12 @@ class FakeMCPServer(MCPServer): tool_filter: ToolFilter = None, server_name: str = "fake_mcp_server", require_approval: object | None = None, + failure_error_function: ToolErrorFunction | None | _UnsetType = _UNSET, ): super().__init__( use_structured_content=False, require_approval=require_approval, # type: ignore[arg-type] + failure_error_function=failure_error_function, ) self.tools: list[MCPTool] = tools or [] self.tool_calls: list[str] = [] diff --git a/tests/mcp/test_mcp_util.py b/tests/mcp/test_mcp_util.py index a9a8b259..9d9822c3 100644 --- a/tests/mcp/test_mcp_util.py +++ b/tests/mcp/test_mcp_util.py @@ -225,6 +225,99 @@ async def test_mcp_tool_timeout_handling(): assert "Timed out" in result +@pytest.mark.asyncio +async def test_mcp_tool_failure_error_function_agent_default(): + """Agent-level failure_error_function should handle MCP tool failures.""" + + def custom_failure(_ctx: RunContextWrapper[Any], _exc: Exception) -> str: + return "custom_mcp_failure" + + server = CrashingFakeMCPServer() + server.add_tool("crashing_tool", {}) + + agent = Agent( + name="test-agent", + mcp_servers=[server], + mcp_config={"failure_error_function": custom_failure}, + ) + run_context = RunContextWrapper(context=None) + tools = await agent.get_mcp_tools(run_context) + function_tool = next(tool for tool in tools if tool.name == "crashing_tool") + assert isinstance(function_tool, FunctionTool) + + tool_context = ToolContext( + context=None, + tool_name="crashing_tool", + tool_call_id="test_call_custom_1", + tool_arguments="{}", + ) + + result = await function_tool.on_invoke_tool(tool_context, "{}") + assert result == "custom_mcp_failure" + + +@pytest.mark.asyncio +async def test_mcp_tool_failure_error_function_server_override(): + """Server-level failure_error_function should override agent defaults.""" + + def agent_failure(_ctx: RunContextWrapper[Any], _exc: Exception) -> str: + return "agent_failure" + + def server_failure(_ctx: RunContextWrapper[Any], _exc: Exception) -> str: + return "server_failure" + + server = CrashingFakeMCPServer(failure_error_function=server_failure) + server.add_tool("crashing_tool", {}) + + agent = Agent( + name="test-agent", + mcp_servers=[server], + mcp_config={"failure_error_function": agent_failure}, + ) + run_context = RunContextWrapper(context=None) + tools = await agent.get_mcp_tools(run_context) + function_tool = next(tool for tool in tools if tool.name == "crashing_tool") + assert isinstance(function_tool, FunctionTool) + + tool_context = ToolContext( + context=None, + tool_name="crashing_tool", + tool_call_id="test_call_custom_2", + tool_arguments="{}", + ) + + result = await function_tool.on_invoke_tool(tool_context, "{}") + assert result == "server_failure" + + +@pytest.mark.asyncio +async def test_mcp_tool_failure_error_function_server_none_raises(): + """Server-level None should re-raise MCP tool failures.""" + + server = CrashingFakeMCPServer(failure_error_function=None) + server.add_tool("crashing_tool", {}) + + agent = Agent( + name="test-agent", + mcp_servers=[server], + mcp_config={"failure_error_function": default_tool_error_function}, + ) + run_context = RunContextWrapper(context=None) + tools = await agent.get_mcp_tools(run_context) + function_tool = next(tool for tool in tools if tool.name == "crashing_tool") + assert isinstance(function_tool, FunctionTool) + + tool_context = ToolContext( + context=None, + tool_name="crashing_tool", + tool_call_id="test_call_custom_3", + tool_arguments="{}", + ) + + with pytest.raises(AgentsException): + await function_tool.on_invoke_tool(tool_context, "{}") + + @pytest.mark.asyncio async def test_agent_convert_schemas_true(): """Test that setting convert_schemas_to_strict to True converts non-strict schemas to strict.