Model context protocol (MCP)

Model context protocol（MCP）标准化了应用向语言模型暴露工具和上下文的方式。来自官方文档：

MCP 是一种开放协议，用于标准化应用如何向 LLM 提供上下文。可以把 MCP 想象成 AI 应用的 USB-C 接口。 就像 USB-C 提供了一种将设备连接到各种外设和配件的标准方式一样，MCP 也提供了一种将 AI 模型连接到不同数据源和工具的标准方式。

Agents Python SDK 支持多种 MCP 传输方式。这使你能够复用现有的 MCP 服务，或构建自己的服务，以向智能体暴露基于文件系统、HTTP 或连接器的工具。

MCP 集成选择

在将 MCP 服务接入智能体之前，请先决定工具调用应在何处执行，以及你可访问哪些传输方式。下表总结了 Python SDK 支持的选项。

你需要的能力	推荐选项
让 OpenAI 的 Responses API 代表模型调用可公开访问的 MCP 服务	通过 HostedMCPTool 使用Hosted MCP server tools
连接你在本地或远程运行的 Streamable HTTP 服务	通过 MCPServerStreamableHttp 使用Streamable HTTP MCP servers
与实现了 HTTP + Server-Sent Events 的服务通信	通过 MCPServerSse 使用HTTP with SSE MCP servers
启动本地进程并通过 stdin/stdout 通信	通过 MCPServerStdio 使用stdio MCP servers

下面的章节将逐一介绍每种选项、如何配置，以及何时优先选择某种传输方式。

智能体级 MCP 配置

除了选择传输方式外，你还可以通过设置 Agent.mcp_config 来调整 MCP 工具的准备方式。

from agents import Agent

agent = Agent(
    name="Assistant",
    mcp_servers=[server],
    mcp_config={
        # Try to convert MCP tool schemas to strict JSON schema.
        "convert_schemas_to_strict": True,
        # If None, MCP tool failures are raised as exceptions instead of
        # returning model-visible error text.
        "failure_error_function": None,
    },
)

注意：

• convert_schemas_to_strict 为尽力而为模式。如果某个 schema 无法转换，则使用原始 schema。
• failure_error_function 控制如何将 MCP 工具调用失败反馈给模型。
• 当未设置 failure_error_function 时，SDK 使用默认工具错误格式化器。
• 服务级别的 failure_error_function 会覆盖该服务上的 Agent.mcp_config["failure_error_function"]。

传输方式间的共享模式

选择传输方式后，大多数集成都需要做出相同的后续决策：

• 如何只暴露部分工具（工具过滤）。
• 服务是否还提供可复用提示词（Prompts）。
• 是否应缓存 list_tools()（缓存）。
• MCP 活动在追踪中的呈现方式（追踪）。

对于本地 MCP 服务（MCPServerStdio、MCPServerSse、MCPServerStreamableHttp），审批策略和每次调用的 _meta 负载也是共享概念。Streamable HTTP 章节展示了最完整的示例，相同模式也适用于其他本地传输方式。

1. Hosted MCP server tools

Hosted 工具将完整的工具往返流程放入 OpenAI 基础设施中。你的代码无需列举和调用工具， HostedMCPTool 会将服务标签（以及可选连接器元数据）转发给 Responses API。模型会列出远程服务的工具并调用它们，而无需额外回调到你的 Python 进程。Hosted 工具目前适用于支持 Responses API Hosted MCP 集成的 OpenAI 模型。

基本 Hosted MCP 工具

通过向智能体的 tools 列表添加 HostedMCPTool 来创建 Hosted 工具。tool_config 字典对应你发送到 REST API 的 JSON：

import asyncio

from agents import Agent, HostedMCPTool, Runner

async def main() -> None:
    agent = Agent(
        name="Assistant",
        tools=[
            HostedMCPTool(
                tool_config={
                    "type": "mcp",
                    "server_label": "gitmcp",
                    "server_url": "https://gitmcp.io/openai/codex",
                    "require_approval": "never",
                }
            )
        ],
    )

    result = await Runner.run(agent, "Which language is this repository written in?")
    print(result.final_output)

asyncio.run(main())

Hosted 服务会自动暴露其工具；你不需要将其添加到 mcp_servers。

如果你希望 Hosted 工具检索以延迟方式加载 Hosted MCP 服务，请设置 tool_config["defer_loading"] = True 并将 ToolSearchTool 添加到智能体。这仅在 OpenAI Responses 模型上受支持。完整的工具检索设置与限制请参见 Tools。

流式输出 Hosted MCP 结果

Hosted 工具支持与工具调用完全相同的流式结果。使用 Runner.run_streamed 在模型仍在运行时 消费增量 MCP 输出：

result = Runner.run_streamed(agent, "Summarise this repository's top languages")
async for event in result.stream_events():
    if event.type == "run_item_stream_event":
        print(f"Received: {event.item}")
print(result.final_output)

可选审批流程

如果某个服务可以执行敏感操作，你可以在每次工具执行前要求人工或程序化审批。在 tool_config 中配置 require_approval，可使用单一策略（"always"、"never"）或按工具名映射到策略的字典。若要在 Python 内做决策，请提供 on_approval_request 回调。

from agents import MCPToolApprovalFunctionResult, MCPToolApprovalRequest

SAFE_TOOLS = {"read_project_metadata"}

def approve_tool(request: MCPToolApprovalRequest) -> MCPToolApprovalFunctionResult:
    if request.data.name in SAFE_TOOLS:
        return {"approve": True}
    return {"approve": False, "reason": "Escalate to a human reviewer"}

agent = Agent(
    name="Assistant",
    tools=[
        HostedMCPTool(
            tool_config={
                "type": "mcp",
                "server_label": "gitmcp",
                "server_url": "https://gitmcp.io/openai/codex",
                "require_approval": "always",
            },
            on_approval_request=approve_tool,
        )
    ],
)

该回调可以是同步或异步的，并且会在模型需要审批数据以继续运行时触发。

基于连接器的 Hosted 服务

Hosted MCP 也支持 OpenAI 连接器。你可以不指定 server_url，改为提供 connector_id 和访问令牌。Responses API 会处理认证，Hosted 服务将暴露该连接器的工具。

import os

HostedMCPTool(
    tool_config={
        "type": "mcp",
        "server_label": "google_calendar",
        "connector_id": "connector_googlecalendar",
        "authorization": os.environ["GOOGLE_CALENDAR_AUTHORIZATION"],
        "require_approval": "never",
    }
)

完整可运行的 Hosted 工具示例（包括流式传输、审批和连接器）位于 examples/hosted_mcp。

2. Streamable HTTP MCP servers

当你希望自行管理网络连接时，请使用 MCPServerStreamableHttp。当你控制传输层，或希望在自有基础设施中运行服务并保持低延迟时，Streamable HTTP 服务是理想选择。

import asyncio
import os

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp
from agents.model_settings import ModelSettings

async def main() -> None:
    token = os.environ["MCP_SERVER_TOKEN"]
    async with MCPServerStreamableHttp(
        name="Streamable HTTP Python Server",
        params={
            "url": "http://localhost:8000/mcp",
            "headers": {"Authorization": f"Bearer {token}"},
            "timeout": 10,
        },
        cache_tools_list=True,
        max_retry_attempts=3,
    ) as server:
        agent = Agent(
            name="Assistant",
            instructions="Use the MCP tools to answer the questions.",
            mcp_servers=[server],
            model_settings=ModelSettings(tool_choice="required"),
        )

        result = await Runner.run(agent, "Add 7 and 22.")
        print(result.final_output)

asyncio.run(main())

构造函数接受以下附加选项：

• client_session_timeout_seconds 控制 HTTP 读取超时。
• use_structured_content 控制是否优先使用 tool_result.structured_content 而非文本输出。
• max_retry_attempts 和 retry_backoff_seconds_base 为 list_tools() 与 call_tool() 添加自动重试。
• tool_filter 让你只暴露部分工具（见工具过滤）。
• require_approval 为本地 MCP 工具启用人机协作审批策略。
• failure_error_function 自定义模型可见的 MCP 工具失败消息；将其设为 None 可改为抛出错误。
• tool_meta_resolver 在 call_tool() 前注入每次调用的 MCP _meta 负载。

本地 MCP 服务的审批策略

MCPServerStdio、MCPServerSse 和 MCPServerStreamableHttp 都接受 require_approval。

支持形式：

• 对所有工具使用 "always" 或 "never"。
• True / False（等价于 always/never）。
• 按工具配置的映射，例如 {"delete_file": "always", "read_file": "never"}。
• 分组对象： {"always": {"tool_names": [...]}, "never": {"tool_names": [...]}}。

async with MCPServerStreamableHttp(
    name="Filesystem MCP",
    params={"url": "http://localhost:8000/mcp"},
    require_approval={"always": {"tool_names": ["delete_file"]}},
) as server:
    ...

完整的暂停/恢复流程请参见 Human-in-the-loop 和 examples/mcp/get_all_mcp_tools_example/main.py。

使用 tool_meta_resolver 的每次调用元数据

当你的 MCP 服务期望在 _meta 中接收请求元数据（例如租户 ID 或追踪上下文）时，请使用 tool_meta_resolver。下例假设你将 dict 作为 context 传给 Runner.run(...)。

from agents.mcp import MCPServerStreamableHttp, MCPToolMetaContext


def resolve_meta(context: MCPToolMetaContext) -> dict[str, str] | None:
    run_context_data = context.run_context.context or {}
    tenant_id = run_context_data.get("tenant_id")
    if tenant_id is None:
        return None
    return {"tenant_id": str(tenant_id), "source": "agents-sdk"}


server = MCPServerStreamableHttp(
    name="Metadata-aware MCP",
    params={"url": "http://localhost:8000/mcp"},
    tool_meta_resolver=resolve_meta,
)

如果你的运行上下文是 Pydantic 模型、dataclass 或自定义类，请改用属性访问来读取租户 ID。

MCP 工具输出：文本与图像

当 MCP 工具返回图像内容时，SDK 会自动将其映射为图像工具输出项。混合文本/图像响应会作为输出项列表转发，因此智能体可以像消费常规工具调用的图像输出一样消费 MCP 图像结果。

3. HTTP with SSE MCP servers

Warning

MCP 项目已弃用 Server-Sent Events 传输。对于新集成请优先使用 Streamable HTTP 或 stdio，仅为遗留服务保留 SSE。

如果 MCP 服务实现了 HTTP with SSE 传输，请实例化 MCPServerSse。除传输方式外，其 API 与 Streamable HTTP 服务完全一致。

from agents import Agent, Runner
from agents.model_settings import ModelSettings
from agents.mcp import MCPServerSse

workspace_id = "demo-workspace"

async with MCPServerSse(
    name="SSE Python Server",
    params={
        "url": "http://localhost:8000/sse",
        "headers": {"X-Workspace": workspace_id},
    },
    cache_tools_list=True,
) as server:
    agent = Agent(
        name="Assistant",
        mcp_servers=[server],
        model_settings=ModelSettings(tool_choice="required"),
    )
    result = await Runner.run(agent, "What's the weather in Tokyo?")
    print(result.final_output)

4. stdio MCP servers

对于作为本地子进程运行的 MCP 服务，请使用 MCPServerStdio。SDK 会启动该 进程、保持管道打开，并在上下文管理器退出时自动关闭。该选项适合快速概念验证，或服务仅暴露命令行入口点的场景。

from pathlib import Path
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

current_dir = Path(__file__).parent
samples_dir = current_dir / "sample_files"

async with MCPServerStdio(
    name="Filesystem Server via npx",
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
) as server:
    agent = Agent(
        name="Assistant",
        instructions="Use the files in the sample directory to answer questions.",
        mcp_servers=[server],
    )
    result = await Runner.run(agent, "List the files available to you.")
    print(result.final_output)

5. MCP 服务管理器

当你有多个 MCP 服务时，请使用 MCPServerManager 提前连接它们，并将已连接的子集暴露给智能体。 构造选项和重连行为见 MCPServerManager API reference。

from agents import Agent, Runner
from agents.mcp import MCPServerManager, MCPServerStreamableHttp

servers = [
    MCPServerStreamableHttp(name="calendar", params={"url": "http://localhost:8000/mcp"}),
    MCPServerStreamableHttp(name="docs", params={"url": "http://localhost:8001/mcp"}),
]

async with MCPServerManager(servers) as manager:
    agent = Agent(
        name="Assistant",
        instructions="Use MCP tools when they help.",
        mcp_servers=manager.active_servers,
    )
    result = await Runner.run(agent, "Which MCP tools are available?")
    print(result.final_output)

关键行为：

• 当 drop_failed_servers=True（默认）时，active_servers 仅包含连接成功的服务。
• 失败会记录在 failed_servers 和 errors 中。
• 设置 strict=True 可在首次连接失败时抛出异常。
• 调用 reconnect(failed_only=True) 仅重试失败服务，或调用 reconnect(failed_only=False) 重启所有服务。
• 使用 connect_timeout_seconds、cleanup_timeout_seconds 和 connect_in_parallel 来调优生命周期行为。

通用服务能力

以下章节适用于各类 MCP 服务传输方式（具体 API 取决于服务类）。

工具过滤

每个 MCP 服务都支持工具过滤，这样你就可以只暴露智能体所需的函数。过滤可在 构造时静态进行，也可在每次运行时动态进行。

静态工具过滤

使用 create_static_tool_filter 配置简单的允许/阻止列表：

from pathlib import Path

from agents.mcp import MCPServerStdio, create_static_tool_filter

samples_dir = Path("/path/to/files")

filesystem_server = MCPServerStdio(
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
    tool_filter=create_static_tool_filter(allowed_tool_names=["read_file", "write_file"]),
)

当同时提供 allowed_tool_names 与 blocked_tool_names 时，SDK 会先应用允许列表，再从剩余集合中移除被阻止工具。

动态工具过滤

对于更复杂的逻辑，可传入一个接收 ToolFilterContext 的可调用对象。该对象可以是同步或异步的，当工具应被暴露时返回 True。

from pathlib import Path

from agents.mcp import MCPServerStdio, ToolFilterContext

samples_dir = Path("/path/to/files")

async def context_aware_filter(context: ToolFilterContext, tool) -> bool:
    if context.agent.name == "Code Reviewer" and tool.name.startswith("danger_"):
        return False
    return True

async with MCPServerStdio(
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
    tool_filter=context_aware_filter,
) as server:
    ...

过滤上下文会暴露当前 run_context、请求工具的 agent 以及 server_name。

Prompts

MCP 服务还可提供 prompts，用于动态生成智能体指令。支持 prompts 的服务会暴露两个 方法：

• list_prompts() 枚举可用的提示词模板。
• get_prompt(name, arguments) 获取具体提示词，可选传入参数。

from agents import Agent

prompt_result = await server.get_prompt(
    "generate_code_review_instructions",
    {"focus": "security vulnerabilities", "language": "python"},
)
instructions = prompt_result.messages[0].content.text

agent = Agent(
    name="Code Reviewer",
    instructions=instructions,
    mcp_servers=[server],
)

缓存

每次智能体运行都会在每个 MCP 服务上调用 list_tools()。远程服务可能引入明显延迟，因此所有 MCP 服务类都提供 cache_tools_list 选项。仅当你确信工具定义不会频繁变化时才将其设为 True。若之后要强制刷新列表，请在服务实例上调用 invalidate_tools_cache()。

追踪

追踪 会自动捕获 MCP 活动，包括：

1. 调用 MCP 服务列举工具。
2. 工具调用中的 MCP 相关信息。

延伸阅读

• Model Context Protocol – 规范与设计指南。
• examples/mcp – 可运行的 stdio、SSE 和 Streamable HTTP 示例。
• examples/hosted_mcp – 完整 Hosted MCP 演示，包括审批与连接器。