前言
在前面的文章中，我们写了ToolRegistry方法。现在再来读MCP SDK，不再是黑盒调用。按理说，这篇沿着call_tool追踪完整链路。回答一个主要问题：SDK比手写多做了什么？
一、客户端：请求的一生
call_tool
→ CallToolRequest // 构造请求对象
→ model_dump // 序列化成 JSON dict
→ JSONRPCDispatcher.send_raw_request
├─ 分配 request_id
├─ _pending = 内存流等待者 // 关键设计
├─ JSONRPCRequest 写入传输层
└─ receive.receive 挂起等响应
→ 响应按 id 反查 _pending → 唤醒 → CallToolResult
_pending关联表是客户端最值得学的设计。
在并发场景下同时发N个请求。响应乱序返回——没有这张表，无法把每个响应送回正确的等待者。_pending是“request_id → 等待者”的Map，响应到达时按JSON‑RPC响应的id字段反查。
前端类比：同时发多个fetch。每个都有自己的Promise在await——谁先返回谁先resolve，互不覆盖。_pending就是那个
在前面文章中我的串行Agent用不到这个机制——但这就是MCP为生产并发预留的设计。
二、服务端：两层路由
收到 JSON-RPC 请求
→ 第一层路由：method 字段
→ _request_handlers
→ 第二层路由：params.name
→ ToolManager._tools
→ Tool.run
-
两层路由的分离是协议设计的关键：
-
第一层按 method 路由——协议层的方法名
-
第二层按 name 路由——业务层的工具名
\_request_handlers: {method: handler} 表；怎么说呢，\_tools: {name: Tool} 表。这跟之前手写的TolRegistry完全同构——只是多了一层协议路由，让SDK能同时支持 tools/resources/prompts 三类能力。
三、Tool.run：校验、执行、转换
async def run:
# 校验
validate_arguments # jsonschema 校验
# 执行
if fn_is_async:
result = await fn # 异步函数直接 await
else这方面，result = anyio.to_thread.run_sync # 同步函数丢线程池
# 转换
if convert_result:
result = self.fn_metadata.convert_result # → CallToolResult
return result
-
synchronous functions do not block event loop.
-
anyio.to_thread.run_sync = Promise.resolve + Web Worker — 把阻塞操作移出主线程。
-
The previous implementation used inspect.iscoroutinefunction;MCP uses fn_is_async flag—functionally equivalent.
四、错误分层的智慧
except MCPError:
raise # 协议级错误 → 原样抛 → JSON-RPC error 响应
except Exception as e:
raise ToolError # 执行失败 → isError=True
-->
-
MCPError → 协议错误 → JSON‑RPC error 响应：“这次请求本身有问题”。
-
ToolError → 执行失败 → CallToolResult： “工具跑崩了”。这是导致使用者痛点的关键区别。
-
This distinction tells caller wher to retry:
-
If protocol error – retry useless .
-
If execution failure – retry may help .
-
The hand‑written version collapsed all exceptions into {"success": false}。losing this decision factor.
-
MCP’s layered error handling gives callers actionable information.
痛点对照的观点是，
| 痛点/需求 |
手写实现时遇到的问题 |
MCP SDK 如何解决 |
| # 并发冲突 |
需要自行管理 requestid 与 Promise 的映射；易出错
导致某些工具调用结果被覆盖或丢失。 |
|
| # 阻塞同步工具 |
同步函数会占用事件循环，导致整个 Agent 卡顿。其实， |
|
| # 错误信息缺失 |
所有异常统一包装成{"success":false}。
缺少重试判断依据， |
|
五、MCP vs 手写：替换成本与设计取舍
| 工具表定义方式 | dct = {schema。fn},TolManager.tools :{name : Tool}'
| TolManager.tools :{name : Tool}'
|
| 请求匹配策略 | 串行，无关联表<_pending>并发安全错误处理统一 {'success':False}
| 两层路由+ requesthandlers + tools + MCPError/ToolError 分层 sync/async
inspect.iscoroutinefunction ↔ fnisasync + anyio.tothread.runsync
结果转换直接返回 dict convertresult -> CallToolResult传输直接函数调用stdio/SSE/HTTP 可插拔
|
|
|
MCP 的生产级设计包含：
-
"并发关联表 " 能确保无论多少请求并行，都能把每个响应送回对应等待者，从而消除“随机丢失结果”的痛点。
-
"错误分层": 提供“该不该重试”的关键信息。避免因为误判而无效重试或遗漏必要重试，从而提高程序鲁棒性。"传输抽象": 可插拔。让你无需改动业务代码即可切换不同网络环境或部署模式，为运维和弹性
提供灵活性。"同步/异步透明化": 用任何异步框架时无需担心阻塞；任何同步函数都安全投进线程池。不会卡住事件循环，提高开发效率和程序吞吐量。其实，"功能一致性与可
性": 与手写实现同构。但多了一层协议路由，使得SDK可以一次支持工具调用、资源读取和提示生成等三类能力，无需额外编码。**如果你正面临以下痛点**：
-
并发调用混乱导致结果错位；
-
同步工具卡住整个服务；
-
错误信息模糊无法判断是否重试；
-
想快速切换不同传输方式；
使用MCP SDK，你只需：
python
@mcp.tool
def search_catalog: ...
result = await agent.call_tool
print
通过上述几行代码。你即可获得：
-
可靠的并发；
-
非阻塞执行；
-
细粒度错误反馈；
-
灵活部署。
©2026 MCP团队 • 这篇文章采用 MIT 开源许可证。<\/footer>
