跳转至

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 服务 托管的 MCP 服务工具,通过 HostedMCPTool
连接你在本地或远程运行的可流式传输的 HTTP 服务 可流式传输的 HTTP MCP 服务,通过 MCPServerStreamableHttp
与实现了带 Server-Sent Events 的 HTTP 的服务通信 带 SSE 的 HTTP MCP 服务,通过 MCPServerSse
启动本地进程并通过 stdin/stdout 通信 stdio MCP 服务,通过 MCPServerStdio

下文将逐一介绍每个选项、其配置方法,以及在何种情况下优先选择某种传输方式。

1. 托管的 MCP 服务工具

托管工具将完整的工具往返流程放在 OpenAI 的基础设施中。你的代码无需列出和调用工具,HostedMCPTool 会将服务标签(以及可选的连接器元数据)转发给 Responses API。模型会列出远程服务的工具并直接调用它们,无需回调你的 Python 进程。托管工具目前适用于支持 Responses API 托管 MCP 集成的 OpenAI 模型。

基础的托管 MCP 工具

在智能体的 tools 列表中添加一个 HostedMCPTool 即可创建托管工具。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())

托管服务会自动暴露其工具;你无需将其添加到 mcp_servers

托管 MCP 结果的流式传输

托管工具以与工具调用完全相同的方式支持流式传输。向 Runner.run_streamed 传递 stream=True,即可在模型仍在工作时消费增量 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,
        )
    ],
)

该回调可以是同步或异步的,当模型需要审批数据以继续运行时会被调用。

由连接器支持的托管服务

托管 MCP 也支持 OpenAI 连接器。无需指定 server_url,而是提供 connector_id 和访问令牌。Responses API 负责身份验证,托管服务会暴露连接器的工具。

import os

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

完整可运行的托管工具示例(包括流式传输、审批和连接器)位于 examples/hosted_mcp

2. 可流式传输的 HTTP MCP 服务

当你希望自行管理网络连接时,请使用 MCPServerStreamableHttp。当你可控传输层,或希望在自有基础设施中运行服务并保持较低延迟时,可流式传输的 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_attemptsretry_backoff_seconds_baselist_tools()call_tool() 增加自动重试。
  • tool_filter 允许你只暴露工具的子集(参见工具过滤)。

3. 带 SSE 的 HTTP MCP 服务

如果 MCP 服务实现了带 SSE 的 HTTP 传输,请实例化 MCPServerSse。除传输方式不同外,其 API 与可流式传输的 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 服务

对于作为本地子进程运行的 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)

工具过滤

每个 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_namesblocked_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

提示词

MCP 服务还可以提供动态生成智能体 instructions 的提示词。支持提示词的服务会暴露两个方法:

  • 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 相关信息。

MCP Tracing Screenshot

延伸阅读