github-featured/openai--openai-agents-python
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat: #1788 add configurable MCP tool failure handlers (#2378)

Browse Source
This commit is contained in:
Kazuhiro Sera
2026-01-30 09:18:06 +09:00
committed by GitHub
parent 21ac1d28fd
commit 1bc96cda27
5 changed files with 215 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/agent.py
+14 -1
View File
@@ -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]:
src/agents/mcp/server.py
+47 -1
View File
@@ -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
src/agents/mcp/util.py
+57 -24
View File
@@ -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:
tests/mcp/helpers.py
+4 -1
View File
@@ -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] = []
tests/mcp/test_mcp_util.py
+93
View File
@@ -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.