1. 为什么我要自己写一个 stdio 版 MCP SDKMCPModel Context Protocol这两年被讨论得很多但真正动手写一个能跑通的 SDK很多人会卡在第一步客户端和服务端到底怎么说话。协议里给了 stdio、SSE、Streamable HTTP 三种传输方式SSE 正在被 Streamable HTTP 取代而 stdio 是最适合本地工具接入的一种——它不需要开端口不需要网络配置父进程拉起子进程用标准输入输出交换 JSON 消息就行。这篇要做的就是从零实现一个基于标准 IO 传输的 MCP SDK把服务端和客户端的通信链路在本地跑通同时把模型调用这一侧接到 TaoToken 的统一 Key 上。TaoToken 是一个聚合多家大模型能力的 API 通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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 本身只负责工具调用真正决定「模型要不要调这个工具」的是大模型。所以你需要一个能稳定调用的模型 APITaoToken 在这里扮演的就是统一入口。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, encodingutf-8)) stdout anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encodingutf-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, stdinasyncio.subprocess.PIPE, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, ) async def read_stderr(): while True: line await process.stderr.readline() if not line: break print(fServer 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(), timeout5.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, typestr, 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 Planhttps://taotoken.net/coding-plan 。它面向的就是这种持续调用、多工具编排的场景。配置方式和你现在写的 stdio 骨架兼容只需要把模型侧的 base_url 和 Key 换成统一通道即可。接入文档在 https://taotoken.net/doc 里面有不同客户端的配置示例。Claude Code 用户可以直接参考 https://taotoken.net/claudecode-anthropic 的说明把 Anthropic 兼容配置指向统一通道。最后给一个实用建议先把 stdio 链路用纯本地工具跑通确认消息收发没问题再接入模型。顺序反了的话一旦出错你分不清是传输层的问题还是模型层的问题。我踩过的坑就是先接了模型结果排查了半天发现是 stdout 日志污染。
从0到1实现一个基于标准IO传输的MCP SDK：TaoToken统一Key接入与stdio配置实战
1. 为什么我要自己写一个 stdio 版 MCP SDKMCPModel Context Protocol这两年被讨论得很多但真正动手写一个能跑通的 SDK很多人会卡在第一步客户端和服务端到底怎么说话。协议里给了 stdio、SSE、Streamable HTTP 三种传输方式SSE 正在被 Streamable HTTP 取代而 stdio 是最适合本地工具接入的一种——它不需要开端口不需要网络配置父进程拉起子进程用标准输入输出交换 JSON 消息就行。这篇要做的就是从零实现一个基于标准 IO 传输的 MCP SDK把服务端和客户端的通信链路在本地跑通同时把模型调用这一侧接到 TaoToken 的统一 Key 上。TaoToken 是一个聚合多家大模型能力的 API 通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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 本身只负责工具调用真正决定「模型要不要调这个工具」的是大模型。所以你需要一个能稳定调用的模型 APITaoToken 在这里扮演的就是统一入口。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, encodingutf-8)) stdout anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encodingutf-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, stdinasyncio.subprocess.PIPE, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, ) async def read_stderr(): while True: line await process.stderr.readline() if not line: break print(fServer 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(), timeout5.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, typestr, 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 Planhttps://taotoken.net/coding-plan 。它面向的就是这种持续调用、多工具编排的场景。配置方式和你现在写的 stdio 骨架兼容只需要把模型侧的 base_url 和 Key 换成统一通道即可。接入文档在 https://taotoken.net/doc 里面有不同客户端的配置示例。Claude Code 用户可以直接参考 https://taotoken.net/claudecode-anthropic 的说明把 Anthropic 兼容配置指向统一通道。最后给一个实用建议先把 stdio 链路用纯本地工具跑通确认消息收发没问题再接入模型。顺序反了的话一旦出错你分不清是传输层的问题还是模型层的问题。我踩过的坑就是先接了模型结果排查了半天发现是 stdout 日志污染。