Multi-Server Orchestration: Agents Calling Many MCP Servers at Once

An agent tackling a real enterprise task often needs tools from several systems in one reasoning session, look up a customer in the CRM, check their open tickets, search the knowledge base for a relevant policy. This lesson builds the session-management layer for holding several MCP connections concurrently within the agent itself, as an alternative to routing everything through a dedicated federation gateway (Lesson 17), and addresses a cost that grows specifically with this pattern: the context-window and tool-selection burden of a large aggregated catalog.

Managing Concurrent Sessions

from mcp.client.session import ClientSession
import asyncio

class MultiServerAgentSession:
    def __init__(self, server_configs: dict[str, dict]):
        self.server_configs = server_configs  # {"crm": {...}, "kb": {...}, "support": {...}}
        self.sessions: dict[str, ClientSession] = {}
        self.unavailable: set[str] = set()

    async def connect_all(self):
        """Connect to every configured server concurrently; isolate failures per server."""
        results = await asyncio.gather(
            *(self._connect_one(name, cfg) for name, cfg in self.server_configs.items()),
            return_exceptions=True,
        )
        for name, result in zip(self.server_configs, results):
            if isinstance(result, Exception):
                self.unavailable.add(name)
                logging.warning(f"MCP server '{name}' unavailable at session start: {result}")

    async def _connect_one(self, name: str, cfg: dict):
        session = await ClientSession.connect(cfg["url"], headers=cfg.get("headers", {}))
        self.sessions[name] = session

Merging Tool Catalogs with Namespacing

    async def get_merged_tools(self) -> list[dict]:
        """Aggregate tools from all currently-healthy servers, namespaced by server name."""
        all_tools = []
        for name, session in self.sessions.items():
            if name in self.unavailable:
                continue
            try:
                tools = await session.list_tools()
                for t in tools:
                    t.name = f"{name}.{t.name}"  # Same convention as Lesson 17's federation gateway
                all_tools.extend(tools)
            except Exception:
                self.unavailable.add(name)  # Mark unhealthy mid-session, not just at connect time
        return all_tools

    async def call_tool(self, namespaced_name: str, arguments: dict) -> dict:
        server_name, _, tool_name = namespaced_name.partition(".")
        if server_name in self.unavailable:
            return {"isError": True, "message": f"Server '{server_name}' is currently unavailable"}
        return await self.sessions[server_name].call_tool(tool_name, arguments)

Graceful Degradation Under Partial Failure

flowchart TD
    Start["Agent session starts"] --> Connect["Connect to CRM, KB, Support\nconcurrently"]
    Connect --> Check{"KB server\nreachable?"}
    Check -->|no| Degrade["Exclude kb.* tools from\nmerged catalog, mark unavailable"]
    Check -->|yes| Full["Full catalog: crm.*, kb.*, support.*"]
    Degrade --> Continue["Agent proceeds with\ncrm.* and support.* tools only"]
    Full --> Continue2["Agent proceeds with\nfull tool catalog"]
    Continue --> Retry["Periodic reconnect attempt\nto KB in background"]

    style Start fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Check fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Degrade fill:#fef2f2,stroke:#dc2626,color:#0F172A
    style Full fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Continue fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Continue2 fill:#f0fdf9,stroke:#0D9488,color:#0F172A

async def background_reconnect_loop(session_mgr: MultiServerAgentSession, interval: int = 30):
    """Periodically retry unavailable servers so a transient outage self-heals
    without requiring the agent session to be restarted."""
    while True:
        await asyncio.sleep(interval)
        for name in list(session_mgr.unavailable):
            try:
                await session_mgr._connect_one(name, session_mgr.server_configs[name])
                session_mgr.unavailable.discard(name)
                logging.info(f"MCP server '{name}' reconnected")
            except Exception:
                pass  # Still down, try again next interval

Managing Catalog Size: When More Connected Servers Stops Helping

As an agent connects to more servers, whether directly (this lesson) or via federation (Lesson 17), the aggregated tool catalog grows correspondingly, and it is worth being deliberate about the point at which this stops being a straightforward win. Beyond raw context-window token cost (every tool’s name, description, and schema occupies tokens on every turn the model considers its options), a large flat catalog increases the model’s tool-selection search space directly: more tools that are superficially similar to each other (multiple search_* variants across different servers, for instance) raises the odds of the model picking a plausible-but-wrong one, the same underlying failure mode Lesson 5 described for naming collisions within a single server, just occurring at the scale of an entire aggregated multi-server catalog.

flowchart LR
    Servers["15 connected MCP servers"] --> Catalog["Aggregated catalog:\n400 tools"]
    Catalog --> Cost1["Token cost: every turn\ncarries all 400 tool schemas"]
    Catalog --> Cost2["Selection cost: more near-duplicate\noptions increases wrong-tool-picked rate"]

    style Servers fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Catalog fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Cost1 fill:#fef2f2,stroke:#dc2626,color:#0F172A
    style Cost2 fill:#fef2f2,stroke:#dc2626,color:#0F172A

The practical mitigation is relevance scoping: connect a given agent session only to the servers genuinely relevant to its role or the task it was launched for, rather than defaulting every agent to maximal connectivity across the entire organization’s server fleet just because it is technically possible.

# Scope connections per agent role, rather than connecting every agent
# to every server that exists in the organization by default.
AGENT_ROLE_SERVERS: dict[str, list[str]] = {
    "sales_agent": ["crm", "kb"],
    "support_agent": ["kb", "support", "crm"],
    "ops_agent": ["warehouse", "crm"],
}

def build_session_for_role(role: str) -> MultiServerAgentSession:
    relevant_servers = {name: ALL_SERVER_CONFIGS[name] for name in AGENT_ROLE_SERVERS[role]}
    return MultiServerAgentSession(relevant_servers)

For cases where a broader catalog genuinely is needed but selection accuracy has degraded, a second-tier mitigation is a two-stage tool selection pattern (a lightweight preliminary step that narrows the full catalog down to a small relevant subset before the main reasoning turn sees any tool schemas at all), though this adds its own latency and complexity and is worth reaching for only once relevance scoping alone has been tried and found insufficient for a genuinely broad-scope agent.

This closes Module 8. Module 9 turns to making all of this observable and testable in production: distributed tracing across these multi-server calls, contract testing for individual tools, and patterns for tools whose execution takes longer than a single request-response cycle.