就MCP涉及的消息交换模式来说，既有从客户端到服务端的请求，比如原语（或者组件）的读取与工具的执行；也有从服务器到客户端的请求与通知，比如日志传递、进度报告、LLM采样和信息征询（Elicitation）等，所以MCP的传输层必需支持双向通信。我们知道FastMCP提供了四种传输协议：In-Memory、STDIO、SSE和Streamable-HTTP，其中In-Memory是为自家客户端设计的，其他三种则是MCP框架普遍支持的传输协议。就双工通信来说，还有一种通信协议不得不提，那就是WebSocket。这四种就是MultiServerMCPClient支持的四种传输协议。

当我们初始化一个MultiServerMCPClient对象的时候，需要配置针对MCP服务器的连接。针对连接的设置体现在__init__方法的connections参数上，这是一个将Key作为MCP服务器名称的字典，Value就是指向服务器的连接，而连接的设置决定于采用的传输协议。表示连接的Connection为四种连接类型的联合，它们分别对应上述的四种传输协议。

class MultiServerMCPClient:
    def __init__(
        self,
        connections: dict[str, Connection] | None = None,
        *,
        callbacks: Callbacks | None = None,
        tool_interceptors: list[ToolCallInterceptor] | None = None,
        tool_name_prefix: bool = False,
    ) -> None

    Connection = (StdioConnection | SSEConnection | StreamableHttpConnection | WebsocketConnection)

1. StdioConnection

STDIO基于标准输入/输出，专为本地开发和桌面应用设计。在这种传输模式下，客户端将服务器作为一个子进程启动，通过标准输入（stdin）和标准输出（stdout）发送请求和接收响应。由于采用跨进程通信，其极低延迟，无需配置网络端口或身份验证。安全性也高，所以适合本地工具集成、CLI工具开发。针对STDIO的连接通过如下这个名为StdioConnection的TypedDict表示，FastMCP中与之对等的类型是StdioTransport。

class StdioConnection(TypedDict):
    transport: Literal["stdio"]
    command: str
    args: list[str]
    env: NotRequired[dict[str, str] | None]
    cwd: NotRequired[str | Path | None]
    encoding: NotRequired[str]
    encoding_error_handler: NotRequired[EncodingErrorHandler]
    session_kwargs: NotRequired[dict[str, Any] | None]

EncodingErrorHandler = Literal["strict", "ignore", "replace"]

各字段说明如下：

• transport：指定传输方式，固定为stdio；
• command：要执行的命令。例如python、node或直接是可执行文件的路径；
• args：传递给命令的参数列表。如果命令是python，参数可能是["server.py"]；
• env：环境变量。如果你需要以环境变量的形式提供所需的配置，可以写在这里；
• cwd：当前工作目录。指定在哪个文件夹下运行该命令；
• encoding：指定输入输出的编码格式，默认通常是utf-8；
• encoding_error_handler：编码错误处理程序。当遇到无法解析的字符时，决定是报错、忽略还是替换；
• session_kwargs：底层的会话参数。允许向底层的MCP SDK会话传递额外的自定义设置。

如果MCP服务器程序定义在server.py中，如下这种创建的MultiServerMCPClient的方式会以子进程的形式启动服务器，然后以STDIO协议与它建立连接。

client = MultiServerMCPClient(
    {
        "server": {
            "transport": "stdio",
            "command": "python",
            "args": ["server.py"],
        }
    }
)

2. SSEConnection

SSE是基于HTTP的单向推送协议，允许服务器通过一条持久的HTTP连接持续向客户端推送数据。SSE是单向的（仅用于服务器推送到客户端）,为了实现客户端与MCP服务端之间的“双向对话”，它采用了双通道设计：

• 下行通道 (SSE Connection): 这是一个从服务端到客户端的长连接。客户端请求服务器的一个特定端点（比如/sse），服务器保持连接不挂断。服务器通过这个通道把工具执行结果、通知、进度等推送给客户端;
• 上行通道 (POST Request)：这是一个从客户端到服务端的短连接（比如/messages）。每当客户端想要调用一个工具或发送指令时，它会发起一个标准的POST请求，发完连接就断开了。

针对SSE的连接通过如下这个名为SSEConnection的TypedDict表示，FastMCP中与之对等的类型是SSETransport。

class SSEConnection(TypedDict):
    transport: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, Any] | None]
    timeout: NotRequired[float]
    sse_read_timeout: NotRequired[float]
    session_kwargs: NotRequired[dict[str, Any] | None]
    httpx_client_factory: NotRequired[McpHttpClientFactory | None]
    auth: NotRequired[httpx.Auth]

class McpHttpClientFactory(Protocol):
    def __call__(
        self,
        headers: dict[str, str] | None = None,
        timeout: httpx.Timeout | None = None,
        auth: httpx.Auth | None = None,
    ) -> httpx.AsyncClient

各字段说明如下：

• transport: 指定传输方式，固定为sse；
• url: MCP服务器的SSE接口地址。如果在本地以SSE启动FastMCP服务器，默认地址为http://127.0.0.1:8000/sse
• headers: 自定义HTTP请求头;
• timeout: 建立连接的超时时间，默认5秒；
• sse_read_timeout: 持续监听的超时时间。这是客户端在没有收到任何新事件的情况下，保持连接开启的最长时间，默认300秒；
• session_kwargs: 透传给底层ClientSession的参数;
• httpx_client_factory: 这是一个函数工厂，用于生成自定义的httpx.AsyncClient;
• auth: HTTP认证信息（如Basic Auth或OAuth等）。

3.StreamableHttpConnection

SSE已经是一个过时的协议，Streamable-HTTP为SSE的升级版。和SSE一样，Streamable-HTTP也通过建立两个连接的方式实现双工通信，但它实现得更加灵活：

• 两个连接对应的终结点共享相同的路径（比如/mcp），而SSE的两个通道具有对应的终结点的路径（比如/sse和/messages）；
• 客户端利用上行通道发送POST请求执行相应的操作，如果操作没用采用后台任务的形式被调度执行，会立即执行返回的结果会利用此连接返回；SSE总是利用下行通道（sse长连接）以通知的形式返回操作执行的结果；
• 由于上行通道可以用于响应POST请求的结果，下行通道未必能用得上（比如在一个Session中就单纯地执行一次工具调用），所以SSE长连接会采用延迟创建的方式；SSE发送的第一个GET请求就是为了创建sse长连接；
• 如何支持HTTP2和HTTP3（QUIC），可以直接利用它们提供的多路复用，此时不必创建双连接；SSE会忽略通信双方针对HTTP2/3的支持。

针对Streamable-HTTP的连接通过如下这个名为StreamableHttpConnection的TypedDict表示，FastMCP中与之对等的类型是StreamableHttpTransport。除了transport字段固定设置为streamable_http，多了一个terminate_on_close字段外（表示在连接关闭时是否终结Session）,StreamableHttpConnection和SSEConnection具有相同的设置。

class StreamableHttpConnection(TypedDict):
    transport: Literal["streamable_http"]
    url: str
    headers: NotRequired[dict[str, Any] | None]
    timeout: NotRequired[timedelta]
    sse_read_timeout: NotRequired[timedelta]
    terminate_on_close: NotRequired[bool]
    session_kwargs: NotRequired[dict[str, Any] | None]
    httpx_client_factory: NotRequired[McpHttpClientFactory | None]
    auth: NotRequired[httpx.Auth]

4. WebsocketConnection

WebSocket是一种在单个TCP连接上进行全双工通信的协议。它解决了传统HTTP协议中服务器无法主动推送数据的痛点，是实现实时交互的核心技术。其核心特点包括：

• 双向通信：服务器和客户端地位平等，任何一方都可以随时主动向对方发送数据；
• 持久连接：一旦握手成功，连接将一直保持（直到一方主动关闭），无需像HTTP那样频繁地建立和断开TCP连接；
• 更小的开销：在建立连接后，后续交换的数据帧头部非常小（通常仅几个字节），大幅节省了带宽资源；
• 基于 HTTP 握手：它通过 HTTP 协议发起初始连接（Upgrade 机制），因此能很好地兼容现有的80/443端口，不容易被防火墙拦截。

针对WebSocket的连接通过如下这个名为WebsocketConnection的TypedDict表示。表示传输类型的transport字段固定为websocket，必需的配置连接MCP服务器终结点的URL（比如wss://://example.com）。

class WebsocketConnection(TypedDict):
    transport: Literal["websocket"]
    url: str
    session_kwargs: NotRequired[dict[str, Any] | None]

5. 连接管理

MultiServerMCPClient有如下这个有意思的设计：它重写了__aenter__和__aexit__方法，但是直接抛出NotImplementedError，并通过一段精心设计的错误消息告诉不能像很多客户端对象一样将MultiServerMCPClient作为一个上下文管理器。可能设计者觉得这一点太重要了，以至于他要画蛇添足似的重写一个抛出异常的方法提醒我们这一点。

class MultiServerMCPClient:
    async def __aenter__(self) -> "MultiServerMCPClient":
        raise NotImplementedError(ASYNC_CONTEXT_MANAGER_ERROR)

    def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        raise NotImplementedError(ASYNC_CONTEXT_MANAGER_ERROR)

ASYNC_CONTEXT_MANAGER_ERROR = (
    "As of langchain-mcp-adapters 0.1.0, MultiServerMCPClient cannot be used as a "
    "context manager (e.g., async with MultiServerMCPClient(...)). "
    "Instead, you can do one of the following:\n"
    "1. client = MultiServerMCPClient(...)\n"
    "   tools = await client.get_tools()\n"
    "2. client = MultiServerMCPClient(...)\n"
    "   async with client.session(server_name) as session:\n"
    "       tools = await load_mcp_tools(session)"
)

从提供的错误消息可知，我们不能像使用fastmcp.client.Client一样，将async with应用到MultiServerMCPClient来管理其声明周期，MultiServerMCPClient更倾向于以单例对象的方式被使用。如果我们只需要一次单一调用，直接调用对应的方法就可以了，此时它采用的是一种无状态的交互模式：先建立Session、然后执行操作最后关闭Session。如果需要多个操作需要在会话维持的状态下进行，或者涉及服务端针对客户端的反向请求和通知，就调用session方法创建针对某个服务器的ClientSession对象，并调用哪些基于Session的全局函数。