Movatterモバイル変換


[0]ホーム

URL:


跳转至

Model context protocol (MCP)

Model context protocol(MCP)标准化了应用如何向语言模型暴露工具与上下文。官方文档中写道:

MCP is an open protocol that standardizes how applications provide context to LLMs. Think of MCP like a USB-C port for AIapplications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCPprovides a standardized way to connect AI models to different data sources and tools.

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

选择 MCP 集成方式

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

你的需求推荐选项
让 OpenAI 的 Responses API 代表模型调用可公开访问的 MCP 服务托管 MCP 服务工具,通过HostedMCPTool
连接你在本地或远程运行的 Streamable HTTP 服务Streamable HTTP MCP 服务,通过MCPServerStreamableHttp
与实现了带 Server-Sent Events 的 HTTP 的服务通信HTTP with SSE 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 镜像一致:

importasynciofromagentsimportAgent,HostedMCPTool,Runnerasyncdefmain()->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=awaitRunner.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")asyncforeventinresult.stream_events():ifevent.type=="run_item_stream_event":print(f"Received:{event.item}")print(result.final_output)

可选审批流

如果服务可执行敏感操作,你可以在每次工具执行前要求人工或程序化审批。通过在tool_config 中配置require_approval,可设置单一策略("always""never")或将工具名映射到策略的字典。若需在 Python 内做决定,可提供on_approval_request 回调。

fromagentsimportMCPToolApprovalFunctionResult,MCPToolApprovalRequestSAFE_TOOLS={"read_project_metadata"}defapprove_tool(request:MCPToolApprovalRequest)->MCPToolApprovalFunctionResult:ifrequest.data.nameinSAFE_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 会处理鉴权,托管服务则暴露该连接器的工具。

importosHostedMCPTool(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. Streamable HTTP MCP 服务

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

importasyncioimportosfromagentsimportAgent,Runnerfromagents.mcpimportMCPServerStreamableHttpfromagents.model_settingsimportModelSettingsasyncdefmain()->None:token=os.environ["MCP_SERVER_TOKEN"]asyncwithMCPServerStreamableHttp(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,)asserver:agent=Agent(name="Assistant",instructions="Use the MCP tools to answer the questions.",mcp_servers=[server],model_settings=ModelSettings(tool_choice="required"),)result=awaitRunner.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. HTTP with SSE MCP 服务

如果 MCP 服务实现了带 SSE 的 HTTP 传输,请实例化MCPServerSse。除传输方式不同外,其 API 与 Streamable HTTP 服务完全相同。

fromagentsimportAgent,Runnerfromagents.model_settingsimportModelSettingsfromagents.mcpimportMCPServerSseworkspace_id="demo-workspace"asyncwithMCPServerSse(name="SSE Python Server",params={"url":"http://localhost:8000/sse","headers":{"X-Workspace":workspace_id},},cache_tools_list=True,)asserver:agent=Agent(name="Assistant",mcp_servers=[server],model_settings=ModelSettings(tool_choice="required"),)result=awaitRunner.run(agent,"What's the weather in Tokyo?")print(result.final_output)

4. stdio MCP 服务

对于作为本地子进程运行的 MCP 服务,使用MCPServerStdio。SDK 会拉起进程、保持管道打开,并在上下文管理器退出时自动关闭。该选项适合快速原型验证,或当服务仅以命令行入口暴露时。

frompathlibimportPathfromagentsimportAgent,Runnerfromagents.mcpimportMCPServerStdiocurrent_dir=Path(__file__).parentsamples_dir=current_dir/"sample_files"asyncwithMCPServerStdio(name="Filesystem Server via npx",params={"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem",str(samples_dir)],},)asserver:agent=Agent(name="Assistant",instructions="Use the files in the sample directory to answer questions.",mcp_servers=[server],)result=awaitRunner.run(agent,"List the files available to you.")print(result.final_output)

工具过滤

每个 MCP 服务都支持工具过滤,以便你仅暴露智能体所需的函数。过滤可在构造时设置,或在每次运行时动态设置。

静态工具过滤

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

frompathlibimportPathfromagents.mcpimportMCPServerStdio,create_static_tool_filtersamples_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 表示应暴露该工具。

frompathlibimportPathfromagents.mcpimportMCPServerStdio,ToolFilterContextsamples_dir=Path("/path/to/files")asyncdefcontext_aware_filter(context:ToolFilterContext,tool)->bool:ifcontext.agent.name=="Code Reviewer"andtool.name.startswith("danger_"):returnFalsereturnTrueasyncwithMCPServerStdio(params={"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem",str(samples_dir)],},tool_filter=context_aware_filter,)asserver:...

过滤上下文会提供活动的run_context、请求工具的agent,以及server_name

提示词

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

  • list_prompts() 枚举可用的提示模板。
  • get_prompt(name, arguments) 获取具体提示词,可选携带参数。
fromagentsimportAgentprompt_result=awaitserver.get_prompt("generate_code_review_instructions",{"focus":"security vulnerabilities","language":"python"},)instructions=prompt_result.messages[0].content.textagent=Agent(name="Code Reviewer",instructions=instructions,mcp_servers=[server],)

缓存

每次智能体运行都会在每个 MCP 服务上调用list_tools()。远程服务可能带来显著延迟,因此所有 MCP 服务类都暴露了cache_tools_list 选项。仅在你确信工具定义不频繁变化时将其设为True。如需之后强制刷新列表,可在服务实例上调用invalidate_tools_cache()

追踪

Tracing 会自动捕获 MCP 活动,包括:

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

MCP Tracing Screenshot

延伸阅读


[8]ページ先頭

©2009-2025 Movatter.jp