openai-agents-python 实战：用 MCPServerManager 在 FastAPI 中统一管理多 MCP 服务器生命周期

openai-agents-python 实战：用 MCPServerManager 在 FastAPI 中统一管理多 MCP 服务器生命周期

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

本指南以仓库中的 manager_example 示例为主线，讲解如何在 FastAPI 应用中借助MCPServerManager（Streamable HTTP 传输）把多个 MCP 服务器的连接、清理与重连收敛到单一生命周期管理器中，避免因某个 MCP 服务器不可用导致 Agent 运行失败。读完本文，你将掌握 MCP 服务器的启动方式、管理器模式下的 FastAPI 集成写法、各 REST 端点的调用方法，以及MCPServerManager的核心参数与底层行为。

一、示例背景：为什么需要 MCP Server Manager

MCP（Model Context Protocol）让 Agent 能够以统一协议接入外部工具。在实际的多 Agent / 多服务场景中，一个应用往往要同时对接多个 MCP 服务器。如果每个服务器都由应用手动connect()/cleanup()，会带来两个典型问题：

• 某个服务器不可用时，整个运行失败：手动模式下，任何一个服务器的连接异常都可能中断 Agent 的Runner.run()。
• 生命周期分散、难以统一管理：连接、清理散落在各处，服务重启或重连时容易出现资源泄漏或状态不一致。

MCPServerManager正是为此设计：它在启动时统一尝试连接所有服务器，然后只把连接成功的子集（active_servers）暴露给 Agent，并把失败记录在failed_servers与errors中。示例 README 也明确说明，该示例面向 MCP Python SDK v2，并使用仓库锁定的开发环境运行；而 Agents SDK 客户端本身同时支持 MCP v1 与 v2。

仓库中的文档 docs/mcp.md 第 5 节对管理器的定位做了同样的描述：当你有多个 MCP 服务器时，用MCPServerManager提前连接它们，并将成功连接的子集交给 Agent 使用。其 API 参考位于 docs/ref/mcp/manager.md，实现位于 src/agents/mcp/manager.py。

二、示例结构总览

examples/mcp/manager_example/目录包含四个文件：

整个示例的依赖关系为：FastAPI 应用（app.py）→ MCPServerManager（SDK 内部）→ Streamable HTTP MCP 服务器（mcp_server.py）。

三、启动 MCP 服务器（Streamable HTTP）

3.1 默认启动方式

在仓库根目录执行：

uv run python examples/mcp/manager_example/mcp_server.py

服务器默认监听http://localhost:8000/mcp。这里的 MCP 服务器本身就是一个 FastAPI 风格的服务，通过mcp.run(transport="streamable-http", ...)启动（见 mcp_server.py 第 21-26 行）。

3.2 覆盖主机与端口

通过环境变量覆盖：

export STREAMABLE_HTTP_HOST=127.0.0.1 export STREAMABLE_HTTP_PORT=8000

对应的源码常量位于 mcp_server.py 第 5-6 行：STREAMABLE_HTTP_HOST默认127.0.0.1，STREAMABLE_HTTP_PORT默认8000，端口值会被int()强转，因此必须是合法整数。

3.3 示例中 MCP 服务器的工具

mcp_server.py 用@mcp.tool()装饰器定义了两个工具：

@mcp.tool() def add(a: int, b: int) -> int: return a + b @mcp.tool() def echo(message: str) -> str: return f"echo: {message}"

它们将作为后续/tools、/add、/run端点的验证对象。

四、启动 FastAPI 应用

应用默认监听http://127.0.0.1:9001：

uv run python examples/mcp/manager_example/app.py

启动前可配置以下环境变量（见 app.py 第 11-15 行）：

其中USE_MCP_MANAGER的解析逻辑为os.getenv("USE_MCP_MANAGER", "1") != "0"，因此只有显式设置为"0"才会关闭管理器。

4.1 生命周期（lifespan）中的管理器用法

核心集成点在 FastAPI 的 lifespan 中（app.py 第 31-52 行）：

@asynccontextmanager async def lifespan(app: FastAPI): server = MCPServerStreamableHttp({"url": MCP_SERVER_URL}) inactive_server = MCPServerStreamableHttp({"url": INACTIVE_MCP_SERVER_URL}) servers = [server, inactive_server] if USE_MCP_MANAGER: async with MCPServerManager( servers=servers, connect_in_parallel=True, ) as manager: app.state.mcp_manager = manager app.state.mcp_servers = servers yield return await server.connect() app.state.mcp_servers = servers app.state.active_servers = [server] try: yield finally: await server.cleanup()

要点：

• 两个服务器都通过MCPServerStreamableHttp({"url": ...})构造，其中inactive_server指向http://localhost:8001/mcp，正常情况下该地址没有服务，用来演示管理器对失败服务器的"丢弃"行为。
• 管理器模式启用时，async with MCPServerManager(...)进入时自动connect_all()，退出时自动cleanup_all()（见 manager.py 中__aenter__/__aexit__的实现，第 261-267 行）；管理器实例挂到app.state.mcp_manager上供请求处理函数使用。
• 手动模式（USE_MCP_MANAGER=0）下，应用显式调用server.connect()与server.cleanup()，且只把主服务器放入active_servers——这正是管理器模式试图消除的"手动、分散"写法。

这种"同一任务内管理生命周期"的做法非常重要：MCPServerManager的 docstring 明确指出，它保证 MCP 的 connect/cleanup 在同一个 task 中执行，从而避免服务器不可用时导致 run 失败，也让清理逻辑得以在async with退出时可靠触发。

五、运行冒烟测试（不调用模型）

冒烟测试用于在不调用模型的前提下验证 MCP 管理器与应用端点的集成：

uv run python -m examples.mcp.manager_example.smoke_test

smoke_test.py 的执行流程：

1. 通过socket绑定一个临时空闲端口（_free_port，第 25-29 行），并设置STREAMABLE_HTTP_HOST/STREAMABLE_HTTP_PORT后以子进程方式启动 MCP 服务器（_start_mcp_server，第 59-71 行）。
2. 轮询等待端口就绪（_wait_for_port，超时 10 秒，第 31-42 行），若服务器提前退出则直接报错。
3. 把MCP_SERVER_URL与INACTIVE_MCP_SERVER_URL都指向该临时服务器，并强制USE_MCP_MANAGER=1，再导入 app 模块（_load_app_module，第 74-87 行）。注释说明：把两个配置指向同一服务器，是为了让冒烟测试停留在干净的 app 集成路径上。
4. 通过httpx.ASGITransport在内存中驱动 FastAPI 应用，依次断言：
   • /health的connected_servers包含临时服务器 URL、failed_servers为空；
   • /tools返回的工具集合包含add与echo；
   • POST /add {"a": 2, "b": 3}返回的文本内容中包含"5"。

这组断言同时验证了"失败服务器会被丢弃（此场景下无失败）"以及"管理器只暴露连接成功的服务器"这一核心行为。

六、HTTP 端点调用指南

应用启动后即可用 curl 验证（对应 app.py 第 58-117 行的端点实现）。

6.1 健康检查

curl http://127.0.0.1:9001/health

响应示例（管理器模式）：

{ "connected_servers": ["FastAPI Example Server"], "failed_servers": [] }

管理器模式下，connected_servers与failed_servers分别来自manager.active_servers与manager.failed_servers；由于示例中http://localhost:8001/mcp没有服务，该服务器会被记录为失败并从active_servers中剔除，因此实际只会看到主服务器（app.py 第 58-71 行）。

6.2 列出 MCP 工具

curl http://127.0.0.1:9001/tools

实现中取第一个活跃服务器并调用list_tools()，返回工具名列表（第 74-80 行）：

{ "tools": ["add", "echo"] }

若没有活跃服务器，则返回空列表{"tools": []}。

6.3 直接调用 MCP 工具

curl -X POST http://127.0.0.1:9001/add \ -H 'Content-Type: application/json' \ -d '{"a": 2, "b": 3}'

端点解析AddRequest（a、b均为 int），通过active_servers[0].call_tool("add", {"a": req.a, "b": req.b})调用 MCP 工具并返回result.model_dump(mode="json")（第 83-89 行）。若没有可用服务器则返回 503（No MCP servers available）。

6.4 重连失败的 MCP 服务器（需启用管理器）

curl -X POST http://127.0.0.1:9001/reconnect \ -H 'Content-Type: application/json' \ -d '{"failed_only": true}'

• failed_only: true（默认）：仅重试之前失败的服务器；
• failed_only: false：清理并重启所有服务器。

该端点仅在USE_MCP_MANAGER=1时可用，否则返回 400（MCPServerManager is disabled）。响应体为重连后活跃服务器的名称列表（第 111-117 行）。

6.5 运行 Agent（需要 OPENAI_API_KEY）

export OPENAI_API_KEY=... curl -X POST http://127.0.0.1:9001/run \ -H 'Content-Type: application/json' \ -d '{"input": "Add 4 and 9."}'

/run端点用manager.active_servers构造 Agent，并执行Runner.run()（app.py 第 92-108 行）：

agent = Agent( name="FastAPI Agent", instructions="Use the MCP tools when needed.", mcp_servers=servers, model_settings=ModelSettings(tool_choice="auto"), ) result = await Runner.run(starting_agent=agent, input=req.input) return {"output": result.final_output}

未设置OPENAI_API_KEY时返回 400。注意这里的servers来自_get_active_servers()：管理器模式下即manager.active_servers，只有连接成功的服务器才会进入 Agent 的工具集，这正是管理器容错能力的直接体现。

七、MCPServerManager 核心参数与底层行为

MCPServerManager的完整构造参数定义在 src/agents/mcp/manager.py 第 192-202 行，并通过 agents.mcp 对外导出。

关键行为（与 docs/mcp.md 第 388-395 行的"Key behaviors"一致）：

• 只暴露成功子集：active_servers仅包含成功连接的服务器（在drop_failed_servers=True时）。
• 失败可观测：failed_servers与errors分别记录失败服务器及其异常；reconnect()后成功重连的服务器会从失败集合中移除（_remove_failed_server）。
• 严格模式：strict=True时在第一个连接失败处抛错。
• 重连语义：reconnect(failed_only=True)只重试失败服务器；reconnect(failed_only=False)清理并重启全部服务器。
• 生命周期操作串行化：connect_all()、reconnect()、cleanup_all()之间通过asyncio.Lock串行执行（_acquire_lifecycle_lock），若一个生命周期操作正在进行，另一个会等待而不是并发连接/清理同一批服务器。
• 超时校验：连接/清理超时在构造与赋值时都会经过_validate_lifecycle_timeout校验——必须是正的有穷秒数或None，0会被拒绝（因为它会产生立即截止的 deadline）。

并发连接与 task 亲和性

从源码结构看，connect_in_parallel=True时每个服务器由一个_ServerWorker负责：worker 内部维护一个asyncio.Queue，顺序处理connect与cleanup命令（manager.py 第 37-110 行）。这样多个服务器的连接可以并发进行，同时connect/cleanup仍发生在同一 worker task 内，避免破坏依赖 task 上下文的库（例如 AnyIO 的 cancel scope）。

超时实现的细节

超时并非简单调用asyncio.wait_for：_run_with_timeout_in_task（第 113-148 行）优先使用 Python 3.11+ 的asyncio.timeout上下文管理器，并在旧版本上通过loop.call_later配合task.cancel()实现"任务内超时"，以保证连接与清理的 task 亲和性不被破坏。

八、把管理器模式与手动模式对比

九、快速上手清单

1. 启动 MCP 服务器：uv run python examples/mcp/manager_example/mcp_server.py（默认http://localhost:8000/mcp）。
2. 启动 FastAPI 应用：uv run python examples/mcp/manager_example/app.py（默认http://127.0.0.1:9001）。
3. 运行冒烟测试：uv run python -m examples.mcp.manager_example.smoke_test。
4. 用 curl 依次验证/health、/tools、POST /add；启用管理器时可用POST /reconnect重连失败服务器；设置OPENAI_API_KEY后可通过POST /run让 Agent 调用 MCP 工具。
5. 关闭管理器：export USE_MCP_MANAGER=0，对比两种模式下失败服务器的处理差异。

十、进一步阅读

• 示例说明：examples/mcp/manager_example/README.md
• MCP 使用指南：docs/mcp.md（第 5 节为 MCP server manager）
• 管理器 API 参考：docs/ref/mcp/manager.md
• 核心实现：src/agents/mcp/manager.py
• 管理器导出定义：src/agents/mcp/init.py
• 冒烟测试：examples/mcp/manager_example/smoke_test.py

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