数据截至 (上游 commit 743ddffd0a07)
第 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 |