企业官网建设流程全解析

openai-agents-python MCP 集成完全指南：四类传输、托管工具与服务器生命周期管理

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

本篇指南基于 openai-agents-python 官方文档 docs/ko/mcp.md 及对应源码编写，系统讲解如何在 Agents SDK 中接入 Model Context Protocol（MCP）服务器。读完本文，你将掌握四类传输方式（Hosted、Streamable HTTP、SSE、stdio）的选型与配置、Agent 级 MCP 配置参数、mcp 包 v1/v2 兼容性差异，以及工具过滤、提示词、分页、缓存、审批策略与MCPServerManager生命周期管理的完整实战方案。

什么是 MCP 以及安全前提

Model Context Protocol（MCP）是标准化"应用程序向语言模型暴露工具与上下文"的开放协议。官方文档用了一个类比：MCP 之于 AI 应用，就像 USB-C 之于设备——它为 AI 模型连接各类数据源和工具提供了统一接口。

Agents Python SDK 支持多种 MCP 传输方式，因此你可以复用现有 MCP 服务器，或自建服务器，把文件系统、HTTP、连接器（connector）等工具暴露给 Agent。

警告（连接前确认 MCP 服务器可信）：MCP 工具会读取模型上下文中的敏感数据，并使用你提供的凭据执行操作。请只连接可信服务器、使用最小权限凭据、把访问令牌放在认证字段或请求头而非 URL 中、对敏感操作要求审批。可参考 OpenAI 的 MCP 安全指南。

如何选择 MCP 集成方式

在把 MCP 服务器接入 Agent 之前，先决定两点：工具调用在哪里执行、需要访问哪种传输方式。下表总结了 Python SDK 支持的选项：

选择传输方式后，大多数集成还需要决定以下公共问题：只暴露部分工具怎么做（工具过滤）、服务器是否也提供可复用提示词（Prompts）、是否缓存list_tools()结果、以及 MCP 活动在 Trace 中如何呈现（Tracing）。对于本地 MCP 服务器（MCPServerStdio、MCPServerSse、MCPServerStreamableHttp），审批策略和每次调用的_meta负载同样是跨传输的公共概念。

mcp Python SDK v1 与 v2 的兼容处理

Agents SDK 通过mcp>=1.19.0,<3的依赖范围（见 pyproject.toml，且要求 Python 3.10+）同时支持mcpPython 包的两个主版本。注意：已安装的mcp包版本与服务器协商的 MCP 协议版本是两回事。Agents SDK 会检测已安装包的主版本并自动调整 stdio、SSE、Streamable HTTP 的连接逻辑，因此常规服务器配置不需要做版本切换设置。

当安装了 MCP Python SDK v2 时，Agents SDK 会对已配置的本地传输使用mode="auto"创建 v2mcp.Client。客户端先以已安装 MCP SDK 支持的最新协议版本发送server/discover探测；新服务器会响应探测，客户端采用其结果；旧服务器不支持server/discover时，客户端回退到传统的initialize握手并使用其中协商的协议版本。也就是说，安装 v2 并不会强制所有连接使用最新 MCP 协议版本。

多数应用应让依赖解析器自行选择兼容版本；如果应用必须锁定某个主版本，请与openai-agents一起添加显式约束：

# MCP Python SDK v1 pip install "mcp>=1.19.0,<2" # MCP Python SDK v2 pip install "mcp>=2,<3"

HTTP 传输的自定义必须使用已安装 MCP 包所"拥有"的 HTTP 栈：

源码层面可以印证这一策略：src/agents/mcp/server.py 中，_validate_v2_http_auth会在 v2 环境下校验auth必须是httpx2.Auth实例，_validated_v2_http_client_factory会校验工厂返回值必须是httpx2.AsyncClient；而MCPServerStreamableHttp.create_streams在 v2 下遇到ignore_initialized_notification_failure=True时会直接抛出UserError（见 server.py）。

实践建议：

• 尽量像下文 Streamable HTTP 示例那样使用Authorization请求头，它在两个包版本下行为一致；
• 如果应用提供params["auth"]或params["httpx_client_factory"]，其类型必须匹配已安装mcp包的主版本；
• 若使用了ignore_initialized_notification_failure = True，升级前需保持mcp<2或先禁用该选项；
• 上述本地mcp依赖要求不适用于由 OpenAI Responses API 托管的HostedMCPTool。

Agent 级 MCP 配置：mcp_config

除了选择传输，还可以通过Agent.mcp_config调整 MCP 工具的加载方式。在 src/agents/agent.py 中，Agent同时持有mcp_servers: list[MCPServer]和mcp_config: MCPConfig（一个 dict 字段），get_tools()会读取其中的三个键（默认值分别为False、未设置、False，见 agent.py）：

from agents import Agent agent = Agent( name="Assistant", mcp_servers=[server], mcp_config={ # 尝试把 MCP 工具 schema 转换为严格 JSON schema "convert_schemas_to_strict": True, # 若为 None，MCP 工具失败会以异常抛出， # 而不是返回模型可见的错误文本 "failure_error_function": None, # 为本地 MCP 工具名加上服务器名前缀 "include_server_in_tool_names": True, }, )

三个键的行为：

• convert_schemas_to_strict是尽力而为（best-effort）：schema 无法转换时保留原 schema；
• failure_error_function控制 MCP 工具调用失败如何呈现给模型；未设置时 SDK 使用默认的工具错误格式化器；
• 服务器级的failure_error_function会覆盖该服务器对应的Agent.mcp_config["failure_error_function"]；
• include_server_in_tool_names需显式开启。开启后每个本地 MCP 工具会以确定性生成的服务器前缀名暴露给模型，避免多个 MCP 服务器发布同名工具时冲突。生成的名字是 ASCII 安全的、遵守FunctionTool实例的名字长度上限，且不会与该 Agent 已配置的本地FunctionTool名称或已激活的 handoff 冲突。SDK 调用时仍使用原服务器上的原始 MCP 工具名。

1. 托管 MCP 服务器工具（HostedMCPTool）

托管工具把整个工具往返过程交给 OpenAI 基础设施处理：你不再从代码里拉取工具列表或发起调用，HostedMCPTool只是把服务器标签和可选的连接器元数据传给 Responses API，由模型自行检索并调用远程服务器的工具——不产生回到 Python 进程的额外回调。目前托管工具仅在与 Responses API 托管 MCP 集成兼容的 OpenAI 模型上工作。实现见 src/agents/tool.py 中的HostedMCPTool类。

基本托管 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", 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。

若希望托管工具延迟加载，设置tool_config["defer_loading"] = True并给 Agent 添加ToolSearchTool。该能力仅 OpenAI Responses 模型支持，完整配置与限制见 docs/tools.md。

托管 MCP 结果的流式输出

托管工具支持的结果流式与函数工具完全相同。要在模型继续工作的同时消费增量 MCP 输出，使用Runner.run_streamed：

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

回调可以是同步或异步函数，每次模型继续执行前需要审批数据时都会调用。

基于连接器的托管服务器

托管 MCP 也支持 OpenAI 连接器：提供connector_id和访问令牌，而不指定server_url，由 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 目录（含simple.py、on_approval.py、connectors.py、human_in_the_loop.py等）。

2. Streamable HTTP MCP 服务器

需要直接管理网络连接时使用MCPServerStreamableHttp。适合要精细控制传输、或在自有基础设施内低延迟运行服务器的场景：

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())

构造函数支持的附加选项（默认值来自 server.py 的签名）：

• client_session_timeout_seconds（默认 5）：控制 MCP ClientSession 读超时。以datetime.timedelta可表示、最小 1 微秒的正有限值设置超时，None和0表示禁用，其余值在构造时即被拒绝。源码中的_client_session_read_timeout（server.py）逐条校验了类型、有限性和微秒下限；
• use_structured_content（默认False）：切换是否优先使用tool_result.structured_content而非文本输出。默认关闭是出于向后兼容——多数 MCP 服务器仍会把结构化内容重复放在tool_result.content里，默认启用会造成内容重复；
• max_retry_attempts（默认 0，即不重试）与retry_backoff_seconds_base（默认 1.0）：为list_tools()和call_tool()增加自动重试。重试延迟按指数退避计算（base × 2^n），可选retry_backoff_seconds_max设置上限（见 server.py）；
• tool_filter：只暴露部分工具（见下文工具过滤）；
• require_approval：为本地 MCP 工具启用人在环（HITL）审批策略；
• failure_error_function：自定义呈现给模型的 MCP 工具失败消息；设为None则改为抛出错误；
• tool_meta_resolver：在call_tool()之前插入每次调用的 MCP_meta负载；
• 另有custom_data_extractor与tool_input_guardrails/tool_output_guardrails（对服务器上每个工具统一施加的输入/输出护栏）。

params字典中的连接参数在源码中也有默认值（server.py）：timeout默认 5 秒，sse_read_timeout默认 300 秒，terminate_on_close默认True；name缺省时由 URL 生成。

本地 MCP 服务器的审批策略

MCPServerStdio、MCPServerSse、MCPServerStreamableHttp都支持require_approval，接受的形态（见 server.py 中的类型定义）：

• 对所有工具生效的"always"或"never"；
• True（等同"always"）与False（等同"never"）；
• 逐工具映射，如{"delete_file": "always", "read_file": "never"}；
• 分组对象：{"always": {"tool_names": [...]}, "never": {"tool_names": [...]}}；
• 自定义可调用对象LocalMCPApprovalCallable。

async with MCPServerStreamableHttp( name="Filesystem MCP", params={"url": "http://localhost:8000/mcp"}, require_approval={"always": {"tool_names": ["delete_file"]}}, ) as server: ...

完整的暂停/恢复流程参考 docs/human_in_the_loop.md 与 examples/mcp/get_all_mcp_tools_example/main.py。

用 tool_meta_resolver 注入每次调用的元数据

当 MCP 服务器期望在_meta中收到请求元数据（如租户 ID、追踪上下文）时使用tool_meta_resolver。以下示例假设把一个 dict 作为Runner.run(...)的context传入：

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 工具输出：文本、图片与其他内容

当 MCP 结果使用内容块时，SDK 会把文本内容转为文本输出、把图片内容映射为工具输出的图片类型条目；对于音频和资源块等其他 MCP 内容类型，SDK 把该块序列化为有效 JSON 后以文本输出传递。包含多个内容块的响应以输出条目列表传递。当use_structured_content=True且存在无错误的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( 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 负责创建进程并保持管道打开，在上下文管理器退出时自动关闭。适合快速做概念验证、或服务器只暴露命令行入口的场景。其params会被组装成StdioServerParameters（command、args、env、cwd、encoding，编码默认utf-8，见 server.py）：

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)

5. MCP 服务器管理器（MCPServerManager）

服务器数量较多时，用MCPServerManager预先连接，并把连接成功的服务器暴露给 Agent。构造参数与默认值（见 src/agents/mcp/manager.py）：

• connect_timeout_seconds（默认 10）、cleanup_timeout_seconds（默认 10）：支持正有限秒数或None（禁用），0 会被拒绝（会立即超时），构造与赋值时都会校验；
• drop_failed_servers（默认True）：active_servers只包含连接成功的服务器；
• strict（默认False）：设为True时首次连接失败即抛错；
• connect_in_parallel（默认False）：为每个服务器使用独立工作器任务并行连接。

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)

从源码结构看，Manager 内部为每个服务器维护一个基于asyncio.Queue的工作器任务（_ServerWorker），connect_all()、reconnect()、cleanup_all()的调用被串行化：若某服务器已有生命周期任务在执行，其他生命周期操作会等待其结束，而不会对同一服务器并发连接或清理。关键行为：

• 失败记录在failed_servers与errors（服务器 → 错误映射）中；
• reconnect(failed_only=True)只重试失败的服务器，reconnect(failed_only=False)重启所有服务器。

API 细节参见 docs/ref/mcp/manager.md。

跨传输的公共服务器能力

工具过滤

每个 MCP 服务器都支持工具过滤，让 Agent 只看到需要的函数。过滤既可以在构造时静态配置，也可以在每次运行时动态执行。所有服务器类都继承自MCPServer基类，过滤逻辑在list_tools()返回后、缓存填充前统一应用。

静态过滤：配置简单的允许/阻止列表使用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。

提示词（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], )

分页

内置的本地 MCP 服务器类在查询工具与提示词列表时会自动跟随nextCursor：list_tools()在应用过滤或填充缓存前收集完整工具列表；list_prompts()返回带nextCursor=None的合并结果。若后续页面出错或服务器循环返回 cursor，操作会抛错，而不是暴露部分结果或写入缓存。

资源（Resources）仍使用显式分页：把list_resources()或list_resource_templates()返回的nextCursor作为cursor参数再次传入以获取下一页。

缓存

每次 Agent 运行都会对所有 MCP 服务器调用list_tools()；远程服务器可能带来可观延迟，因此所有 MCP 服务器类都提供cache_tools_list选项。只有确信工具定义很少变化时才设为True；之后需要刷新时可调用服务器实例上的invalidate_tools_cache()。源码中缓存命中路径会检查缓存是否脏以及列表是否非空（server.py），且工具在缓存前会做深拷贝（_snapshot_tools），防止调用方篡改缓存的 schema。

追踪

Tracing 会自动捕获 MCP 活动，包括：

1. 向 MCP 服务器查询工具列表的调用；
2. 工具调用中与 MCP 相关的信息。

延伸阅读

• MCP 官方规范与设计指南；
• 可运行的 stdio、SSE、Streamable HTTP 样本：examples/mcp（含filesystem_example、sse_example、streamablehttp_example、tool_filter_example、prompt_server等）；
• 包含审批与连接器的完整托管 MCP 演示：examples/hosted_mcp；
• API 参考：docs/ref/mcp/server.md、docs/ref/mcp/manager.md、docs/ref/mcp/util.md。

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python