MCP Util

ToolFilterCallable 模块属性

ToolFilterCallable = Callable[
    ["ToolFilterContext", "MCPTool"], MaybeAwaitable[bool]
]

一个确定工具是否应该可用的函数。

参数

名称	类型	描述	默认
context		包含运行上下文、代理和服务器名称的上下文信息。	required
tool		要过滤的 MCP 工具。	required

返回值

类型	描述
	工具是否应该可用（True）或被过滤掉（False）。

ToolFilter 模块属性

ToolFilter = Union[
    ToolFilterCallable, ToolFilterStatic, None
]

一个工具过滤器，可以是函数、静态配置或 None（无过滤）。

HttpClientFactory

Bases: Protocol

HTTP 客户端工厂函数的协议。

此接口与 MCP SDK 的 McpHttpClientFactory 匹配，但定义在本地以避免访问内部 MCP SDK 模块。

源代码位于 src/agents/mcp/util.py

class HttpClientFactory(Protocol):
    """Protocol for HTTP client factory functions.

    This interface matches the MCP SDK's McpHttpClientFactory but is defined locally
    to avoid accessing internal MCP SDK modules.
    """

    def __call__(
        self,
        headers: Optional[dict[str, str]] = None,
        timeout: Optional[httpx.Timeout] = None,
        auth: Optional[httpx.Auth] = None,
    ) -> httpx.AsyncClient: ...

ToolFilterContext 数据类

上下文信息，可供工具筛选函数使用。

源代码位于 src/agents/mcp/util.py

@dataclass
class ToolFilterContext:
    """Context information available to tool filter functions."""

    run_context: RunContextWrapper[Any]
    """The current run context."""

    agent: "AgentBase"
    """The agent that is requesting the tool list."""

    server_name: str
    """The name of the MCP server."""

run_context 实例属性

run_context: RunContextWrapper[Any]

当前的运行上下文。

agent 实例属性

agent: AgentBase

请求工具列表的代理。

server_name 实例属性

server_name: str

MCP 服务器的名称。

ToolFilterStatic

基础: TypedDict

使用允许列表和阻止列表的静态工具过滤器配置。

源代码位于 src/agents/mcp/util.py

class ToolFilterStatic(TypedDict):
    """Static tool filter configuration using allowlists and blocklists."""

    allowed_tool_names: NotRequired[list[str]]
    """Optional list of tool names to allow (whitelist).
    If set, only these tools will be available."""

    blocked_tool_names: NotRequired[list[str]]
    """Optional list of tool names to exclude (blacklist).
    If set, these tools will be filtered out."""

allowed_tool_names 实例属性

allowed_tool_names: NotRequired[list[str]]

允许的工具名称的可选列表（白名单）。如果设置，只有这些工具可用。

blocked_tool_names 实例属性

blocked_tool_names: NotRequired[list[str]]

要排除的工具名称的可选列表（黑名单）。如果设置，这些工具将被过滤掉。

MCPUtil

MCP 和 Agents SDK 工具之间互操作的实用工具集。

源代码位于 src/agents/mcp/util.py

class MCPUtil:
    """Set of utilities for interop between MCP and Agents SDK tools."""

    @classmethod
    async def get_all_function_tools(
        cls,
        servers: list["MCPServer"],
        convert_schemas_to_strict: bool,
        run_context: RunContextWrapper[Any],
        agent: "AgentBase",
    ) -> 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_tool_names = {tool.name for tool in server_tools}
            if len(server_tool_names & tool_names) > 0:
                raise UserError(
                    f"Duplicate tool names found across MCP servers: "
                    f"{server_tool_names & tool_names}"
                )
            tool_names.update(server_tool_names)
            tools.extend(server_tools)

        return tools

    @classmethod
    async def get_function_tools(
        cls,
        server: "MCPServer",
        convert_schemas_to_strict: bool,
        run_context: RunContextWrapper[Any],
        agent: "AgentBase",
    ) -> list[Tool]:
        """Get all function tools from a single MCP server."""

        with mcp_tools_span(server=server.name) as span:
            tools = await server.list_tools(run_context, agent)
            span.span_data.result = [tool.name for tool in tools]

        return [cls.to_function_tool(tool, server, convert_schemas_to_strict) for tool in tools]

    @classmethod
    def to_function_tool(
        cls, tool: "MCPTool", server: "MCPServer", convert_schemas_to_strict: bool
    ) -> FunctionTool:
        """Convert an MCP tool to an Agents SDK function tool."""
        invoke_func = functools.partial(cls.invoke_mcp_tool, server, tool)
        schema, is_strict = tool.inputSchema, False

        # MCP spec doesn't require the inputSchema to have `properties`, but OpenAI spec does.
        if "properties" not in schema:
            schema["properties"] = {}

        if convert_schemas_to_strict:
            try:
                schema = ensure_strict_json_schema(schema)
                is_strict = True
            except Exception as e:
                logger.info(f"Error converting MCP schema to strict mode: {e}")

        return FunctionTool(
            name=tool.name,
            description=tool.description or "",
            params_json_schema=schema,
            on_invoke_tool=invoke_func,
            strict_json_schema=is_strict,
        )

    @classmethod
    async def invoke_mcp_tool(
        cls, server: "MCPServer", tool: "MCPTool", context: RunContextWrapper[Any], input_json: str
    ) -> str:
        """Invoke an MCP tool and return the result as a string."""
        try:
            json_data: dict[str, Any] = json.loads(input_json) if input_json else {}
        except Exception as e:
            if _debug.DONT_LOG_TOOL_DATA:
                logger.debug(f"Invalid JSON input for tool {tool.name}")
            else:
                logger.debug(f"Invalid JSON input for tool {tool.name}: {input_json}")
            raise ModelBehaviorError(
                f"Invalid JSON input for tool {tool.name}: {input_json}"
            ) from e

        if _debug.DONT_LOG_TOOL_DATA:
            logger.debug(f"Invoking MCP tool {tool.name}")
        else:
            logger.debug(f"Invoking MCP tool {tool.name} with input {input_json}")

        try:
            result = await server.call_tool(tool.name, json_data)
        except Exception as e:
            logger.error(f"Error invoking MCP tool {tool.name}: {e}")
            raise AgentsException(f"Error invoking MCP tool {tool.name}: {e}") from e

        if _debug.DONT_LOG_TOOL_DATA:
            logger.debug(f"MCP tool {tool.name} completed.")
        else:
            logger.debug(f"MCP tool {tool.name} returned {result}")

        # If structured content is requested and available, use it exclusively
        if server.use_structured_content and result.structuredContent:
            tool_output = json.dumps(result.structuredContent)
        else:
            # Fall back to regular text content processing
            # The MCP tool result is a list of content items, whereas OpenAI tool
            # outputs are a single string. We'll try to convert.
            if len(result.content) == 1:
                tool_output = result.content[0].model_dump_json()
            elif len(result.content) > 1:
                tool_results = [item.model_dump(mode="json") for item in result.content]
                tool_output = json.dumps(tool_results)
            else:
                # Empty content is a valid result (e.g., "no results found")
                tool_output = "[]"

        current_span = get_current_span()
        if current_span:
            if isinstance(current_span.span_data, FunctionSpanData):
                current_span.span_data.output = tool_output
                current_span.span_data.mcp_data = {
                    "server": server.name,
                }
            else:
                logger.warning(
                    f"Current span is not a FunctionSpanData, skipping tool output: {current_span}"
                )

        return tool_output

get_all_function_tools 异步 类方法

get_all_function_tools(
    servers: list[MCPServer],
    convert_schemas_to_strict: bool,
    run_context: RunContextWrapper[Any],
    agent: AgentBase,
) -> list[Tool]

从 MCP 服务器列表中获取所有函数工具。

源代码位于 src/agents/mcp/util.py

@classmethod
async def get_all_function_tools(
    cls,
    servers: list["MCPServer"],
    convert_schemas_to_strict: bool,
    run_context: RunContextWrapper[Any],
    agent: "AgentBase",
) -> 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_tool_names = {tool.name for tool in server_tools}
        if len(server_tool_names & tool_names) > 0:
            raise UserError(
                f"Duplicate tool names found across MCP servers: "
                f"{server_tool_names & tool_names}"
            )
        tool_names.update(server_tool_names)
        tools.extend(server_tools)

    return tools

get_function_tools 异步 类方法

get_function_tools(
    server: MCPServer,
    convert_schemas_to_strict: bool,
    run_context: RunContextWrapper[Any],
    agent: AgentBase,
) -> list[Tool]

从单个 MCP 服务器获取所有函数工具。

源代码位于 src/agents/mcp/util.py

@classmethod
async def get_function_tools(
    cls,
    server: "MCPServer",
    convert_schemas_to_strict: bool,
    run_context: RunContextWrapper[Any],
    agent: "AgentBase",
) -> list[Tool]:
    """Get all function tools from a single MCP server."""

    with mcp_tools_span(server=server.name) as span:
        tools = await server.list_tools(run_context, agent)
        span.span_data.result = [tool.name for tool in tools]

    return [cls.to_function_tool(tool, server, convert_schemas_to_strict) for tool in tools]

to_function_tool 类方法

to_function_tool(
    tool: Tool,
    server: MCPServer,
    convert_schemas_to_strict: bool,
) -> FunctionTool

将 MCP 工具转换为 Agents SDK 函数工具。

源代码位于 src/agents/mcp/util.py

@classmethod
def to_function_tool(
    cls, tool: "MCPTool", server: "MCPServer", convert_schemas_to_strict: bool
) -> FunctionTool:
    """Convert an MCP tool to an Agents SDK function tool."""
    invoke_func = functools.partial(cls.invoke_mcp_tool, server, tool)
    schema, is_strict = tool.inputSchema, False

    # MCP spec doesn't require the inputSchema to have `properties`, but OpenAI spec does.
    if "properties" not in schema:
        schema["properties"] = {}

    if convert_schemas_to_strict:
        try:
            schema = ensure_strict_json_schema(schema)
            is_strict = True
        except Exception as e:
            logger.info(f"Error converting MCP schema to strict mode: {e}")

    return FunctionTool(
        name=tool.name,
        description=tool.description or "",
        params_json_schema=schema,
        on_invoke_tool=invoke_func,
        strict_json_schema=is_strict,
    )

invoke_mcp_tool 异步 类方法

invoke_mcp_tool(
    server: MCPServer,
    tool: Tool,
    context: RunContextWrapper[Any],
    input_json: str,
) -> str

调用 MCP 工具并以字符串形式返回结果。

源代码位于 src/agents/mcp/util.py

@classmethod
async def invoke_mcp_tool(
    cls, server: "MCPServer", tool: "MCPTool", context: RunContextWrapper[Any], input_json: str
) -> str:
    """Invoke an MCP tool and return the result as a string."""
    try:
        json_data: dict[str, Any] = json.loads(input_json) if input_json else {}
    except Exception as e:
        if _debug.DONT_LOG_TOOL_DATA:
            logger.debug(f"Invalid JSON input for tool {tool.name}")
        else:
            logger.debug(f"Invalid JSON input for tool {tool.name}: {input_json}")
        raise ModelBehaviorError(
            f"Invalid JSON input for tool {tool.name}: {input_json}"
        ) from e

    if _debug.DONT_LOG_TOOL_DATA:
        logger.debug(f"Invoking MCP tool {tool.name}")
    else:
        logger.debug(f"Invoking MCP tool {tool.name} with input {input_json}")

    try:
        result = await server.call_tool(tool.name, json_data)
    except Exception as e:
        logger.error(f"Error invoking MCP tool {tool.name}: {e}")
        raise AgentsException(f"Error invoking MCP tool {tool.name}: {e}") from e

    if _debug.DONT_LOG_TOOL_DATA:
        logger.debug(f"MCP tool {tool.name} completed.")
    else:
        logger.debug(f"MCP tool {tool.name} returned {result}")

    # If structured content is requested and available, use it exclusively
    if server.use_structured_content and result.structuredContent:
        tool_output = json.dumps(result.structuredContent)
    else:
        # Fall back to regular text content processing
        # The MCP tool result is a list of content items, whereas OpenAI tool
        # outputs are a single string. We'll try to convert.
        if len(result.content) == 1:
            tool_output = result.content[0].model_dump_json()
        elif len(result.content) > 1:
            tool_results = [item.model_dump(mode="json") for item in result.content]
            tool_output = json.dumps(tool_results)
        else:
            # Empty content is a valid result (e.g., "no results found")
            tool_output = "[]"

    current_span = get_current_span()
    if current_span:
        if isinstance(current_span.span_data, FunctionSpanData):
            current_span.span_data.output = tool_output
            current_span.span_data.mcp_data = {
                "server": server.name,
            }
        else:
            logger.warning(
                f"Current span is not a FunctionSpanData, skipping tool output: {current_span}"
            )

    return tool_output

create_static_tool_filter

create_static_tool_filter(
    allowed_tool_names: Optional[list[str]] = None,
    blocked_tool_names: Optional[list[str]] = None,
) -> Optional[ToolFilterStatic]

从允许列表和阻止列表参数创建静态工具过滤器。

这是创建 ToolFilterStatic 的便捷函数。

参数

名称	类型	描述	默认
allowed_tool_names	可选[列表[字符串]]	允许的工具名称的可选列表（白名单）。	None
blocked_tool_names	可选[列表[字符串]]	要排除的工具名称的可选列表（黑名单）。	None

返回值

类型	描述
可选[ToolFilterStatic]	如果指定了任何过滤，则为 ToolFilterStatic，否则为 None。

源代码位于 src/agents/mcp/util.py

def create_static_tool_filter(
    allowed_tool_names: Optional[list[str]] = None,
    blocked_tool_names: Optional[list[str]] = None,
) -> Optional[ToolFilterStatic]:
    """Create a static tool filter from allowlist and blocklist parameters.

    This is a convenience function for creating a ToolFilterStatic.

    Args:
        allowed_tool_names: Optional list of tool names to allow (whitelist).
        blocked_tool_names: Optional list of tool names to exclude (blacklist).

    Returns:
        A ToolFilterStatic if any filtering is specified, None otherwise.
    """
    if allowed_tool_names is None and blocked_tool_names is None:
        return None

    filter_dict: ToolFilterStatic = {}
    if allowed_tool_names is not None:
        filter_dict["allowed_tool_names"] = allowed_tool_names
    if blocked_tool_names is not None:
        filter_dict["blocked_tool_names"] = blocked_tool_names

    return filter_dict