feat: add MCPServerManager for safely managing server lifecycle (#2350)

Author: seratch

examples/mcp/manager_example/README.md

@@ -0,0 +1,71 @@
+# MCP Manager Example (FastAPI)
+
+This example shows how to use `MCPServerManager` to keep MCP server lifecycle
+management in a single task inside a FastAPI app with the Streamable HTTP
+transport.
+
+## Run the MCP server (Streamable HTTP)
+
+```
+uv run python examples/mcp/manager_example/mcp_server.py
+```
+
+The server listens at `http://localhost:8000/mcp` by default.
+
+You can override the host/port with:
+
+```
+export STREAMABLE_HTTP_HOST=127.0.0.1
+export STREAMABLE_HTTP_PORT=8000
+```
+
+This example also configures an inactive MCP server at
+`http://localhost:8001/mcp` to demonstrate how the manager drops failed
+servers. You can override it with:
+
+```
+export INACTIVE_MCP_SERVER_URL=http://localhost:8001/mcp
+```
+
+## Run the FastAPI app
+
+```
+uv run python examples/mcp/manager_example/app.py
+```
+
+The app listens at `http://127.0.0.1:9001`.
+
+## Toggle MCP manager usage
+
+By default, the app uses `MCPServerManager`. To disable it:
+
+```
+export USE_MCP_MANAGER=0
+```
+
+## Try the endpoints
+
+```
+curl http://127.0.0.1:9001/health
+curl http://127.0.0.1:9001/tools
+curl -X POST http://127.0.0.1:9001/add \
+  -H 'Content-Type: application/json' \
+  -d '{"a": 2, "b": 3}'
+```
+
+Reconnect failed MCP servers (manager must be enabled):
+
+```
+curl -X POST http://127.0.0.1:9001/reconnect \
+  -H 'Content-Type: application/json' \
+  -d '{"failed_only": true}'
+```
+
+To use `/run`, set `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."}'
+```

examples/mcp/manager_example/app.py

@@ -0,0 +1,130 @@
+import os
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+
+from agents import Agent, Runner
+from agents.mcp import MCPServer, MCPServerManager, MCPServerStreamableHttp
+from agents.model_settings import ModelSettings
+
+MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8000/mcp")
+INACTIVE_MCP_SERVER_URL = os.getenv("INACTIVE_MCP_SERVER_URL", "http://localhost:8001/mcp")
+APP_HOST = "127.0.0.1"
+APP_PORT = 9001
+USE_MCP_MANAGER = os.getenv("USE_MCP_MANAGER", "1") != "0"
+
+
+class AddRequest(BaseModel):
+    a: int
+    b: int
+
+
+class RunRequest(BaseModel):
+    input: str
+
+
+class ReconnectRequest(BaseModel):
+    failed_only: bool = True
+
+
+@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()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+@app.get("/health")
+async def health() -> dict[str, object]:
+    if USE_MCP_MANAGER:
+        manager: MCPServerManager = app.state.mcp_manager
+        return {
+            "connected_servers": [server.name for server in manager.active_servers],
+            "failed_servers": [server.name for server in manager.failed_servers],
+        }
+
+    active_servers = _get_active_servers()
+    return {
+        "connected_servers": [server.name for server in active_servers],
+        "failed_servers": [],
+    }
+
+
+@app.get("/tools")
+async def list_tools() -> dict[str, object]:
+    active_servers = _get_active_servers()
+    if not active_servers:
+        return {"tools": []}
+    tools = await active_servers[0].list_tools()
+    return {"tools": [tool.name for tool in tools]}
+
+
+@app.post("/add")
+async def add(req: AddRequest) -> dict[str, object]:
+    active_servers = _get_active_servers()
+    if not active_servers:
+        raise HTTPException(status_code=503, detail="No MCP servers available")
+    result = await active_servers[0].call_tool("add", {"a": req.a, "b": req.b})
+    return {"result": result.model_dump(mode="json")}
+
+
+@app.post("/run")
+async def run_agent(req: RunRequest) -> dict[str, object]:
+    if not os.getenv("OPENAI_API_KEY"):
+        raise HTTPException(status_code=400, detail="OPENAI_API_KEY is required")
+
+    servers = _get_active_servers()
+    if not servers:
+        raise HTTPException(status_code=503, detail="No MCP servers available")
+
+    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}
+
+
+@app.post("/reconnect")
+async def reconnect(req: ReconnectRequest) -> dict[str, object]:
+    if not USE_MCP_MANAGER:
+        raise HTTPException(status_code=400, detail="MCPServerManager is disabled")
+    manager: MCPServerManager = app.state.mcp_manager
+    servers = await manager.reconnect(failed_only=req.failed_only)
+    return {"connected_servers": [server.name for server in servers]}
+
+
+def _get_active_servers() -> list[MCPServer]:
+    if USE_MCP_MANAGER:
+        manager: MCPServerManager = app.state.mcp_manager
+        return list(manager.active_servers)
+    return list(app.state.active_servers)
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host=APP_HOST, port=APP_PORT)

examples/mcp/manager_example/mcp_server.py

@@ -0,0 +1,26 @@
+import os
+
+from mcp.server.fastmcp import FastMCP
+
+STREAMABLE_HTTP_HOST = os.getenv("STREAMABLE_HTTP_HOST", "127.0.0.1")
+STREAMABLE_HTTP_PORT = int(os.getenv("STREAMABLE_HTTP_PORT", "8000"))
+
+mcp = FastMCP(
+    "FastAPI Example Server",
+    host=STREAMABLE_HTTP_HOST,
+    port=STREAMABLE_HTTP_PORT,
+)
+
+
+@mcp.tool()
+def add(a: int, b: int) -> int:
+    return a + b
+
+
+@mcp.tool()
+def echo(message: str) -> str:
+    return f"echo: {message}"
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")

src/agents/agent.py

@@ -118,7 +118,8 @@ class AgentBase(Generic[TContext]):
 
     NOTE: You are expected to manage the lifecycle of these servers. Specifically, you must call
     `server.connect()` before passing it to the agent, and `server.cleanup()` when the server is no
-    longer needed.
+    longer needed. Consider using `MCPServerManager` from `agents.mcp` to keep connect/cleanup
+    in the same task.
     """
 
     mcp_config: MCPConfig = field(default_factory=lambda: MCPConfig())

src/agents/mcp/__init__.py

@@ -1,4 +1,5 @@
 try:
+    from .manager import MCPServerManager
     from .server import (
         MCPServer,
         MCPServerSse,
@@ -28,6 +29,7 @@
     "MCPServerStdioParams",
     "MCPServerStreamableHttp",
     "MCPServerStreamableHttpParams",
+    "MCPServerManager",
     "MCPUtil",
     "ToolFilter",
     "ToolFilterCallable",