MCP Client：Agent侧的协议客户端开发

可视化演示

角色定位

老板第一次把总机摆在桌上时，秘书处特意嘱咐：你只需要对总机说话，剩下的事不用管。这台总机的全部意义，就是充当老板与外部机构之间的“翻译和管道”。老板不必知道气象局的具体分机号，也不必纠结电话线怎么走，他只要对着总机说一句“我要北京明天的天气”，剩下的环节全部由这台机器负责衔接。

在 MCP 协议中，这个“管道”被具体化为 Client 对象。它与运行中的 MCP Server 一一对应，是协议通信的真正发起方。你可以把 MCP Client 理解为一个协议会话的代名词：它不负责思考“为什么调用”，也不负责决定“调用哪个工具”，它只负责让一次远程调用安全、准确、可追溯地完成。

MCP Client 的核心职责可以收敛为三件事：第一，能力发现，向 Server 发起 initialize 和 tools/list，弄清对方提供什么工具、每个工具接受什么参数；第二，工具调用，把 Agent 的意图封装成 tools/call 请求发出去，并检验返回结果；第三，状态维护，记录这个连接里有哪些工具、最近调用了什么、结果是否出错，这些状态是后续决策的上下文来源。

你会发现，接线员的工作从来不是替老板做决策——他负责查号码、拨电话、记本子。MCP Client 也是如此：它被设计得足够“笨”，只做协议层的事，把聪明留给 Host 和 Agent。

生命周期

总机员每天上班有一套固定流程：先检查线路，报出自己的工号，和对方确认合作范围，确认无误后才会开始接活；下班时再挂断电话、合上记录本。这套“上班—上岗—干活—下班”的流程，在 MCP 协议里被称为 Client 与 Server 的连接生命周期。

MCP 协议中，Client 与 Server 的每次连接都遵循一套严格的状态序列。为什么不能省掉握手直接调用？就像你不能电话一接通就开始念订单而不先报身份一样，Server 需要知道“你是谁、协议版本是否兼容、你能处理哪些能力”，Client 也需要知道“你提供哪些能力、工具长什么样”。

技术细节上，初始化阶段 Client 发送 initialize 请求，带上协议版本、客户端能力和客户端信息；Server 回复协议版本、服务端能力和服务器信息。随后 Client 发送 notifications/initialized 通知，双方进入 ready 状态。运行期可以收发 tools/list、tools/call 等消息。最后是关闭阶段，Host 决定不再使用该连接，Client 清理本地会话状态和未完成请求。

就像下班前必须挂断电话、合上记录本，一次不完整的关闭会留下占着的线路。在 MCP 开发里，忘记关闭 Client 连接是最常见的资源泄漏根源之一，尤其在长驻服务中，每个未关闭的会话都会持续占用内存和网络句柄。

多Server管理

桌上的电话不止一部，气象局、交易所、数据库各占一条专线。接线员心里有本通讯录，哪条线对应哪个机构，绝不能串线——不能把天气查询拨到交易所去。更关键的是，每条专线的通话记录是分开记的，不能混在一起。

一个 Agent 要提供有用能力，几乎总要连接多个 Server：一个做联网搜索、一个访问公司数据库、一个调用内部业务 API。有些初学者以为“连接一个 Server 就够了”，但真实场景中，一个工具调用动辄需要三四个外部服务协同。

工程上的标准做法是：每个 Server 连接都使用独立的 Client 会话，状态绝不混用。再用一个会话注册表管理这些会话，key 是 server 标识，value 是会话对象。调用前先查表拿到正确的会话，连接不可用时移除条目并允许重建。这样既保证了路由正确，也隔离了故障——某个 Server 挂了不会拖垮其他连接。

接线员最怕串线，MCP Client 管理多 Server 时最怕会话串扰。如果你在代码里让两个 Server 共用同一个 Client 对象，就会发现消息 ID 冲突、工具清单互相覆盖——这就像把两家机构的电话打到彼此的记录本上，场面会彻底失控。

工具调用

老板说“我要北京天气”，秘书处写一张转办单，总机员把它翻译成标准格式报给对方，对方回复后再原样带回来。这里有一个细节：转办单上必须有一个编号，机构的回复上也要带同一个编号，总机员才能确认这是那张单子的回执。

MCP 协议里的“标准格式”就是 JSON-RPC 2.0 消息。一次完整的工具调用分四步：工具发现、参数构建、发送请求、校验响应。Client 先通过 tools/list 拿到工具清单，包括工具名称和参数 Schema；Agent 侧基于 Schema 补全参数后，Client 构造 tools/call 请求，包含唯一 id、方法名和 params；Server 返回带同一 id 的响应。

Client 在收到响应时要做的第一件事不是解析业务数据，而是校验 id 是否与某个未完成请求匹配。这个匹配机制决定了并发能力：多个工具调用可以同时在空中飞行，谁回来就发给谁。响应还带 isError 标记，即使是业务错误也会以正常消息返回，Client 只需如实上报给 Host。

总机员拿着编号去对账，Client 拿着 id 去对消息。如果你在封装 SDK 时忽略 id 配对，就会出现响应错乱、超时误判——就像把张三的订单回执交给了李四，客户体验会瞬间崩塌。

错误重试

也有拨不通的时候：占线、无人接、说到一半断线。老练的接线员从不立刻重拨，也不无限重拨。他会先看是哪种失败——是号码不存在（协议错），还是对方忙（资源紧张），还是对方接了但信号断了（连接中断），然后按场景决定要不要重试。

MCP Client 同样要把错误分成三类：连接类错误（握手失败、连接中断）、协议类错误（非法 JSON-RPC、超时）、业务类错误（工具执行失败返回 error）。处理手段包括：给建连和读写设置超时时间；失败后只重试有限次数；每次重试等待时间按指数规律增长，避免把服务打爆。

更 subtle 的一点是幂等。像“转账”“创建订单”这类有副作用的操作，重放时必须保证只执行一次。办法是复用请求 id 或业务 token，Server 端识别到同一 id 就返回上次结果，而不是再执行一遍。MCP 的 JSON-RPC 结构中，请求 id 天然适合承担这个角色。

重拨的智慧在于：等得够久，又不至于让机构被电话打爆。没有退避的重试是“雪崩发生器”——一旦 Server 出现过载，所有 Client 同时立刻重试，瞬间把 Server 彻底压垮；有了指数退避，大家自动错开节奏，Server 才有机会缓过来。

与Host协作

秘书处和总机员各司其职：前者决定找谁、问什么，后者负责打通线路、原样传话。秘书处不会冲到机房里去拨号，总机员也绝不会替老板拍板“这事不用问了”——一旦越权，整个办公秩序就乱了。

在 MCP 架构里，Host 与 Client 的关系正是这种分工。Host 是 Agent 应用的核心容器，掌握着任务编排、模型调度和用户交互；Client 只是 Host 手下一个“外卖骑手”，负责把任务送到 Server 再取回来。两者通过回调协作，职责边界非常清晰。

技术实现上，Host 负责高层编排：决定调用哪个工具、什么时候创建新连接、收到结果后下一步做什么；Client 负责协议细节：initialize 握手、消息编码解码、重试策略、id 配对，并通过回调把结果和通知交还给 Host。Host 依赖 Client 的稳定传输，Client 不感知 Host 的业务意图。

总机如果擅自替老板回答或者替秘书处做决定，那就越权了。MCP 里的 Client 如果试图介入业务编排，就会让职责混乱、协议层与应用层耦合，最终难以测试和维护。记住这句话：Host 是大脑，Client 是手脚。

SDK对比

市面上的总机品牌众多，功能都能打通电话，但操作台布局、配套培训、扩容方案各不相同。有的总机按键少、易上手，适合初创团队；有的总机扩展槽多、支持并发线路，适合大企业；选型从来不是“哪个最好”，而是“哪个最适合你的团队”。

MCP 生态同样如此，官方 SDK 已经覆盖 Python、TypeScript、Java/Kotlin、Ruby、C# 等主流语言，还有社区维护的 Go、Rust 等实现。它们遵循同一份协议规范，所以 Server 侧互操作没有问题，差异集中在开发体验和运行特性上。

选型时主要看三个维度：传输层支持（是否同时支持 stdio、HTTP+SSE、Streamable HTTP）；异步模型（Python 的 asyncio、TypeScript 的回调与响应式流、Java 的阻塞与响应式）；能力完善度（工具发现、自动重连、对 notifications 和采样的支持）。AI Agent 原型常用 Python SDK，Web 集成偏爱 TypeScript，企业级系统倾向 Java。

选总机不是越贵越好，而是看秘书处的人会不会用、设备稳不稳。选 MCP SDK 也不是越流行越好：如果你的 Agent 跑在 Node 服务里，硬塞一个 Python SDK 就非常别扭；如果你的团队全是 Java 工程师，为了快速原型引入不熟悉的语言栈，后续维护成本会远高于节省的那点开发时间。

AI 应用场景

Python 演示

import time  class MCPSession:    def __init__(self, server_id, endpoint):        self.server_id = server_id        self.endpoint = endpoint        self.connected = False        self.tools = {}     def initialize(self, server_meta):        """握手：登记工具表，进入就绪状态。"""        if not isinstance(server_meta, dict) or "tools" not in server_meta:            raise ValueError("invalid server metadata")        self.tools = server_meta["tools"]        self.connected = True        return {"status": "ok", "tools": sorted(self.tools)}     def call_tool(self, name, args):        """执行工具调用，失败时抛出异常。"""        if not self.connected:            raise ConnectionError("session not initialized")        if name not in self.tools:            raise KeyError("tool not found: " + name)        return self.tools[name](args)     def close(self):        """优雅关闭：清空会话状态。"""        self.connected = False        self.tools = {}  class MCPClientManager:    def __init__(self):        self.sessions = {}     def add_server(self, server_id, endpoint, server_meta):        """新增一个 Server 连接，并完成初始化握手。"""        session = MCPSession(server_id, endpoint)        result = session.initialize(server_meta)        self.sessions[server_id] = session        return result     def active_sessions(self):        """返回当前仍处于连接状态的会话列表。"""        return [sid for sid, s in self.sessions.items() if s.connected]     def call_with_retry(self, server_id, tool, args, max_retries=3, base_delay=0.2, max_delay=2.0):        """带指数退避的工具调用。"""        session = self.sessions.get(server_id)        if session is None:            raise KeyError("no such server: " + server_id)        last_exc = None        for attempt in range(max_retries + 1):            try:                return session.call_tool(tool, args)            except Exception as exc:                last_exc = exc                if attempt == max_retries:                    break                delay = min(base_delay * (2 ** attempt), max_delay)                time.sleep(delay)        raise last_exc     def close(self):        """关闭所有会话。"""        for session in self.sessions.values():            session.close()        self.sessions.clear()  def flaky_tool(args):    raise RuntimeError("service busy")  def main():    calc_meta = {        "tools": {            "add": lambda args: args["a"] + args["b"],            "mul": lambda args: args["a"] * args["b"],            "flaky": flaky_tool,        }    }    weather_meta = {        "tools": {            "query": lambda args: {"city": args["city"], "temp": 22},        }    }     manager = MCPClientManager()    print(manager.add_server("calc", "stdio://calc", calc_meta))    print(manager.add_server("weather", "http://weather", weather_meta))     print(manager.call_with_retry("calc", "add", {"a": 3, "b": 4}))    print(manager.call_with_retry("weather", "query", {"city": "Beijing"}))     try:        manager.call_with_retry("calc", "flaky", {}, max_retries=1, base_delay=0.01)    except RuntimeError as exc:        print("retry failed:", exc)     print("active:", manager.active_sessions())    manager.close()  if __name__ == "__main__":    main()

1. session.initialize(server_meta)模拟 MCP 初始化握手：登记工具表，把 connected 置为 True，对应现实中的 initialize 请求。
2. self.sessions[server_id] = session用字典作为会话注册表，这就是文章里“多 Server 管理”的落点。
3. session.call_tool(tool, args)根据名字动态分发到具体工具函数，对应 JSON-RPC 的 tools/call。
4. delay = min(base_delay * (2 ** attempt), max_delay)指数退避公式的代码实现，等待时间按翻倍增长并封顶。
5. manager.call_with_retry("calc", "add", {"a": 3, "b": 4})对外暴露的容错调用入口，失败会自动重试。
6. manager.close()遍历所有会话执行优雅关闭，对应生命周期里的 close 阶段。

常见误区

面试题