第 5 章:客户端与传输层

前几章都在讲服务器。这一章讲对称的另一半:Client。它最漂亮的设计是——你几乎不用关心传输方式,一个入参它自己猜。

5.1 Client 是什么

Client(client/client.py:130)是 FastMCP 的客户端门面:连上任意 MCP 服务器,然后 call_tool / list_tools / read_resource / get_prompt。它也负责客户端侧的回调能力:sampling(服务器反过来请求你这边的大模型)、elicitation(向用户要输入)、roots、进度、日志。

最小用法(概念示意):

# 示意,非源码。连上一个 HTTP 服务器并调工具
from fastmcp import Client

async with Client("http://localhost:8000/mcp") as client:
    tools = await client.list_tools()
    result = await client.call_tool("add", {"a": 1, "b": 2})

注意 Client 是惰性导入的:fastmcp/__init__.py 的 __getattr__ 只有你真访问 fastmcp.Client 时才导入客户端那一串依赖(源码注释说这是为了让「只做服务器」的用户不用为客户端导入链买单,见 #3292)。

5.2 核心设计:一个入参,推断传输

Client(transport) 的 transport 参数类型极宽——URL、文件路径、FastMCP 对象、配置字典……都行。构造时调 infer_transport(client/transports/inference.py:65)把它归一化成具体传输类。规则:

你传入	推断出的传输	场景
已是 `ClientTransport`	原样用	你自己精确指定
`FastMCP` / `FastMCP1Server` 对象	`FastMCPTransport`(内存)	进程内,零网络(测试用)
`.py` 文件路径	`PythonStdioTransport`	起个 Python 子进程,stdio 通信
`.js` 文件路径	`NodeStdioTransport`	起个 Node 子进程
`http://` / `https://` URL	`StreamableHttpTransport`(默认)	远端 HTTP 服务器
以 `/sse` 结尾的 URL	`SSETransport`	老式 SSE 端点
dict / `MCPConfig`	`MCPConfigTransport`	一次连多台服务器

真实分派逻辑(inference.py:65 起,节选):

# client/transports/inference.py:65 infer_transport —— 按类型分派
if isinstance(transport, ClientTransport):
    return transport
elif _is_fastmcp_server(transport):
    inferred_transport = FastMCPTransport(mcp=transport)         # 内存传输
elif isinstance(transport, Path | str) and Path(transport).exists():
    if str(transport).endswith(".py"):
        inferred_transport = PythonStdioTransport(script_path=transport)
    ...
elif isinstance(transport, AnyUrl | str) and str(transport).startswith("http"):
    ...  # sse 结尾 → SSETransport,否则 StreamableHttpTransport
elif isinstance(transport, dict | MCPConfig):
    inferred_transport = MCPConfigTransport(config=transport)

设计价值: 「连本地脚本」「连远端 HTTP」「连内存对象」「连一堆服务器」在用户眼里是同一个 API,只是入参不同。传输差异被 ClientTransport 抽象(client/transports/base.py:36)彻底藏起来。

5.3 杀手锏:内存传输 = 零网络测试

把一个 FastMCP 服务器对象直接传给 Client,infer_transport 会给你 FastMCPTransport(client/transports/memory.py:19)——客户端和服务器在同一进程内直接对话,不起子进程、不开端口、不走网络:

# 示意,非源码。进程内测一台服务器,零网络
from fastmcp import FastMCP, Client

mcp = FastMCP("test")

@mcp.tool
def add(a: int, b: int) -> int:
    return a + b

async def test():
    async with Client(mcp) as client:        # ← 直接传服务器对象
        r = await client.call_tool("add", {"a": 2, "b": 3})
        assert r.data == 5

这是 FastMCP 测试体验的核心:同一份代码,测试时走内存、生产时走 stdio/HTTP,业务逻辑一字不改。 仓库自己的测试套件(tests/)大量用这个模式。

5.4 多服务器:MCPConfig

传一个配置字典(形如 {"mcpServers": {"weather": {...}, "calendar": {...}}}),infer_transport 给你 MCPConfigTransport(client/transports/config.py:29)。它把多台服务器用各自的名字做前缀挂成一个统一客户端:工具叫 weather_get_forecast、calendar_list_events。若配置里只有一台,就直连不加前缀。这和第 4 章服务器侧的命名空间是同一个思路,只是发生在客户端。

5.5 连接生命周期

Client 是 async 上下文管理器(client/client.py:534 的 __aenter__ / __aexit__)。进入时 _connect(:540)建会话、跑 MCP 初始化握手(initialize,:483),协商双方能力(sampling/elicitation/roots 支持情况)。退出时 _disconnect(:630)带一个安全超时兜底,防止无响应的服务器把关闭卡死。auto_initialize 默认开,所以你通常不用手动握手。

代码地图

主题	文件路径	符号名
客户端门面	`fastmcp_slim/fastmcp/client/client.py`	`Client`、`Client.__init__`
惰性导入	`fastmcp_slim/fastmcp/__init__.py`	`__getattr__`
传输推断	`fastmcp_slim/fastmcp/client/transports/inference.py`	`infer_transport`
传输基类	`fastmcp_slim/fastmcp/client/transports/base.py`	`ClientTransport`
内存传输	`fastmcp_slim/fastmcp/client/transports/memory.py`	`FastMCPTransport`
stdio 传输	`fastmcp_slim/fastmcp/client/transports/stdio.py`	`PythonStdioTransport`、`NodeStdioTransport`
HTTP 传输	`fastmcp_slim/fastmcp/client/transports/http.py`	`StreamableHttpTransport`
多服务器配置传输	`fastmcp_slim/fastmcp/client/transports/config.py`	`MCPConfigTransport`
连接/握手	`fastmcp_slim/fastmcp/client/client.py`	`_connect`、`initialize`、`_disconnect`

5.1 Client 是什么​

5.2 核心设计:一个入参,推断传输​

5.3 杀手锏:内存传输 = 零网络测试​

5.4 多服务器:MCPConfig​

5.5 连接生命周期​

代码地图​