Class MCPServerManager

src/agents/mcp/manager.py:108–411

Manage MCP server lifecycles and expose only connected servers. Use this helper to keep MCP connect/cleanup on the same task and avoid run failures when a server is unavailable. The manager will attempt to connect each server and then expose the connected subset via `active_servers`

class MCPServerManager(AbstractAsyncContextManager["MCPServerManager"]):
    """Manage MCP server lifecycles and expose only connected servers.

    Use this helper to keep MCP connect/cleanup on the same task and avoid
    run failures when a server is unavailable. The manager will attempt to
    connect each server and then expose the connected subset via
    `active_servers`.

    Basic usage:
        async with MCPServerManager([server_a, server_b]) as manager:
            agent = Agent(
                name="Assistant",
                instructions="...",
                mcp_servers=manager.active_servers,
            )

    FastAPI lifespan example:
        @asynccontextmanager
        async def lifespan(app: FastAPI):
            async with MCPServerManager([server_a, server_b]) as manager:
                app.state.mcp_manager = manager
                yield

        app = FastAPI(lifespan=lifespan)

    Important behaviors:
    - `active_servers` only includes servers that connected successfully.
      `failed_servers` holds the failures and `errors` maps servers to errors.
    - `drop_failed_servers=True` removes failed servers from `active_servers`
      (recommended). If False, `active_servers` will still include all servers.
    - `strict=True` raises on the first connection failure. If False, failures
      are recorded and the run can proceed with the remaining servers.
    - `reconnect(failed_only=True)` retries failed servers and refreshes
      `active_servers`.
    - `connect_in_parallel=True` uses a dedicated worker task per server to
      allow concurrent connects while preserving task affinity for cleanup.
    """

    def __init__(
        self,
        servers: Iterable[MCPServer],
        *,
        connect_timeout_seconds: float | None = 10.0,
        cleanup_timeout_seconds: float | None = 10.0,
        drop_failed_servers: bool = True,
        strict: bool = False,
        suppress_cancelled_error: bool = True,
        connect_in_parallel: bool = False,
    ) -> None:
        self._all_servers = list(servers)
        self._active_servers = list(servers)
        self.connect_timeout_seconds = connect_timeout_seconds
        self.cleanup_timeout_seconds = cleanup_timeout_seconds
        self.drop_failed_servers = drop_failed_servers
        self.strict = strict
        self.suppress_cancelled_error = suppress_cancelled_error
        self.connect_in_parallel = connect_in_parallel
        self._workers: dict[MCPServer, _ServerWorker] = {}