跳过内容

模型上下文协议 (MCP)

模型上下文协议 (MCP) 标准化了应用程序向语言模型暴露工具和上下文的方式。来自官方文档

MCP 是一种开放协议,用于标准化应用程序向 LLM 提供上下文的方式。可以将 MCP 视为 AI 应用程序的 USB-C 端口。就像 USB-C 为连接设备到各种外围设备和配件提供了一种标准化的方式一样,MCP 为连接 AI 模型到不同的数据源和工具提供了一种标准化的方式。

Agents Python SDK 理解多种 MCP 传输方式。这使您可以重用现有的 MCP 服务器或构建自己的服务器,以将文件系统、HTTP 或 Connector 支持的工具暴露给 Agent。

选择 MCP 集成

在将 MCP 服务器连接到 Agent 之前,请确定工具调用应该在何处执行以及您可以访问哪些传输方式。下表总结了 Python SDK 支持的选项。

您需要的东西 推荐选项
让 OpenAI 的 Responses API 代表模型调用可公开访问的 MCP 服务器 托管 MCP 服务器工具 通过 HostedMCPTool
连接到您本地或远程运行的可流式传输 HTTP 服务器 可流式传输 HTTP MCP 服务器 通过 MCPServerStreamableHttp
与实现 HTTP 和服务器发送事件 (SSE) 的服务器通信 使用 SSE 的 HTTP MCP 服务器 通过 MCPServerSse
启动一个本地进程并通过 stdin/stdout 进行通信 stdio MCP 服务器 通过 MCPServerStdio

以下部分将介绍每个选项,如何配置它以及何时优先选择一种传输方式而不是另一种传输方式。

1. 托管 MCP 服务器工具

托管工具将整个工具往返过程推送到 OpenAI 的基础设施中。与其让您的代码列出和调用工具,不如 HostedMCPTool 将服务器标签(以及可选的 Connector 元数据)转发到 Responses API。模型会列出远程服务器的工具并调用它们,而无需回调到您的 Python 进程。托管工具当前适用于支持 Responses API 的托管 MCP 集成的 OpenAI 模型。

基本的托管 MCP 工具

通过将 HostedMCPTool 添加到 Agent 的 tools 列表中来创建托管工具。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 结果

托管工具以与函数工具完全相同的方式支持流式传输结果。将 stream=True 传递给 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,
        )
    ],
)

回调函数可以是同步的或异步的,并且会在模型需要审批数据才能继续运行时被调用。

Connector 支持的托管服务器

托管 MCP 还支持 OpenAI Connector。与其指定 server_url,不如提供 connector_id 和访问令牌。Responses API 会处理身份验证,并且托管服务器会暴露 Connector 的工具。

import os

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

包含流式传输、审批和 Connector 的完整工作托管工具示例位于 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": "https://: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 服务器实现了 HTTP 和 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": "https://: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 服务器都支持工具过滤器,以便您可以仅暴露 Agent 所需的函数。过滤可以在构造时或每次运行时动态发生。

静态工具过滤

使用 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

提示词 (Prompts)

MCP 服务器还可以提供动态生成 Agent 指令的提示词。支持提示词的服务器暴露两种方法

  • 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],
)

缓存

每次 Agent 运行都会调用每个 MCP 服务器上的 list_tools()。远程服务器可能会引入明显的延迟,因此所有 MCP 服务器类都暴露了一个 cache_tools_list 选项。仅当您确信工具定义不会经常更改时才将其设置为 True。要强制稍后刷新列表,请在服务器实例上调用 invalidate_tools_cache()

追踪

Tracing 自动捕获 MCP 活动,包括

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

MCP Tracing Screenshot

进一步阅读