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 应用中借助MCPServerManagerStreamable HTTP 传输把多个 MCP 服务器的连接、清理与重连收敛到单一生命周期管理器中避免因某个 MCP 服务器不可用导致 Agent 运行失败。读完本文你将掌握 MCP 服务器的启动方式、管理器模式下的 FastAPI 集成写法、各 REST 端点的调用方法以及MCPServerManager的核心参数与底层行为。一、示例背景为什么需要 MCP Server ManagerMCPModel 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/目录包含四个文件文件职责mcp_server.py基于 MCP Python SDK 的 Streamable HTTP 工具服务器暴露add、echo两个工具app.pyFastAPI 应用在 lifespan 中通过MCPServerManager管理 MCP 服务器并提供/health、/tools、/add、/run、/reconnect端点smoke_test.py冒烟测试在临时端口启动 MCP 服务器不调用模型即可验证管理器与应用的集成README.md运行说明本文的主体依据整个示例的依赖关系为FastAPI 应用app.py→ MCPServerManagerSDK 内部→ Streamable HTTP MCP 服务器mcp_server.py。三、启动 MCP 服务器Streamable HTTP3.1 默认启动方式在仓库根目录执行uv run python examples/mcp/manager_example/mcp_server.py服务器默认监听http://localhost:8000/mcp。这里的 MCP 服务器本身就是一个 FastAPI 风格的服务通过mcp.run(transportstreamable-http, ...)启动见 mcp_server.py 第 21-26 行。3.2 覆盖主机与端口通过环境变量覆盖export STREAMABLE_HTTP_HOST127.0.0.1 export STREAMABLE_HTTP_PORT8000对应的源码常量位于 mcp_server.py 第 5-6 行STREAMABLE_HTTP_HOST默认127.0.0.1STREAMABLE_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 fecho: {message}它们将作为后续/tools、/add、/run端点的验证对象。四、启动 FastAPI 应用应用默认监听http://127.0.0.1:9001uv run python examples/mcp/manager_example/app.py启动前可配置以下环境变量见 app.py 第 11-15 行环境变量默认值说明MCP_SERVER_URLhttp://localhost:8000/mcp主 MCP 服务器地址INACTIVE_MCP_SERVER_URLhttp://localhost:8001/mcp一个未激活的 MCP 服务器地址用于演示管理器如何丢弃连接失败的服务器USE_MCP_MANAGER1非0即启用是否启用MCPServerManager设为0可切回手动连接模式其中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( serversservers, connect_in_parallelTrue, ) 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_MANAGER0下应用显式调用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_testsmoke_test.py 的执行流程通过socket绑定一个临时空闲端口_free_port第 25-29 行并设置STREAMABLE_HTTP_HOST/STREAMABLE_HTTP_PORT后以子进程方式启动 MCP 服务器_start_mcp_server第 59-71 行。轮询等待端口就绪_wait_for_port超时 10 秒第 31-42 行若服务器提前退出则直接报错。把MCP_SERVER_URL与INACTIVE_MCP_SERVER_URL都指向该临时服务器并强制USE_MCP_MANAGER1再导入 app 模块_load_app_module第 74-87 行。注释说明把两个配置指向同一服务器是为了让冒烟测试停留在干净的 app 集成路径上。通过httpx.ASGITransport在内存中驱动 FastAPI 应用依次断言/health的connected_servers包含临时服务器 URL、failed_servers为空/tools返回的工具集合包含add与echoPOST /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}端点解析AddRequesta、b均为 int通过active_servers[0].call_tool(add, {a: req.a, b: req.b})调用 MCP 工具并返回result.model_dump(modejson)第 83-89 行。若没有可用服务器则返回 503No 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_MANAGER1时可用否则返回 400MCPServerManager is disabled。响应体为重连后活跃服务器的名称列表第 111-117 行。6.5 运行 Agent需要 OPENAI_API_KEYexport 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( nameFastAPI Agent, instructionsUse the MCP tools when needed., mcp_serversservers, model_settingsModelSettings(tool_choiceauto), ) result await Runner.run(starting_agentagent, inputreq.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 对外导出。参数默认值说明servers必填待管理的 MCP 服务器可迭代对象内部会做去重_unique_serversconnect_timeout_seconds10.0单个服务器连接超时None表示禁用超时cleanup_timeout_seconds10.0单个服务器清理超时None表示禁用超时drop_failed_serversTrue为True时active_servers只包含连接成功的服务器推荐为False时仍包含全部服务器strictFalse为True时遇到第一个连接失败立即抛出异常为False时记录失败并继续用剩余服务器运行suppress_cancelled_errorTrue取消CancelledError在生命周期操作中是否被吞掉并记录到errorsconnect_in_parallelFalse为True时为每个服务器创建独立 worker 任务实现并发连接同时保留 task 亲和性以支持清理关键行为与 docs/mcp.md 第 388-395 行的Key behaviors一致只暴露成功子集active_servers仅包含成功连接的服务器在drop_failed_serversTrue时。失败可观测failed_servers与errors分别记录失败服务器及其异常reconnect()后成功重连的服务器会从失败集合中移除_remove_failed_server。严格模式strictTrue时在第一个连接失败处抛错。重连语义reconnect(failed_onlyTrue)只重试失败服务器reconnect(failed_onlyFalse)清理并重启全部服务器。生命周期操作串行化connect_all()、reconnect()、cleanup_all()之间通过asyncio.Lock串行执行_acquire_lifecycle_lock若一个生命周期操作正在进行另一个会等待而不是并发连接/清理同一批服务器。超时校验连接/清理超时在构造与赋值时都会经过_validate_lifecycle_timeout校验——必须是正的有穷秒数或None0会被拒绝因为它会产生立即截止的 deadline。并发连接与 task 亲和性从源码结构看connect_in_parallelTrue时每个服务器由一个_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 亲和性不被破坏。八、把管理器模式与手动模式对比维度管理器模式USE_MCP_MANAGER1手动模式USE_MCP_MANAGER0连接方式进入async with时统一connect_all()逐个手动await server.connect()失败处理失败服务器自动进入failed_servers不阻断运行任一服务器失败可能直接导致启动异常暴露给 Agentmanager.active_servers成功子集手动维护的active_servers列表重连内置reconnect()可只重试失败的服务器需要自行实现重连逻辑清理退出async with自动cleanup_all()手动finally中调用cleanup()适用场景多服务器、追求容错与统一生命周期管理单服务器、简单演示九、快速上手清单启动 MCP 服务器uv run python examples/mcp/manager_example/mcp_server.py默认http://localhost:8000/mcp。启动 FastAPI 应用uv run python examples/mcp/manager_example/app.py默认http://127.0.0.1:9001。运行冒烟测试uv run python -m examples.mcp.manager_example.smoke_test。用 curl 依次验证/health、/tools、POST /add启用管理器时可用POST /reconnect重连失败服务器设置OPENAI_API_KEY后可通过POST /run让 Agent 调用 MCP 工具。关闭管理器export USE_MCP_MANAGER0对比两种模式下失败服务器的处理差异。十、进一步阅读示例说明examples/mcp/manager_example/README.mdMCP 使用指南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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考