9.0
重磅
AI SCORE
模型发布2025-03-27 02:55
OpenAI Agents SDK正式支持MCP协议
Hacker News · openai.github.io#OpenAI#MCP#Agent
Editor brief · 编辑速览
OpenAI官方在Agents SDK中集成Model Context Protocol，为开发者提供更便利、标准化的Agent能力扩展方案。
模型上下文协议（MCP）标准化了应用程序向语言模型公开工具和上下文的方式。从官方文档中：
MCP 是一个开放协议，标准化了应用程序如何为 LLM 提供上下文。可以把 MCP 看作 AI 应用的 USB-C 端口。就像 USB-C 为连接设备和各种外设提供了标准化方式，MCP 为连接 AI 模型和不同的数据源和工具提供了标准化方式。
Agents Python SDK 理解多种 MCP 传输方式。这让你能够重用现有的 MCP 服务器或构建自己的服务器，以向 agent 公开基于文件系统、HTTP 或连接器的工具。
在将 MCP 服务器接入 agent 之前，需要决定工具调用应该在哪里执行，以及你能够访问哪些传输方式。下面的矩阵总结了 Python SDK 支持的选项。
以下部分逐一介绍每个选项、如何配置它，以及何时优先使用某个传输方式。
除了选择传输方式，你还可以通过设置 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,
# Prefix local MCP tool names with their server name.
"include_server_in_tool_names": True,
},
)
convert_schemas_to_strict 是尽力而为的。如果无法转换某个 schema，则使用原始 schema。
failure_error_function 控制 MCP 工具调用失败如何呈现给 model。
当 failure_error_function 未设置时，SDK 使用默认的工具错误格式化程序。
服务器级别的 failure_error_function 会覆盖该服务器的 Agent.mcp_config["failure_error_function"]。
include_server_in_tool_names 是可选的。启用时，每个本地 MCP 工具都以确定的服务器前缀名称公开给 model，这有助于在多个 MCP 服务器发布同名工具时避免冲突。生成的名称是 ASCII 安全的，保持在函数工具名称长度限制内，并避免同一 agent 上的现有本地函数工具和启用的切换名称冲突。SDK 仍在原始服务器上调用原始 MCP 工具名称。
选择传输方式后，大多数集成需要进行相同的后续决策：
如何仅公开工具的子集（工具过滤）。
服务器是否也提供可重用的 prompt（Prompt）。
list_tools() 是否应该被缓存（缓存）。
MCP 活动如何在追踪中显示（追踪）。
对于本地 MCP 服务器（MCPServerStdio、MCPServerSse、MCPServerStreamableHttp），审批策略和每次调用的 _meta 有效载荷也是共享概念。可流式 HTTP 部分展示了最完整的示例，相同的模式也适用于其他本地传输方式。
托管工具将整个工具往返推送到 OpenAI 的基础设施中。不是由你的代码列出和调用工具，而是 HostedMCPTool 将服务器标签（和可选的连接器元数据）转发给 Responses API。model 列出远程服务器的工具并调用它们，无需额外回调到你的 Python 进程。托管工具目前适用于支持 Responses API 的托管 MCP 集成的 OpenAI model。
通过将 HostedMCPTool 添加到 agent 的工具列表来创建托管工具。tool_config 字典镜像你发送给 REST API 的 JSON：
import asyncio
from agents import Agent, HostedMCPTool, Runner
async def main() -> None:
agent = Agent(
name="Assistant",
instructions="Use the DeepWiki hosted MCP server to inspect openai/openai-agents-python.",
tools=[
HostedMCPTool(
tool_config={
"type": "mcp",
"server_label": "deepwiki",
"server_url": "https://mcp.deepwiki.com/mcp",
"require_approval": "never",
}
)
],
)
result = await Runner.run(
agent,
"Which language is the repository openai/openai-agents-python written in?",
)
print(result.final_output)
asyncio.run(main())
托管服务器自动公开其工具；不需要将其添加到 mcp_servers。
如果你想让托管工具搜索延迟加载托管 MCP 服务器，设置 tool_config["defer_loading"] = True 并将 ToolSearchTool 添加到 agent。这仅在 OpenAI Responses model 上支持。有关完整的工具搜索设置和约束，请参见工具。
托管工具以与函数工具完全相同的方式支持流式传输结果。使用 Runner.run_streamed 在 model 仍在工作时消费增量 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)
如果服务器可以执行敏感操作，你可以在每次工具执行之前要求人工或程序化审批。使用单个策略（"always"、"never"）或将工具名称映射到策略的字典来配置 tool_config 中的 require_approval。要在 Python 中做出决定，提供 on_approval_request 回调。
from agents import MCPToolApprovalFunctionResult, MCPToolApprovalRequest
SAFE_TOOLS = {"read_wiki_structure", "read_wiki_contents", "ask_question"}
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": "deepwiki",
"server_url": "https://mcp.deepwiki.com/mcp",
"require_approval": "always",
},
on_approval_request=approve_tool,
)
],
)
回调可以是同步或异步的，在 model 需要审批数据以继续运行时被调用。
托管 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。
当你想自己管理网络连接时，使用 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_attempts 和 retry_backoff_seconds_base 为 list_tools() 和 call_tool() 添加自动重试。
tool_filter 让你公开工具的子集（见工具过滤）。
require_approval 对本地 MCP 工具启用人工在环的审批策略。
failure_error_function 自定义 model 可见的 MCP 工具失败消息；设置为 None 来抛出错误。
tool_meta_resolver 在 call_tool() 之前注入每次调用的 MCP _meta 有效载荷。
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:
...
有关完整的暂停/恢复流程，请参见人工在环和 examples/mcp/get_all_mcp_tools_example/main.py。
当你的 MCP 服务器在 _meta 中需要请求元数据（例如租户 ID 或追踪上下文）时，使用 tool_meta_resolver。下面的示例假设你将字典作为上下文传递给 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 工具返回图像内容时，SDK 自动将其映射到图像工具输出条目。混合文本/图像响应作为输出项列表转发，以便 agent 可以以与消费常规函数工具的图像输出相同的方式消费 MCP 图像结果。
MCP 项目已废弃服务器发送事件传输。优先对新集成使用可流式 HTTP 或 stdio，仅对遗留服务器保留 SSE。
如果 MCP 服务器实现了 HTTP 与 SSE 传输，实例化 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)
对于作为本地子进程运行的 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 服务器时，使用 MCPServerManager 提前连接它们并将已连接的子集公开给你的 agent。有关构造函数选项和重新连接行为，请参见 MCPServerManager API 参考。
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 服务器都支持工具过滤，以便你可以仅公开 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_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。
MCP 服务器也可以提供能够动态生成 agent 指令的 prompt。支持 prompt 的服务器公开两个方法：
list_prompts() 枚举可用的 prompt 模板。
get_prompt(name, arguments) 获取具体 prompt，可选带参数。
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()。
追踪自动捕获 MCP 活动，包括：
调用 MCP 服务器列出工具。
工具调用上的 MCP 相关信息。