openai-agents-python MCP 集成全指南五种传输方式、审批策略与高级配置实战【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonModel Context ProtocolMCP为应用程序向语言模型暴露工具与上下文提供了一套标准化协议。本文以 openai-agents-python 官方文档docs/zh/mcp.md为骨架系统讲解如何把 MCP 服务器接入智能体从托管式 MCP 工具、Streamable HTTP、SSE、stdio 到服务器管理器覆盖每一种传输方式的配置方法、审批策略、工具筛选、分页、缓存与追踪等通用能力并深入到 server.py 与 tool.py 的源码实现帮助你在真实项目中做出正确的集成选型并快速落地。MCP 是什么为什么要在智能体中使用它MCP 是一种开放协议用于标准化应用程序向 LLM 提供上下文的方式。官方将其类比为“AI 应用程序的 USB-C 端口”正如 USB-C 提供了一种将设备连接到各种外围设备和配件的标准化方式MCP 也提供了一种将 AI 模型连接到不同数据源和工具的标准化方式。Agents Python SDK 支持多种 MCP 传输方式因此你可以复用现有的 MCP 服务器也可以自行构建服务器向智能体公开由文件系统、HTTP 或连接器支持的工具。⚠️连接前务必进行信任验证MCP 工具可以公开模型上下文中的数据并使用你提供的凭据执行操作。请仅连接你信任的服务器使用最小权限凭据将访问令牌放在授权字段或标头中而不是 URL 中并要求对敏感操作进行审批。如何选择 MCP 集成方式将 MCP 服务器接入智能体之前需要先确定应在何处执行工具调用以及你可以访问哪些传输方式。下表汇总了 Python SDK 支持的选项需求推荐选项让 OpenAI 的 Responses API 代表模型调用可公开访问的 MCP 服务器通过HostedMCPTool使用托管式 MCP 服务器工具连接到你在本地或远程运行的 Streamable HTTP 服务器通过MCPServerStreamableHttp使用Streamable HTTP MCP 服务器与实现了基于 Server-Sent Events 的 HTTP 的服务器通信通过MCPServerSse使用基于 SSE 的 HTTP MCP 服务器启动本地进程并通过 stdin/stdout 通信通过MCPServerStdio使用stdio MCP 服务器其中HostedMCPTool定义在 src/agents/tool.py三种本地 MCP 服务器类则全部实现于 src/agents/mcp/server.pyMCPServerStdio在 L1888 附近、MCPServerSse在 L2030 附近、MCPServerStreamableHttp在 L2200 附近并统一通过 src/agents/mcp/init.py 对外导出。MCP Python SDK v1 与 v2 的兼容性Agents SDK 通过依赖版本范围mcp1.19.0,3同时支持mcpPython 软件包的两个主要版本。需要特别注意的是已安装的mcp软件包版本与同服务器协商的 MCP 协议版本相互独立。SDK 会检测已安装软件包的主版本并自动适配 stdio、SSE 和 Streamable HTTP 连接因此普通服务器配置不需要提供版本切换选项。安装 MCP Python SDK v2 后Agents SDK 会围绕配置的本地传输方式创建带有modeauto的 v2mcp.Client。客户端首先使用已安装 MCP SDK 所支持的最新协议版本发送server/discover探测请求现代服务器会响应此探测请求客户端随后采用响应结果较旧的服务器不支持server/discover时客户端会回退到旧版initialize握手并使用在该过程中协商的协议版本。因此安装 MCP Python SDK v2并不会强制所有连接都使用最新的 MCP 协议版本。大多数应用程序应让依赖解析器选择兼容版本如果你的应用程序必须固定使用某个主版本请在openai-agents旁添加显式约束# MCP Python SDK v1 pip install mcp1.19.0,2 # MCP Python SDK v2 pip install mcp2,3HTTP 传输自定义与包版本强相关HTTP 传输自定义必须使用已安装 MCP 软件包所拥有的 HTTP 栈二者不能混用自定义项MCP Python SDK v1MCP Python SDK v2params[auth]httpx.Authhttpx2.Authparams[httpx_client_factory]返回值httpx.AsyncClienthttpx2.AsyncClientMCPServerStreamableHttp的params[ignore_initialized_notification_failure] True支持不支持连接前会被拒绝应尽可能使用Authorization标头——它在两个软件包版本中均可保持不变。当应用程序提供params[auth]或params[httpx_client_factory]时这些值必须使用已安装mcp包主版本对应的 HTTP 类型。当应用程序设置MCPServerStreamableHttp的params[ignore_initialized_notification_failure] True时必须保留mcp2或在升级前禁用该选项。从源码可以看到这一约束是硬校验而非软提示server.py 中create_streams()检测到 MCP v2 且该参数为True时会直接抛出UserErrornot supported with MCP Python SDK v2。同时 MCPServerStreamableHttpParams 对auth、httpx_client_factory的注释也分别标注了 v1/v2 对应的 HTTP 类型。这些本地mcp依赖要求不适用于HostedMCPTool因为远程 MCP 连接由 OpenAI Responses API 管理本地无需安装或协调mcp包。智能体级 MCP 配置Agent.mcp_config除了选择传输方式外还可以通过设置Agent.mcp_config调整 MCP 工具的准备方式from agents import Agent agent Agent( nameAssistant, 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采用尽力而为的方式。如果某个架构无法转换则使用原始架构。failure_error_function控制如何向模型呈现 MCP 工具调用失败。未设置时SDK 使用默认的工具错误格式化程序。服务器级的failure_error_function会覆盖该服务器的Agent.mcp_config[failure_error_function]。这一优先级在MCPServerStreamableHttp的构造函数注释中有明确说明server.py显式设为None时抛出错误而非转换完全不设置时才回退到智能体级配置或 SDK 默认。include_server_in_tool_names需要主动启用。启用后每个本地 MCP 工具都会使用确定性的服务器前缀名称向模型公开有助于避免多个 MCP 服务器发布同名工具时发生冲突。生成的名称兼容 ASCII、不会超过FunctionTool实例的名称长度限制也不会与同一智能体上本地FunctionTool实例的已配置名称或已启用的任务转移发生冲突。SDK 仍会在原始服务器上调用具有原始名称的 MCP 工具。各传输方式的通用决策模式选择传输方式后大多数集成还需要作出相同的后续决策如何仅公开一部分工具工具筛选。服务器是否还提供可复用的提示词提示词。是否应缓存list_tools()缓存。MCP 活动如何显示在追踪中追踪。对于本地 MCP 服务器MCPServerStdio、MCPServerSse、MCPServerStreamableHttp审批策略和每次调用的_meta载荷也是通用概念。下文 Streamable HTTP 一节给出了最完整的代码示例同样的模式也适用于其他本地传输方式。1. 托管式 MCP 服务器工具Hosted MCP Server Tools托管工具会将整个工具调用往返流程交由 OpenAI 基础设施处理。你的代码无需列出和调用工具HostedMCPTool会将服务器标签以及可选的连接器元数据转发给 Responses API。模型会列出远程服务器的工具并调用它们而无需额外回调你的 Python 进程。目前托管工具适用于支持 Responses API 托管式 MCP 集成的 OpenAI 模型。从源码看HostedMCPTool是一个简单的 dataclasstool.py核心字段只有tool_config类型为Mcp与发送给 REST API 的 JSON 相对应和可选的on_approval_request回调其name属性固定返回hosted_mcp——也就是说工具调用本身完全由模型端驱动。基础托管式 MCP 工具将HostedMCPTool添加到智能体的tools列表即可创建托管工具。tool_config字典与发送给 REST API 的 JSON 相对应import asyncio from agents import Agent, HostedMCPTool, Runner async def main() - None: agent Agent( nameAssistant, instructionsUse 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添加到智能体。仅 OpenAI Responses 模型支持此功能完整的工具搜索设置和限制可参阅 docs/zh/tools.md。托管式 MCP 结果的流式传输托管工具支持流式传输结果其方式与函数工具完全相同。使用Runner.run_streamed可在模型仍在工作时接收增量 MCP 输出result Runner.run_streamed(agent, Summarise this repositorys top languages) async for event in result.stream_events(): if event.type run_item_stream_event: print(fReceived: {event.item}) print(result.final_output)可选审批流程如果服务器能够执行敏感操作可以要求在每次执行工具前进行人工或程序化审批。在tool_config中配置require_approval其值可以是单一策略always、never也可以是将工具名称映射到策略的字典。若要在 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( nameAssistant, tools[ HostedMCPTool( tool_config{ type: mcp, server_label: deepwiki, server_url: https://mcp.deepwiki.com/mcp, require_approval: always, }, on_approval_requestapprove_tool, ) ], )回调可以是同步或异步的并且每当模型需要审批数据才能继续运行时都会调用它。对应类型定义在 tool.pyMCPToolApprovalRequest携带ctx_wrapper运行上下文与data审批请求数据MCPToolApprovalFunctionResult则是一个 TypedDict必填approve: bool可选用reason说明拒绝原因。由连接器支持的托管服务器托管式 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. Streamable HTTP MCP 服务器如果希望自行管理网络连接请使用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( nameStreamable HTTP Python Server, params{ url: http://localhost:8000/mcp, headers: {Authorization: fBearer {token}}, timeout: 10, }, cache_tools_listTrue, max_retry_attempts3, ) as server: agent Agent( nameAssistant, instructionsUse the MCP tools to answer the questions., mcp_servers[server], model_settingsModelSettings(tool_choicerequired), ) result await Runner.run(agent, Add 7 and 22.) print(result.final_output) asyncio.run(main())构造函数完整参数说明对照 MCPServerStreamableHttp.init的签名构造函数除params外还接受以下选项client_session_timeout_seconds控制 MCP ClientSession 的读取超时默认 5 秒。可由datetime.timedelta表示且至少为一微秒的有限正值会设置有限超时None和0会禁用超时。构造服务器时会拒绝其他值。use_structured_content控制是否优先使用tool_result.structured_content而不是文本输出。默认False向后兼容——大多数 MCP 服务器仍会把结构化内容包含在tool_result.content中默认启用会导致重复内容只有当你确认服务器不会在tool_result.content中重复结构化内容时才设为True。max_retry_attempts与retry_backoff_seconds_base为list_tools()和call_tool()添加自动重试指数退避默认不重试retry_backoff_seconds_base默认 1.0 秒另有可选的retry_backoff_seconds_max封顶退避延迟。tool_filter允许你仅公开一部分工具见下文工具筛选。require_approval为本地 MCP 工具启用人机协同审批策略。failure_error_function自定义模型可见的 MCP 工具失败消息设置为None可改为抛出错误。tool_meta_resolver会在call_tool()之前注入每次调用的 MCP_meta载荷。message_handler可选处理 ClientSession 交付的会话消息。tool_input_guardrails/tool_output_guardrails可选对服务器上的每个工具在调用前/返回后应用守卫guardrail。params字典本身MCPServerStreamableHttpParams支持url必填、headers、timeoutHTTP 请求超时默认 5 秒、sse_read_timeoutSSE 连接超时默认 5 分钟、terminate_on_close默认True、httpx_client_factory、auth、ignore_initialized_notification_failure。本地 MCP 服务器的审批策略MCPServerStdio、MCPServerSse和MCPServerStreamableHttp均接受require_approval支持以下形式类型定义见 server.py 中的RequireApprovalPolicy、RequireApprovalMapping、RequireApprovalObject与RequireApprovalSetting对所有工具使用always或never。True要求审批所有工具False不要求审批任何工具分别等同于always和never。按工具配置的映射例如{delete_file: always, read_file: never}。分组对象{always: {tool_names: [...]}, never: {tool_names: [...]}}。此外还支持传入一个可调用对象LocalMCPApprovalCallable接收RunContextWrapper、请求工具的AgentBase和MCPTool返回是否放行的布尔值。async with MCPServerStreamableHttp( nameFilesystem MCP, params{url: http://localhost:8000/mcp}, require_approval{always: {tool_names: [delete_file]}}, ) as server: ...有关完整的暂停/恢复流程请参阅 docs/zh/human_in_the_loop.md 和 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( nameMetadata-aware MCP, params{url: http://localhost:8000/mcp}, tool_meta_resolverresolve_meta, )如果运行上下文是 Pydantic 模型、dataclass 或自定义类请改用属性访问方式读取租户 ID而不是字典式get。MCP 工具输出文本、图像及其他内容当 MCP 结果使用内容块时SDK 的处理规则如下文本内容作为文本输出转发图像内容映射为工具输出中的图像类型条目其他 MCP 内容块类型包括音频和资源块SDK 转发文本输出其值为该内容块的有效 JSON 序列化结果包含多个内容块的响应会作为输出项列表转发如果use_structured_contentTrue选择了非空且无错误的structuredContent载荷则该结构化载荷优先于这些内容块结构化内容缺失或为空时回退到内容块。3. 基于 SSE 的 HTTP MCP 服务器⚠️注意MCP 项目已弃用Server-Sent Events 传输方式。对于新集成请优先使用 Streamable HTTP 或 stdio仅为旧版服务器保留 SSE。如果 MCP 服务器实现了基于 SSE 的 HTTP 传输方式请实例化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( nameSSE Python Server, params{ url: http://localhost:8000/sse, headers: {X-Workspace: workspace_id}, }, cache_tools_listTrue, ) as server: agent Agent( nameAssistant, mcp_servers[server], model_settingsModelSettings(tool_choicerequired), ) result await Runner.run(agent, Whats 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( nameFilesystem Server via npx, params{ command: npx, args: [-y, modelcontextprotocol/server-filesystem, str(samples_dir)], }, ) as server: agent Agent( nameAssistant, instructionsUse 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)一个真实可运行的完整版本位于 examples/mcp/filesystem_example/main.py它在__main__中先通过shutil.which(npx)检查npx是否安装再用trace()包裹一次多轮对话列出文件、读取书目、推荐歌曲并在上下文中把 stdio 服务器挂载到同一个智能体上反复复用。5. MCP 服务器管理器MCPServerManager如果有多个 MCP 服务器请使用MCPServerManager预先连接它们并向智能体公开其中成功连接的服务器子集from agents import Agent, Runner from agents.mcp import MCPServerManager, MCPServerStreamableHttp servers [ MCPServerStreamableHttp(namecalendar, params{url: http://localhost:8000/mcp}), MCPServerStreamableHttp(namedocs, params{url: http://localhost:8001/mcp}), ] async with MCPServerManager(servers) as manager: agent Agent( nameAssistant, instructionsUse MCP tools when they help., mcp_serversmanager.active_servers, ) result await Runner.run(agent, Which MCP tools are available?) print(result.final_output)主要行为当drop_failed_serversTrue默认值时active_servers仅包含成功连接的服务器。失败信息记录在failed_servers和errors中。设置strictTrue可在首次连接失败时抛出异常。调用reconnect(failed_onlyTrue)可重试失败的服务器调用reconnect(failed_onlyFalse)可重启所有服务器。对connect_all()、reconnect()和cleanup_all()的调用会串行执行。如果某个生命周期操作已在运行另一个生命周期操作会等待其完成而不会并发连接或清理相同的服务器。设置connect_timeout_seconds、cleanup_timeout_seconds和connect_in_parallel可调整生命周期行为。两个生命周期超时的默认值均为10 秒。它们接受有限正秒数或使用None禁用在构造和赋值时都会进行验证零会被拒绝因为它会产生立即到期的截止时间。构造函数选项与重连行为的完整 API 参考见 docs/ref/mcp/manager.md可运行示例见 examples/mcp/manager_example含app.py、mcp_server.py与smoke_test.py。通用服务器能力以下各节适用于所有 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_filtercreate_static_tool_filter(allowed_tool_names[read_file, write_file]), )同时提供allowed_tool_names和blocked_tool_names时SDK 会先应用允许列表然后从剩余集合中移除所有被阻止的工具。动态工具筛选对于更复杂的逻辑请传入一个可调用对象该对象接收ToolFilterContext。该可调用对象可以是同步或异步的并在应公开工具时返回Truefrom 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_filtercontext_aware_filter, ) as server: ...筛选器上下文会公开活动的run_context、请求工具的agent以及server_name。可运行示例见 examples/mcp/tool_filter_example/main.py。提示词MCP 服务器还可以提供动态生成智能体指令的提示词。支持提示词的服务器会公开两种方法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( nameCode Reviewer, instructionsinstructions, mcp_servers[server], )服务端与客户端两端完整的提示词示例见 examples/mcp/prompt_serverserver.py提供提示词main.py消费提示词构造智能体。分页内置的本地 MCP 服务器类在列出工具和提示词时会自动跟随nextCursorlist_tools()会先收集完整的工具列表再应用筛选器或填充缓存list_prompts()返回一个合并结果其中包含nextCursorNone如果后续页面失败或服务器重复使用游标该操作会抛出错误而不会公开或缓存部分结果。资源仍需显式分页将list_resources()或list_resource_templates()返回的nextCursor作为cursor参数传回以获取下一页。缓存每次智能体运行都会在每个 MCP 服务器上调用list_tools()。远程服务器可能带来明显的延迟因此所有 MCP 服务器类都公开了cache_tools_list选项。仅当你确信工具定义不会频繁变化时才应将其设置为True。如需稍后强制获取最新列表请在服务器实例上调用invalidate_tools_cache()。从源码看server.py启用缓存后工具列表只会从服务器获取一次能大幅降低延迟避免每次运行都做一次往返反之每次list_tools()都会实时拉取。追踪追踪会自动捕获 MCP 活动包括为列出工具而对 MCP 服务器发起的调用。工具调用中的 MCP 相关信息。上图展示了 MCP 追踪的实际效果你可以看到对 MCP 服务器的list_tools调用以及工具调用上携带的 MCP 相关元数据便于在分布式运行中定位工具列表延迟或调用失败的具体环节。延伸阅读examples/mcp —— 可运行的 stdio、SSE 和 Streamable HTTP 代码示例含 filesystem、git、SSE、Streamable HTTP 远程/自定义客户端、提示词、工具筛选与管理器等完整场景。examples/hosted_mcp —— 完整的托管式 MCP 演示包括审批on_approval.py和连接器connectors.py。docs/zh/human_in_the_loop.md —— 人机协同审批的完整暂停/恢复流程。docs/ref/mcp —— MCP 相关 API 参考server、manager、util 等。结语按场景选择集成路线综合以上内容实际项目中的选型建议可归纳为服务器可公开访问且希望零本地依赖时选托管式HostedMCPTool自建基础设施、追求低延迟与传输可控时选 Streamable HTTP旧版服务器只支持 SSE 时用MCPServerSse新项目避免本地子进程或命令行入口点用MCPServerStdio多服务器接入时用MCPServerManager统一生命周期。无论走哪条路线工具筛选、审批策略、缓存与追踪这四个通用能力都值得在架构设计阶段一并规划这样既能保证安全边界也能让 MCP 集成在长期运行中保持可观测、可维护。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
