1. 从一次 Agent 工具调用失败说起MCP 协议到底解决什么问题如果你正在做 Agent 开发大概率遇到过这种场景模型能聊天、能写代码但一旦让它「帮我查一下数据库里今天的订单量」或者「调用内部接口发一条通知」就开始胡编乱造参数或者干脆告诉你它做不到。原因不复杂——大模型本身只会生成文本它没有手脚真正干活的是外部工具而模型和工具之间缺一套双方都认的「通话规则」。MCP 协议Model Context Protocol就是干这个的。你可以把它理解成 Agent 世界里的 USB-C 接口以前每个工具都要为每个模型单独写一套适配代码A 模型接数据库写一遍B 模型接同一个数据库再写一遍接十个工具就是十套胶水代码。MCP 把这些适配收敛成一套基于 JSON-RPC 2.0 的标准化通信协议客户端通常是 Agent 宿主和服务器工具集只要都遵守这套协议就能互相识别、互相调用。它适合谁适合正在从零搭 Agent 工具层、想让自己的工具被多个模型复用的开发者也适合想搞明白「模型调用工具时底层到底传了什么字节」的工程师。我试过直接读官方文档概念都懂但真到自己写传输层时还是卡住长度头为什么是 8 字节readexactly到底在等什么异步读写和内核管道缓冲区是什么关系这些问题文档不会细讲。所以这篇不走「概念科普 贴个官方示例」的老路而是带你从零手搓一个能跑起来的 MCP 传输服务器把 JSON-RPC 消息怎么组帧、异步收发怎么不阻塞、以及怎么通过 TaoToken 统一 Key 把模型侧接进来一条链路走通。读完你能拿到一份可复制的配置片段、可运行的客户端/服务端代码以及一次真实的请求响应验证结果。MCP 的核心特征其实就四条使用 JSON-RPC 2.0 格式采用「8 字节十六进制长度头 JSON 正文」的帧结构支持工具调用、资源访问、数据订阅客户端与服务器双向通信。记住这四条后面所有代码都是它们的展开。2. TaoToken 前置准备统一 Key 与 API 通道怎么接在动手写传输服务器之前先把模型侧的通道准备好。因为一个完整的 Agent 链路是「模型 → MCP 客户端 → MCP 服务器 → 工具」我们手搓的是中间那段传输层但模型那头得有个稳定的入口否则你验证的时候会卡在「模型调不通」上误以为是自己的 MCP 代码写错了。TaoToken 在这里扮演的角色是统一 Key 的模型接入通道。它的价值在于你不需要为每个模型单独维护一套鉴权和 Base URL一个 Key 就能在多个模型之间切换这对调试 Agent 特别有用——今天用 A 模型测工具调用明天换 B 模型对比参数生成质量改一个 Model ID 就行不用动传输层代码。先拿 Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台里创建 API Key。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 列表在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。创建完记得立刻复制页面刷新后就看不全了这是很多人第一次踩的坑。拿到 Key 之后你要记住三个东西后面配置里反复用配置项值说明Base URLhttps://taotoken.net/api注意这里不加 UTM 参数是纯 API 端点API Keysk-开头的一串从 api-keys 页面复制Model ID例如claude-sonnet-4-5等在模型对话页可查当前可用模型想先确认 Key 能用最直接的方式是去模型对话页发一句话https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。能正常回复说明 Key 和通道没问题再往下写代码。如果你打算长期跑 Agent 任务、频繁调用工具可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合持续性的编码和 Agent 场景比按次调用更划算。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数不确定时以文档为准。这里要强调一点TaoToken 是合规的模型 API 接入通道不是所谓「中转」。你通过它调用的是标准 API代码里就是普通的 HTTP 请求没有任何特殊处理。把 Key 配好我们就进入正题——手搓传输服务器。3. 可复制配置MCP 服务器配置片段与 JSON-RPC 消息样例这一节给你可以直接抄的配置和消息样例。先看 MCP 服务器的标准配置片段。不同宿主Claude Code、Cline、Codex 等配置文件位置不同但核心字段就三个Base URL、Key、Model ID。以常见的settings.json形式为例{ mcpServers: { math-tools: { command: python, args: [math_server.py], env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-你的Key, TAOTOKEN_MODEL_ID: claude-sonnet-4-5 } } } }如果你用的是 TOML 风格的配置比如某些 Codex 场景等价写法是[mcp_servers.math-tools] command python args [math_server.py] [mcp_servers.math-tools.env] TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY sk-你的Key TAOTOKEN_MODEL_ID claude-sonnet-4-5注意三件套必须齐全Base URL 指向https://taotoken.net/apiKey 用你创建的那串Model ID 填当前可用的模型。少任何一个客户端在初始化握手时就会失败。再看 JSON-RPC 消息样例。MCP 的每一次工具调用本质就是一条 JSON-RPC 2.0 请求。客户端发出去的请求长这样{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: add, arguments: {a: 3, b: 5} } }服务器回包{ jsonrpc: 2.0, id: 1, result: 8 }关键点在于id必须原样带回客户端靠它把响应和请求配对。method是tools/call表示工具调用params.name是工具名params.arguments是参数对象。但这两段 JSON 不是直接扔进管道的。MCP 在传输层加了一层帧结构先写 8 字节十六进制长度头再写 JSON 正文。上面那条请求的正文是 116 字节转成十六进制就是00000074所以实际在管道里流动的字节流是00000074{jsonrpc:2.0,id:1,method:tools/call,params:{name:add,arguments:{a:3,b:5}}}前 8 字节是长度头后面紧跟正文中间没有任何换行或分隔符。接收方先读 8 字节解析出长度再按这个长度读正文一个字节不多一个字节不少。这就是 MCP 传输层的全部秘密。为什么用十六进制而不是十进制因为固定 8 位十六进制能表示 4GB 长度且解析时int(raw_len, 16)一行搞定不用处理变长数字的对齐问题。为什么不用换行分隔因为 JSON 正文里可能包含换行用换行做分隔符会误判边界长度头是更稳妥的方案。把配置和消息格式定下来接下来就是让它们真正跑起来。4. 手搓异步传输服务器从管道到 JSON-RPC 收发验证现在写代码。整个系统分两个文件client.py模拟 Agent 宿主和math_server.pyMCP 服务器提供 add/multiply 两个工具。先看服务端。服务端第一件事是建工具注册表用一个装饰器把函数注册进字典#!/usr/bin/env python3 import asyncio import json import sys TOOLS {} def tool(name): def decorator(func): TOOLS[name] func return func return decorator tool(add) def add(a: int, b: int) - int: return a b tool(multiply) def multiply(a: int, b: int) - int: return a * b然后是帧级读写函数这是传输层的核心async def read_frame(reader: asyncio.StreamReader) - dict: raw_len await reader.readexactly(8) json_len int(raw_len, 16) raw_body await reader.readexactly(json_len) return json.loads(raw_body) async def write_frame(writer: asyncio.StreamWriter, obj: dict): body json.dumps(obj, separators(,, :)).encode() header f{len(body):08x}.encode() writer.write(header body) await writer.drain()readexactly(8)的语义是「必须读满 8 字节不够就等」。它靠内核管道缓冲区的「已写入字节数 − 已读取字节数 ≥ 8」这个纯算术判断不会多读也不会读到假字节。f{len(body):08x}把长度格式化成 8 位小写十六进制不足左补零。主循环把 stdin/stdout 包装成异步 reader/writer然后持续读帧、处理、回写async def main(): reader asyncio.StreamReader() reader_protocol asyncio.StreamReaderProtocol(reader) await asyncio.get_event_loop().connect_read_pipe( lambda: reader_protocol, sys.stdin.buffer ) writer_transport, writer_protocol await asyncio.get_event_loop().connect_write_pipe( asyncio.streams.FlowControlMixin, sys.stdout.buffer ) writer asyncio.StreamWriter( writer_transport, writer_protocol, reader, asyncio.get_event_loop() ) while True: try: request await read_frame(reader) except asyncio.IncompleteReadError: break if request.get(method) ! tools/call: continue tool_name request[params][name] args request[params][arguments] tool_id request[id] if tool_name in TOOLS: result TOOLS[tool_name](**args) else: result {error: ftool {tool_name} not found} response {jsonrpc: 2.0, id: tool_id, result: result} await write_frame(writer, response) writer.close() await writer.wait_closed() if __name__ __main__: asyncio.run(main())这里有两个容易混淆的「缓冲区」内核级 pipe 缓冲区在内核空间解决读写速度差用户级StreamReader缓冲区在你的进程空间暂存已读出但还没被业务消费的字节。connect_read_pipe把 fd 0 注册到事件循环内核一有数据就触发协议搬进 reader业务代码从 reader 里舀减少系统调用次数。客户端这边启动子进程拿到 stdin/stdout 句柄构造请求发出去#!/usr/bin/env python3 import asyncio import json import sys async def main(): proc await asyncio.create_subprocess_exec( sys.executable, math_server.py, stdinasyncio.subprocess.PIPE, stdoutasyncio.subprocess.PIPE, ) request { jsonrpc: 2.0, id: 1, method: tools/call, params: {name: add, arguments: {a: 3, b: 5}}, } body json.dumps(request, separators(,, :)).encode() header f{len(body):08x}.encode() proc.stdin.write(header body) await proc.stdin.drain() raw_len await proc.stdout.readexactly(8) json_len int(raw_len, 16) raw_body await proc.stdout.readexactly(json_len) response json.loads(raw_body) print(Server raw response:, raw_body.decode()) print(Result , response[result]) proc.stdin.close() await proc.wait() if __name__ __main__: asyncio.run(main())create_subprocess_exec会复制一份子进程用math_server.py替换子进程内容stdinPIPE告诉内核给子进程建一条匿名管道当它的 stdin。父进程保留写端封装成proc.stdin子进程的 fd 0 指向读端于是父进程写、子进程读方向不能反。跑起来的结果服务器启动等待请求... 收到请求: {jsonrpc: 2.0, id: 1, method: tools/call, params: {name: add, arguments: {a: 3, b: 5}}} 执行加法: 3 5 8 发送响应: {jsonrpc: 2.0, id: 1, result: 8} Server raw response: {jsonrpc:2.0,id:1,result:8} Result 8 客户端断开连接到这里一次完整的 MCP 传输层请求响应就验证通过了。模型侧通过 TaoToken 统一 Key 接入传输层自己手搓整条链路你都能控制。5. 本篇常见报错排查401、local proxy failed、reading choices 与 OAuth代码跑通不代表一路顺风下面这几个报错是我和读者反馈里出现频率最高的对照着查。401 Unauthorized。这个几乎都是 Key 的问题。先确认TAOTOKEN_API_KEY是不是从 api-keys 页面完整复制的有没有多余空格。再确认 Base URL 是不是https://taotoken.net/api注意这里不要带 UTM 参数带参数的 URL 是给浏览器用的API 端点要干净。如果 Key 刚创建等几秒再试偶尔有缓存延迟。local proxy failed。这个报错通常出现在客户端初始化阶段意思是本地连接建立失败。检查你的 MCP 服务器进程是不是真的启动了——手动python math_server.py跑一下看有没有语法错误或依赖缺失。如果服务器秒退客户端自然连不上。另外确认command和args路径写对了相对路径是相对于宿主工作目录不是相对于配置文件。reading choices 相关报错。这类错误一般出现在模型返回体解析阶段说明请求发出去了但响应格式不符合预期。常见原因是 Model ID 填错了或者该模型不支持工具调用。去模型对话页确认当前可用模型列表换一个支持 function calling 的模型再试。如果用的是 Coding Plan确认套餐覆盖了你选的模型。OAuth 相关报错。如果你在 Claude Code 或类似宿主里看到 OAuth 报错通常是宿主的登录态和 MCP 配置冲突了。MCP 服务器走的是独立进程通信不依赖宿主的 OAuth 登录。检查配置里是不是误加了 OAuth 相关字段MCP 的env里只需要 Base URL、Key、Model ID 三件套多余的鉴权字段反而会干扰。JSONDecodeError。这个是自己代码的问题多半是readexactly读的字节数和实际正文长度对不上。检查发送端f{len(body):08x}里的body是不是和实际写入的正文一致有没有在json.dumps之后又改了内容。接收端int(raw_len, 16)解析出来的长度必须和正文严格相等少一字节就报这个错。僵尸进程堆积。如果忘了await proc.wait()子进程结束后会变成 Z 状态占着进程表。用ps aux | grep python能看到defunct标记。养成习惯proc.stdin.close()之后一定跟await proc.wait()。排查顺序建议先确认 Key 和 Base URL401 类再确认服务器进程能独立启动local proxy 类最后看模型和响应格式reading choices 类。大部分问题出在前两步。6. 把这条链路用起来从验证到长期 Agent 任务走到这里你已经有了一个能跑的 MCP 传输服务器也知道了模型侧怎么通过统一 Key 接入。接下来怎么用取决于你的场景。如果只是想验证模型能不能正确生成工具调用参数去模型对话页多发几轮观察它生成的arguments是否符合你的工具签名https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。这一步能帮你快速判断是模型能力问题还是传输层问题。如果要把这套东西接进真实的 Agent 工作流比如让模型持续调用多个工具完成编码任务那 Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。它的调用模型更适合长任务不用每次担心额度。接入过程中遇到参数问题文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。API 端点统一是 https://taotoken.net/api 。最后说个实用技巧调试 MCP 传输层时在write_frame里加一行print(f[SEND] {headerbody}, filesys.stderr)把帧内容打到 stderr 而不是 stdout。因为 stdout 是协议通道往里面打日志会污染帧结构导致客户端解析失败。这个坑我踩过排查了半天才发现是日志打错了地方。stderr 不参与协议通信随便打。