1. 为什么我要自己写一个 stdio 版 MCP SDK
MCP（Model Context Protocol）这两年被讨论得很多，但真正动手写一个能跑通的 SDK，很多人会卡在第一步：客户端和服务端到底怎么说话。协议里给了 stdio、SSE、Streamable HTTP 三种传输方式，SSE 正在被 Streamable HTTP 取代，而 stdio 是最适合本地工具接入的一种——它不需要开端口，不需要网络配置，父进程拉起子进程，用标准输入输出交换 JSON 消息就行。
这篇要做的就是：从零实现一个基于标准 IO 传输的 MCP SDK，把服务端和客户端的通信链路在本地跑通，同时把模型调用这一侧接到 TaoToken 的统一 Key 上。TaoToken 是一个聚合多家大模型能力的 API 通道，官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 入口是 https://taotoken.net/api 。你只需要一个 Key，就能在 MCP 客户端里调用不同模型，不用为每个模型单独配一套鉴权。
适合谁看：写过一点 Python、想搞懂 MCP 底层通信原理的开发者；手里有本地工具（文件处理、数据库查询、脚本执行）想接进 AI 助手的同学；以及已经在用 Claude Code、Cursor 这类工具，想自己写 MCP Server 但被 stdio 配置卡住的人。读完你能得到一个可复制的 stdio 传输骨架，包含 settings.json 和 config.toml 两种配置示例，以及一套验证动作。
我试过把服务端和客户端拆成两个文件分别调试，结果消息对不上，后来发现是 stdout 里混进了日志。这个坑后面会专门讲。
2. TaoToken 前置准备：统一 Key 与 API 通道
在写代码之前，先把模型这一侧准备好。MCP 本身只负责工具调用，真正决定「模型要不要调这个工具」的是大模型。所以你需要一个能稳定调用的模型 API，TaoToken 在这里扮演的就是统一入口。
2.1 拿到统一 Key
打开 https://taotoken.net/api-keys ，登录后创建一个 API Key。这个 Key 是你在 MCP 客户端里配置模型时用的凭证，格式通常是一串以特定前缀开头的字符串。创建完先复制保存，页面刷新后不一定还能看到完整值。
注意：Key 不要硬编码进提交到 Git 的代码里，用环境变量或者本地配置文件承载。
2.2 确认 API 通道地址
TaoToken 的 API 基础地址是 https://taotoken.net/api ，兼容 OpenAI 风格的接口路径。也就是说，你在 MCP 客户端里配置模型时，把 base_url 指向这个地址，把 api_key 填成刚才创建的 Key，就能走通。
如果你用的是 Claude Code 这类工具，它有自己的 Anthropic 兼容配置方式，可以参考 https://taotoken.net/doc 里的接入说明。想先在网页上验证 Key 是否可用，可以直接去 https://taotoken.net/console 看额度，或者用模型对话页面 https://taotoken.net/model-chat 发一条消息试试。
2.3 为什么 MCP 场景下要统一 Key
自己写 MCP SDK 的时候，客户端需要把工具列表塞进模型上下文，模型返回工具调用请求，客户端再去请求服务端执行。这一整套流程里，模型调用可能发生很多次。如果每个模型一套 Key、一套地址，配置会非常乱。用 TaoToken 统一 Key 之后，你只需要维护一份凭证，切换模型只改模型名，不改鉴权逻辑。
3. 可复制的 stdio 传输配置骨架
这一章是核心。我会先给服务端的 stdio 处理代码，再给 IO 服务入口和工具集，然后是客户端的子进程通信代码，最后给两种配置文件示例。
3.1 服务端 stdio 处理：读写内存流
stdio 服务的职责很单一：从标准输入读 JSON 行，清洗后写进内存读流；从内存写流拿结果，序列化后写到标准输出。中间用 anyio 的内存对象流做模块间解耦。
import sys import json import anyio from contextlib import asynccontextmanager from io import TextIOWrapper from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream @asynccontextmanager async def stdio_server(): stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8")) stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8")) read_stream_writer, read_stream = anyio.create_memory_object_stream(0) write_stream, write_stream_reader = anyio.create_memory_object_stream(0) async def stdin_reader(): try: async with read_stream_writer: async for line in stdin: line = line.strip() if not line: continue try: message = json.loads(line) except Exception as exc: await read_stream_writer.send(exc) continue await read_stream_writer.send(message) except anyio.ClosedResourceError: await anyio.lowlevel.checkpoint() async def stdout_writer(): try: async with write_stream_reader: async for message in write_stream_reader: await stdout.write(json.dumps(message) + "\n") await stdout.flush() except anyio.ClosedResourceError: await anyio.lowlevel.checkpoint() async with anyio.create_task_group() as tg: tg.start_soon(stdin_reader) tg.start_soon(stdout_writer) yield read_stream, write_stream这里有两个关键点。第一，create_memory_object_stream(0)的缓冲区大小是 0，意味着发送方会阻塞直到接收方取走消息，这在调试时能帮你定位「消息发出去了但没人收」的问题。第二，stdout 只写协议消息，任何日志都必须走 stderr，否则客户端解析 JSON 会失败。
3.2 IO 服务入口：分发到具体工具
IO 服务类负责拿到读写流，然后按 method 字段分发。
import anyio from stdio import stdio_server from tools import add class IOServer: async def run_io(self, read_stream, write_stream) -> None: async for message in read_stream: if isinstance(message, Exception): await write_stream.send({"type": "error", "message": str(message)}) continue if message.get("method") == "add": result = add(message["args"]) await write_stream.send({"type": "result", "result": result}) continue await write_stream.send({"status": "success", "message": "received"}) async def run_stdio_async(self) -> None: async with stdio_server() as (read_stream, write_stream): await self.run_io(read_stream, write_stream) def run(self): anyio.run(self.run_stdio_async)3.3 工具集：一个加法函数起步
工具就是普通函数，后面你可以换成文件读写、HTTP 请求、数据库查询。
def add(args: list[int]) -> int: """一个简单的加法函数，用于测试 stdio 链路""" return args[0] + args[1]3.4 客户端：子进程通信与超时处理
客户端用 asyncio 拉起服务端子进程，通过 stdin 写请求、stdout 读响应，stderr 单独读出来打印。
import asyncio import json import argparse process = None async def communicate_with_server(command): global process process = await asyncio.create_subprocess_exec( *command, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) async def read_stderr(): while True: line = await process.stderr.readline() if not line: break print(f"Server stderr: {line.decode().strip()}") asyncio.create_task(read_stderr()) await asyncio.sleep(1) for i in range(3): await asyncio.sleep(1) message = {"method": "add", "args": [i, i + 1]} print("Sending:", message) process.stdin.write((json.dumps(message) + "\n").encode()) await process.stdin.drain() try: response_line = await asyncio.wait_for(process.stdout.readline(), timeout=5.0) if response_line: print("Received:", json.loads(response_line.decode().strip())) except asyncio.TimeoutError: print("Timeout: no response in 5s") process.stdin.close() await process.wait() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("command", type=str, nargs="+") args = parser.parse_args() asyncio.run(communicate_with_server(args.command))3.5 settings.json 配置示例
如果你用的是支持 MCP 的编辑器或客户端，通常有一份 settings.json 来声明 MCP Server。下面这个骨架可以直接改。
{ "mcpServers": { "local-stdio-demo": { "command": "python", "args": ["/absolute/path/to/test-server.py"], "env": { "TAOTOKEN_API_KEY": "你的统一Key", "TAOTOKEN_BASE_URL": "https://taotoken.net/api" } } } }要点：command 和 args 必须是绝对路径，相对路径在子进程里会找不到文件；env 里把 TaoToken 的 Key 和地址传进去，服务端工具如果需要调模型就能直接读。
3.6 config.toml 配置示例
有些工具用 TOML 配置，结构类似。
[mcp_servers.local_stdio_demo] command = "python" args = ["/absolute/path/to/test-server.py"] [mcp_servers.local_stdio_demo.env] TAOTOKEN_API_KEY = "你的统一Key" TAOTOKEN_BASE_URL = "https://taotoken.net/api"两种配置的语义一致：告诉客户端用什么命令拉起服务端，以及给服务端注入哪些环境变量。区别只是文件格式。
4. 验证请求与成功结果
配置写完，先别急着接模型，把纯 stdio 链路跑通。
第一步，启动客户端并拉起服务端：
python test-client.py python test-server.py第二步，观察输出。正常情况你会看到类似：
Sending: {'method': 'add', 'args': [0, 1]} Received: {'type': 'result', 'result': 1} Sending: {'method': 'add', 'args': [1, 2]} Received: {'type': 'result', 'result': 3} Sending: {'method': 'add', 'args': [2, 3]} Received: {'type': 'result', 'result': 5}第三步，验证模型侧。把 TaoToken 的 Key 配进客户端，让模型决定是否调用 add 工具。你可以先用 https://taotoken.net/model-chat 确认 Key 能正常对话，再回到本地链路。如果模型返回了工具调用请求，客户端转发给服务端，服务端算出结果回传，整条链路就闭环了。
第四步，检查 stderr。服务端如果打印了日志，应该全部出现在Server stderr:前缀后面，而不是混进 stdout 的 JSON 里。
5. 本篇常见错误排查
5.1 JSON 解析失败：stdout 混入日志
最常见的报错是客户端json.loads抛异常，提示Expecting value。原因几乎都是服务端把日志打到了 stdout。解决方式：所有print改成写 stderr，或者用 logging 配置StreamHandler(sys.stderr)。
5.2 子进程启动失败：路径与解释器
报FileNotFoundError或者子进程立刻退出，先检查 command 是不是绝对路径。另一个坑是虚拟环境：客户端用的 python 和服务端需要的依赖不在同一个环境里。建议在配置里写死虚拟环境的 python 路径，比如/Users/you/venv/bin/python。
5.3 消息发出无响应：缓冲区与换行
stdio 协议通常按行分隔消息。如果你写 JSON 时忘了加\n，服务端的async for line in stdin会一直等，永远读不到完整行。另外stdout.flush()不能省，否则消息可能卡在缓冲区里。
5.4 超时设置过短
客户端wait_for设 5 秒，如果服务端首次启动要加载模型或依赖，可能来不及。调试阶段可以放宽到 15 秒，稳定后再收紧。
5.5 Key 无效或额度问题
如果模型侧报 401 或 403，先去 https://taotoken.net/api-keys 确认 Key 没被删，再去 https://taotoken.net/console 看额度。地址要确认是 https://taotoken.net/api ，不要多加或少加路径段。
6. 把链路接到长期编码与 Agent 场景
纯 stdio 的加法 demo 只是起点。真正有价值的场景是：你有一堆本地工具，想让 AI 在写代码、查文档、跑脚本时自动调用。这时候模型调用会变得频繁，按次计费的方式在长期编码场景下不够划算。
如果你打算把 MCP 用在日常编码或者 Agent 工作流里，可以看看 TaoToken 的 Coding Plan：https://taotoken.net/coding-plan 。它面向的就是这种持续调用、多工具编排的场景。配置方式和你现在写的 stdio 骨架兼容，只需要把模型侧的 base_url 和 Key 换成统一通道即可。
接入文档在 https://taotoken.net/doc ，里面有不同客户端的配置示例。Claude Code 用户可以直接参考 https://taotoken.net/claudecode-anthropic 的说明，把 Anthropic 兼容配置指向统一通道。
最后给一个实用建议：先把 stdio 链路用纯本地工具跑通，确认消息收发没问题，再接入模型。顺序反了的话，一旦出错你分不清是传输层的问题还是模型层的问题。我踩过的坑就是先接了模型，结果排查了半天发现是 stdout 日志污染。
