import contextlib import datetime from collections.abc import AsyncIterator from typing import TYPE_CHECKING, Any from mcp import ClientSession from typing_extensions import Unpack from fastmcp import _install_hints from fastmcp.client.transports.base import ClientTransport, SessionKwargs from fastmcp.client.transports.memory import FastMCPTransport from fastmcp.mcp_config import ( MCPConfig, MCPServerTypes, RemoteMCPServer, StdioMCPServer, TransformingRemoteMCPServer, TransformingStdioMCPServer, _coerce_tool_transform_configs, ) from fastmcp.utilities.logging import get_logger if TYPE_CHECKING: from fastmcp.server.server import FastMCP logger = get_logger(__name__) class MCPConfigTransport(ClientTransport): """Transport for connecting to one or more MCP servers defined in an MCPConfig. This transport provides a unified interface to multiple MCP servers defined in an MCPConfig object or dictionary matching the MCPConfig schema. It supports two key scenarios: 1. If the MCPConfig contains exactly one server, it creates a direct transport to that server. 2. If the MCPConfig contains multiple servers, it creates a composite client by mounting all servers on a single FastMCP instance, with each server's name, by default, used as its mounting prefix. In the multiserver case, tools are accessible with the prefix pattern `{server_name}_{tool_name}` and resources with the pattern `protocol://{server_name}/path/to/resource`. This is particularly useful for creating clients that need to interact with multiple specialized MCP servers through a single interface, simplifying client code. Examples: ```python from fastmcp import Client # Create a config with multiple servers config = { "mcpServers": { "weather": { "url": "https://weather-api.example.com/mcp", "transport": "http" }, "calendar": { "url": "https://calendar-api.example.com/mcp", "transport": "http" } } } # Create a client with the config client = Client(config) async with client: # Access tools with prefixes weather = await client.call_tool("weather_get_forecast", {"city": "London"}) events = await client.call_tool("calendar_list_events", {"date": "2023-06-01"}) # Access resources with prefixed URIs icons = await client.read_resource("weather://weather/icons/sunny") ``` """ def __init__(self, config: MCPConfig | dict, name_as_prefix: bool = True): if isinstance(config, dict): config = MCPConfig.from_dict(config) self.config = config self.name_as_prefix = name_as_prefix self._transports: list[ClientTransport] = [] if not self.config.mcpServers: raise ValueError("No MCP servers defined in the config") # For single server, create transport eagerly so it can be inspected if len(self.config.mcpServers) == 1: self.transport = next(iter(self.config.mcpServers.values())).to_transport() self._transports.append(self.transport) @contextlib.asynccontextmanager async def connect_session( self, **session_kwargs: Unpack[SessionKwargs] ) -> AsyncIterator[ClientSession]: # Single server - delegate directly to pre-created transport if len(self.config.mcpServers) == 1: async with self.transport.connect_session(**session_kwargs) as session: yield session return # Multiple servers - create composite with mounted proxies, connecting # each ProxyClient so its underlying transport session stays alive for # the duration of this context (fixes session persistence for # streamable-http backends — see #2790). try: from fastmcp.server.server import FastMCP except ImportError as exc: raise ImportError( _install_hints.full_package("MCP configs with multiple servers") ) from exc timeout = session_kwargs.get("read_timeout_seconds") composite = FastMCP[Any](name="MCPRouter") async with contextlib.AsyncExitStack() as stack: # Close any previous transports from prior connections to avoid leaking for t in self._transports: await t.close() self._transports = [] for name, server_config in self.config.mcpServers.items(): try: transport, _client, proxy = await self._create_proxy( name, server_config, timeout, stack ) except Exception: # Broad catch is intentional: failure modes # are diverse (OSError, TimeoutError, RuntimeError, etc.) # and the whole point is to skip any server that can't connect. logger.warning( "Failed to connect to MCP server %r, skipping", name, exc_info=True, ) continue self._transports.append(transport) composite.mount(proxy, namespace=name if self.name_as_prefix else None) if not self._transports: raise ConnectionError("All MCP servers failed to connect") async with FastMCPTransport(mcp=composite).connect_session( **session_kwargs ) as session: yield session async def _create_proxy( self, name: str, config: MCPServerTypes, timeout: datetime.timedelta | None, stack: contextlib.AsyncExitStack, ) -> tuple[ClientTransport, Any, "FastMCP[Any]"]: """Create underlying transport, proxy client, and proxy server for a single backend. The ProxyClient is connected via the AsyncExitStack *before* being passed to create_proxy so the factory sees it as connected and reuses the same session for all tool calls (instead of creating fresh copies). Returns a tuple of (transport, proxy_client, proxy_server). """ # Import here to avoid circular dependency from fastmcp.server.providers.proxy import StatefulProxyClient from fastmcp.server.server import create_proxy tool_transforms = None include_tags = None exclude_tags = None # Handle transforming servers - call base class to_transport() for underlying transport if isinstance(config, TransformingStdioMCPServer): transport = StdioMCPServer.to_transport(config) tool_transforms = config.tools include_tags = config.include_tags exclude_tags = config.exclude_tags elif isinstance(config, TransformingRemoteMCPServer): transport = RemoteMCPServer.to_transport(config) tool_transforms = config.tools include_tags = config.include_tags exclude_tags = config.exclude_tags else: transport = config.to_transport() client = StatefulProxyClient(transport=transport, timeout=timeout) # Connect the client *before* create_proxy so _create_client_factory # detects it as connected and reuses it for all tool calls, preserving # the session ID across requests. StatefulProxyClient is used instead # of ProxyClient because its context-restoring handler wrappers prevent # stale ContextVars in the reused session's receive loop. # # StatefulProxyClient.__aexit__ is a no-op (by design, for the # new_stateful() use case), so we cannot rely on enter_async_context # alone to clean up. Instead we connect manually and push an # explicit force-disconnect callback so the subprocess is terminated # when the AsyncExitStack unwinds. await client.__aenter__() # Callbacks run LIFO: transport.close() must run *after* # client._disconnect so push it first. stack.push_async_callback(transport.close) stack.push_async_callback(client._disconnect, force=True) # Create proxy without include_tags/exclude_tags - we'll add them after tool transforms proxy = create_proxy( client, name=f"Proxy-{name}", ) # Add tool transforms FIRST - they may add/modify tags if tool_transforms: from fastmcp.server.transforms import ToolTransform proxy.add_transform( ToolTransform(_coerce_tool_transform_configs(tool_transforms)) ) # Then add enabled filters - they filter based on tags if include_tags: proxy.enable(tags=set(include_tags), only=True) if exclude_tags: proxy.disable(tags=set(exclude_tags)) return transport, client, proxy async def close(self): for transport in self._transports: await transport.close() def __repr__(self) -> str: return f""