手写一个 MCP Server：从 JSON-RPC 到 Streamable HTTP 的底层全解析（TaoToken 统一 Key 接入版）

1. 为什么我要手写一个 MCP ServerMCP Server 说白了就是给大模型装手的进程模型想查数据库、调内部 API、读本地文件都通过它暴露的 Tools 和 Resources 完成。现在很多人用npx mcp-server-xxx一把梭本地跑得挺欢一旦要排查线上问题、或者把公司内部系统封装成 MCP Server就抓瞎了——因为不知道 JSON-RPC 消息长什么样、stdio 和 Streamable HTTP 到底差在哪、握手失败该看哪一行日志。这篇就干一件事不依赖任何 MCP 框架用标准库把协议跑通。你会看到 JSON-RPC 2.0 的四个字段怎么在 stdio 和 HTTP 两条传输层上流动然后手写一个能用的 MCP Server 和一个最小 Client最后用 curl 验证 Streamable HTTP 会话。适合需要自建 MCP Server 并接入 AI 工具的开发者尤其是想把内部系统安全暴露给 Agent 的那批人。我试过直接照官方 SDK 抄结果被日志污染协议流坑了一下午所以下面会把踩过的坑单独拎出来讲。2. TaoToken 统一 Key 的前置准备自建 MCP Server 之后你大概率要把它接到某个 HostCursor、Claude Code、自研 Agent上而 Host 侧调用模型需要 Key。如果每个工具、每个环境各配一套 Key管理成本会爆炸。TaoToken 的思路是统一入口一个 Key 覆盖模型对话、Coding Plan、API 调用MCP Server 侧只需要在配置里引用同一个环境变量即可。你需要先拿到 Key入口在这里控制台创建/管理 Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite接入文档协议与端点说明https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewriteAPI 基地址统一用https://taotoken.net/api注意这个地址不带 UTM 参数写进代码里就用它。Key 建议放环境变量别硬编码export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意MCP Server 本身不直接调模型它只暴露工具真正调模型的是 Host。所以 Key 配在 Host 侧Server 侧只在需要回调模型比如工具内部做二次推理时才用得上。这个边界先分清后面配置才不会乱。3. 可复制配置config.toml 与 settings.json 骨架不同 Host 的配置文件格式不一样这里给两份最常用的骨架。核心都是三件事启动命令、环境变量、传输方式。先看config.toml适合自研 Host 或支持 TOML 的工具[mcp] # 传输方式stdio 或 streamable-http transport stdio [mcp.server.minimal] command python3 args [/opt/mcp/minimal_mcp_server.py] env { TAOTOKEN_API_KEY ${TAOTOKEN_API_KEY}, TAOTOKEN_BASE_URL https://taotoken.net/api } [mcp.server.remote] transport streamable-http url http://127.0.0.1:8765/mcp headers { Authorization Bearer ${TAOTOKEN_API_KEY} }再看settings.jsonClaude Code / Cursor 这类 Host 常用{ mcpServers: { minimal: { command: python3, args: [/opt/mcp/minimal_mcp_server.py], env: { TAOTOKEN_API_KEY: sk-你的key, TAOTOKEN_BASE_URL: https://taotoken.net/api } }, remote: { type: streamable-http, url: http://127.0.0.1:8765/mcp, headers: { Authorization: Bearer sk-你的key } } } }参数对照表方便你按需改字段作用stdio 必填HTTP 必填command启动 Server 的可执行文件是否args启动参数数组是否env注入进程的环境变量是否urlStreamable HTTP 端点否是headers鉴权/会话头否是transport/type传输类型标识是是提示env里引用${TAOTOKEN_API_KEY}是否生效取决于 Host 是否支持变量展开。不确定就直接写值但别把带 Key 的配置文件提交到 Git。4. 手写 MCP Server从 JSON-RPC 到 stdio协议层所有消息都是 JSON-RPC 2.0结构永远是jsonrpc / id / method / params四件套。握手消息长这样{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-06-18, capabilities: {tools: {listChanged: true}}, clientInfo: {name: my-host, version: 1.0.0} } }下面是不依赖任何框架的 stdio 版 Server核心就三点stdout 发消息、stderr 打日志、按 method 分发。#!/usr/bin/env python3 minimal_mcp_server.py — 纯标准库实现的 MCP Serverstdio 传输 import json import sys from typing import Any def send(msg: dict) - None: MCP 走 stdout每行一个 JSON 对象 sys.stdout.write(json.dumps(msg, ensure_asciiFalse) \n) sys.stdout.flush() def log(msg: str) - None: 调试日志必须走 stderr不能污染协议流 sys.stderr.write(f[server] {msg}\n) TOOLS [ { name: add, description: 计算两个整数之和, inputSchema: { type: object, properties: {a: {type: integer}, b: {type: integer}}, required: [a, b], }, }, { name: get_discount, description: 查询商品今日折扣, inputSchema: { type: object, properties: {sku: {type: string}}, required: [sku], }, }, ] def call_tool(name: str, args: dict) - Any: if name add: return {result: args[a] args[b]} if name get_discount: # 真实场景这里会查数据库/调内部 API return {sku: args[sku], discount: 0.85} raise ValueError(funknown tool: {name}) def handle(msg: dict) - None: method msg.get(method) mid msg.get(id) if method initialize: send({ jsonrpc: 2.0, id: mid, result: { protocolVersion: 2025-06-18, capabilities: {tools: {}}, serverInfo: {name: minimal-server, version: 0.1.0}, }, }) elif method notifications/initialized: log(client initialized, ready) elif method tools/list: send({jsonrpc: 2.0, id: mid, result: {tools: TOOLS}}) elif method tools/call: params msg.get(params, {}) try: r call_tool(params[name], params.get(arguments, {})) send({ jsonrpc: 2.0, id: mid, result: { content: [{type: text, text: json.dumps(r, ensure_asciiFalse)}], isError: False, }, }) except Exception as e: send({ jsonrpc: 2.0, id: mid, result: { content: [{type: text, text: str(e)}], isError: True, }, }) elif method ping: send({jsonrpc: 2.0, id: mid, result: {}}) else: log(funhandled method: {method}) if __name__ __main__: for line in sys.stdin: line line.strip() if not line: continue handle(json.loads(line))一个能跑的 MCP Server 骨架就是这么薄。注意notifications/initialized没有id它是通知不是请求Server 不需要回包。5. 最小 Client理解 Host 侧的握手再看 Host 侧怎么跟 Server 对话。一个最小 Client 需要启动子进程 →initialize→ 发initialized通知 → 调工具。#!/usr/bin/env python3 minimal_mcp_client.py — 最小 MCP Client演示完整握手 import json import subprocess import sys proc subprocess.Popen( [sys.executable, minimal_mcp_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1, ) def request(method: str, params: dict, mid: int) - dict: proc.stdin.write(json.dumps( {jsonrpc: 2.0, id: mid, method: method, params: params} ) \n) proc.stdin.flush() return json.loads(proc.stdout.readline()) # 1. 握手协商协议版本与能力 resp request(initialize, { protocolVersion: 2025-06-18, capabilities: {}, clientInfo: {name: minimal-client, version: 0.1.0}, }, mid1) assert resp[result][protocolVersion] 2025-06-18 print(握手成功:, resp[result][serverInfo]) # 2. 通知 Server 初始化完成通知没有 id proc.stdin.write(json.dumps({jsonrpc: 2.0, method: notifications/initialized}) \n) proc.stdin.flush() # 3. 拉取工具清单 tools request(tools/list, {}, mid2)[result][tools] print(工具:, [t[name] for t in tools]) # 4. 调用工具 r request(tools/call, {name: add, arguments: {a: 40, b: 2}}, mid3) print(add(40,2) , r[result][content][0][text])跑起来输出握手成功: {name: minimal-server, version: 0.1.0} 工具: [add, get_discount] add(40,2) {result: 42}整个协议没有魔法就是协商 → 通知 → 请求/响应三次交互和普通 RPC 没有本质区别。6. Streamable HTTP无状态化的关键设计本地用 stdio云端就得上 HTTP。2025-06-18 规范把 SSE 升级为 Streamable HTTP普通请求走 POST 立即返回 JSON需要流式时服务端用Content-Type: text/event-stream推事件客户端拿到sessionId后在后续请求头里带上Mcp-Session-Id保持会话。生产环境最关键的一条POST 请求必须幂等、无状态这样前面挂多少个 Nginx/LB 都不怕。典型请求长这样POST /mcp HTTP/1.1 Host: mcp.example.com Content-Type: application/json Accept: application/json, text/event-stream Mcp-Session-Id: a1b2c3d4e5 {jsonrpc:2.0,id:1,method:tools/call,params:{name:get_discount,arguments:{sku:SKU-001}}}服务端流式响应HTTP/1.1 200 OK Content-Type: text/event-stream event: message data: {jsonrpc:2.0,id:1,result:{content:[{type:text,text:{\sku\: \SKU-001\, \discount\: 0.85}}]}}用 curl 验证握手和会话先发 initializecurl -i -X POST http://127.0.0.1:8765/mcp \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -d {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2025-06-18,capabilities:{},clientInfo:{name:curl,version:1.0}}}响应头里会带Mcp-Session-Id把它记下来后续请求带上curl -i -X POST http://127.0.0.1:8765/mcp \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -H Mcp-Session-Id: a1b2c3d4e5 \ -d {jsonrpc:2.0,id:2,method:tools/list,params:{}}如果返回text/event-stream你会看到event: message加一行data:这就是流式通道在工作。这也是为什么 MCP 网关Gateway在企业里成为标配——它把散落的 stdio Server 统一转换成 HTTP 出口还能顺手做鉴权、限流、审计。7. 本篇常见错排查协议污染stdio 模式下任何多余输出print调试、第三方库的日志都会让 Client 解析崩溃。所有日志走 stderr这是线上事故第一高发点。我踩过的坑就是某个依赖库默认往 stdout 打 banner排查了半天。超时不泄漏流式模式下 SSE 长连接要设置空闲超时Client 用完必须释放否则服务端连接数只涨不跌。典型的生产事故建议在网关层加连接数上限。握手版本不匹配Client 发2025-06-18Server 回了个旧版本assert直接挂。排查时先看initialize的响应体别急着看工具逻辑。Session 丢失Streamable HTTP 下忘了带Mcp-Session-Id服务端会当成新会话工具状态全丢。curl 验证时务必把响应头里的 session id 复制到下一个请求。安全边界MCP 统一了接线却不会自动装保险丝。生产环境必须做到最小权限账号、只读优先、root 目录限定、OAuth Token 短期化、高危操作默认禁止。工具内部如果要回调模型做二次推理Key 从环境变量读别写进代码。8. 接入与验证把 Server 挂到 Host 上Server 跑通后把它挂到 Host 上验证。stdio 版直接把settings.json里的command/args指向你的脚本HTTP 版填url和headers。验证模型侧是否正常可以用模型对话页面发一条消息确认 Host 能列出你的工具模型对话验证工具是否被正确识别https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewriteCoding Plan长期编码/Agent 场景统一 Key 覆盖https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite接入文档协议细节与端点https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite如果你在 Claude Code 里接Anthropic 兼容入口在这里https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite排障时优先看 API Keys 和接入文档两页Key 权限、端点格式、协议版本对不上九成问题都出在这。