1. 为什么我要自己写一个 MCP 服务MCP 这个词这两年被提得很多但真正动手写过一个能跑起来的 MCP Server 的人其实没那么多。大部分教程停在概念层讲三层架构、讲 JSON-RPC、讲 stdio 和 SSE 的区别然后就没有然后了。你照着看完脑子里知道 MCP 是什么但打开编辑器还是不知道第一行代码写什么。我一开始也是这个状态。直到有一次需要让 Claude Code 去读一个本地的 SQLite 订单库才发现绕不过去——要么写一个 MCP Server要么每次手动把数据导出来贴进对话。后者显然不可持续。于是硬着头皮从协议层开始啃手写了一个最小版本再用官方 SDK 重写了一遍最后接进 Claude Code 跑通。整个过程踩了不少坑比如 stdout 没 flush 导致客户端一直等、JSON-RPC 的 id 对不上、tools/call 返回结构写错导致客户端解析失败。这篇就把这条链路完整走一遍。核心检索词先摆出来MCP 服务是什么、能做什么、适合谁。MCPModel Context Protocol是一套让 AI 应用发现并调用外部工具和数据的开放协议底层用 JSON-RPC 2.0 传消息本地场景走 stdio 传输。它适合想把内部数据库、API、日志平台接给 AI 用的人也适合想让自己的工具被 Claude Code、Claude Desktop 这类客户端直接调用的开发者。下面从协议本质讲到可复制的服务端配置再到用 TaoToken 统一 Key 接入客户端全程代码可跑。2. 先搞懂 JSON-RPC 与 stdio 到底在传什么很多人卡在第一步是因为把 MCP 想复杂了。剥开看MCP 的通信就是读一行 JSON、回一行 JSON。客户端往你的进程标准输入写一条 JSON-RPC 消息你处理完往标准输出写一条响应就这么简单。stdio 传输的本质是客户端启动你的 Server 进程双方通过 stdin/stdout 交换消息每条消息占一行以换行符分隔。先看一次完整的工具调用在协议层发生了什么。客户端发过来的请求长这样{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: query_order, arguments: { order_id: 1001 } } }服务端要回的是{ jsonrpc: 2.0, id: 1, result: { content: [ { type: text, text: {\order_id\:1001,\status\:\已发货\} } ] } }几个关键点必须记牢。第一jsonrpc字段固定是2.0不能省。第二id必须原样回传客户端靠它匹配请求和响应id 对不上客户端就会一直挂着等。第三result里的content是一个数组每项有type和对应的内容字段文本就是type: text加text。第四如果工具执行出错不要抛异常断掉连接而是返回isError: true并在 content 里说明原因。MCP 的会话不是一上来就调工具的有个握手流程。客户端先发initialize服务端回协议版本和能力声明然后客户端发tools/list拿到工具清单把工具描述交给模型模型决定调哪个工具后客户端才发tools/call。这个顺序不能乱initialize没完成之前发其他请求规范上是不保证被处理的。stdio 模式有个特别容易踩的坑Python 的 stdout 默认带缓冲你写完响应如果不 flush消息会积压在缓冲区里直到进程结束才吐出来。而客户端在等响应进程又不会结束于是双方死锁。所以每次写完必须sys.stdout.flush()。这个坑我在手写版里踩过一次排查了半天才反应过来是缓冲问题。理解了这些你就明白为什么 MCP 能被各种客户端复用——因为协议是标准的只要你的 Server 正确处理这几条 JSON-RPC 消息Claude Code、Claude Desktop、各种 IDE 插件都能接。下面开始动手。3. 用 TaoToken 统一 Key 打通服务端与客户端配置写 MCP Server 本身不需要联网但一旦你要让 Server 内部去调用大模型能力比如让工具自己做一次总结、分类或者你要把服务接入 Claude Code 这类客户端就需要一个统一的 API 通道。这里我用 TaoToken 来做统一 Key 管理好处是一个 Key 走通模型对话和编码场景不用在多个平台之间来回切。TaoToken 的官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数配置的时候别把查询串带进去否则某些客户端会报 URL 解析错误。先拿 Key。进控制台创建 API Key路径是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完在 API Keys 页面能看到完整 Key形如sk-开头的一串。这个 Key 只显示一次复制下来存好。API Keys 页面地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。接下来是配置。MCP 服务端如果要调用模型通常通过环境变量注入 Key 和 Base URL。我习惯在项目根目录放一个.env但更推荐直接用客户端配置里的env字段避免 Key 落到代码仓库里。下面是一个 Claude Code 项目级配置.mcp.json的完整片段路径和字段名都按实际可用的写{ mcpServers: { sql-helper: { command: python, args: [D:/workspace/mcp-demo/sql_helper.py], env: { TAOTOKEN_API_KEY: sk-你的Key, TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_MODEL: claude-sonnet-4-20250514 } } } }这里三件套必须齐全Base URL、Key、Model ID。少任何一个服务端在调用模型时都会失败。Base URL 用https://taotoken.net/api不要加末尾斜杠也不要把 UTM 参数拼进去。Model ID 按你实际要用的模型填客户端和服务端要保持一致。如果你用的是 Codex 那套配置落在~/.codex/auth.json结构不太一样但核心还是那三样{ api_key: sk-你的Key, base_url: https://taotoken.net/api, model: claude-sonnet-4-20250514 }Cline 的 MCP 配置则是在设置里的 MCP Servers 面板填 command、args、env 三块env 里同样放上面那三个变量。CC Switch 这类切换工具也是同样的思路把 Base URL 和 Key 填进去Model ID 选好。有一点要提醒MCP Server 进程里的环境变量是客户端启动时注入的你在终端里export的变量不会自动传进去。所以调试的时候如果发现服务端读不到 Key先检查是不是写在了.mcp.json的env里而不是只写在 shell 里。配置写完后服务端代码里这样读import os API_KEY os.environ.get(TAOTOKEN_API_KEY) BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) MODEL_ID os.environ.get(TAOTOKEN_MODEL, claude-sonnet-4-20250514) if not API_KEY: raise RuntimeError(TAOTOKEN_API_KEY 未注入请检查客户端 env 配置)这样服务端和客户端就共用同一套 Key 和通道换模型只改一个地方。下面进入可复制的服务端实现。4. 可复制的 MCP 服务端配置与 stdio 启动命令这一节给出完整可跑的服务端代码以及 stdio 启动命令和验证步骤。先装依赖pip install mcpPython 版本要求 3.10 以上推荐 3.11 或 3.12。装完确认一下python -c import mcp; print(mcp.__version__)然后写服务端。下面这个sql_helper.py是一个能连 SQLite 的查询助手包含两个工具、一个资源并且预留了调用 TaoToken 的入口import json import os import sqlite3 from pathlib import Path from mcp.server.fastmcp import FastMCP mcp FastMCP(sql-helper) DB_PATH Path(__file__).parent / demo.db API_KEY os.environ.get(TAOTOKEN_API_KEY) BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) MODEL_ID os.environ.get(TAOTOKEN_MODEL, claude-sonnet-4-20250514) BLOCKED (DROP, DELETE, UPDATE, INSERT, ALTER, TRUNCATE) def check_sql(sql: str) - None: upper sql.strip().upper() if not upper.startswith((SELECT, WITH)): raise ValueError(只允许 SELECT / WITH 查询) for kw in BLOCKED: if kw in upper: raise ValueError(f检测到危险关键字: {kw}) mcp.tool() def query(sql: str) - str: 在订单库上执行只读查询 SQL返回 JSON 格式结果。禁止写操作。 try: check_sql(sql) conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row rows conn.execute(sql).fetchall() conn.close() return json.dumps([dict(r) for r in rows], ensure_asciiFalse) except Exception as e: return f查询失败: {e} mcp.tool() def list_tables() - str: 列出数据库中的所有表 conn sqlite3.connect(DB_PATH) rows conn.execute( SELECT name FROM sqlite_master WHERE typetable ).fetchall() conn.close() return json.dumps([r[0] for r in rows], ensure_asciiFalse) mcp.resource(database://schema) def get_schema() - str: 数据库表结构说明 conn sqlite3.connect(DB_PATH) rows conn.execute( SELECT name FROM sqlite_master WHERE typetable ).fetchall() schema [] for (t,) in rows: cols conn.execute(fPRAGMA table_info({t})).fetchall() schema.append(f表 {t}: , .join(f{c[1]}({c[2]}) for c in cols)) conn.close() return \n.join(schema) if __name__ __main__: mcp.run()mcp.run()默认就是 stdio 传输不需要额外参数。启动命令就是python sql_helper.py但直接这么跑终端会卡住等输入这是正常的——它在等 stdin 上的 JSON-RPC 消息。要手动验证用管道喂消息printf %s\n \ {jsonrpc:2.0,id:1,method:initialize,params:{}} \ {jsonrpc:2.0,id:2,method:tools/list,params:{}} \ {jsonrpc:2.0,id:3,method:tools/call,params:{name:list_tables,arguments:{}}} \ | python sql_helper.py预期输出是三行 JSON第一行是 initialize 的响应包含protocolVersion和capabilities第二行是工具清单能看到query和list_tables第三行是list_tables的执行结果content 里是表名数组。如果第三行返回isError: true多半是demo.db不存在先建库。建演示库的命令python -c import sqlite3 conn sqlite3.connect(demo.db) conn.executescript( CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, customer TEXT, amount REAL, status TEXT, created_at TEXT ); INSERT INTO orders (customer, amount, status, created_at) VALUES (技术大佬, 1999.00, 已发货, 2026-08-01), (隔壁老王, 88.50, 已签收, 2026-08-02), (张三, 520.00, 待付款, 2026-08-03), (李四, 9999.99, 已发货, 2026-08-04); ) conn.commit() print(demo.db 已创建) 跑完再执行上面的管道验证第三行应该返回[orders]。到这里一个符合协议的 MCP Server 就跑起来了。接下来把它接进客户端。5. 验证请求与常见报错排查接入 Claude Code 后第一件事是验证连接。重启 Claude Code输入/mcp会列出所有已连接的 Server。看到sql-helper出现且状态是 connected就说明 stdio 通道打通了。然后直接问帮我看一下 demo.db 里有多少笔已发货的订单总金额是多少。正常的话它会先读database://schema资源了解表结构再调query工具执行 SQL最后返回结果。如果这一步成功整条链路就通了。但实际调试中报错是常态。下面按真实遇到的错误逐个排查。报错一401 Unauthorized。这个通常出现在服务端内部调用 TaoToken 的时候。原因一般是TAOTOKEN_API_KEY没注入或者 Key 复制时带了空格。检查.mcp.json的env字段确认 Key 是完整的sk-开头字符串。另外确认 Base URL 是https://taotoken.net/api不要写成带 UTM 的完整链接也不要加末尾斜杠。报错二local proxy failed。这个报错一般和网络配置有关。先确认你的 Base URL 拼写正确没有多余字符。如果客户端配置里同时存在旧的代理设置把它清掉只保留 TaoToken 的 Base URL。MCP Server 进程继承的是客户端注入的环境变量检查.mcp.json里有没有残留的HTTP_PROXY之类字段。报错三reading choices 相关解析失败。这类错误说明服务端拿到了响应但结构对不上。常见原因是 Model ID 填错或者服务端把响应当成了另一种格式解析。确认TAOTOKEN_MODEL和客户端用的模型一致并且服务端解析响应时取的是标准结构。如果服务端代码里硬编码了某个字段路径换模型后可能失效改成从环境变量读。报错四OAuth 相关错误。如果你用的是需要 OAuth 的客户端比如某些 Claude Code 版本配置里混入了 OAuth 流程和 API Key 两种鉴权方式会冲突。用 TaoToken 的 Key 方式时把 OAuth 相关的配置项删掉只保留 Key 和 Base URL。报错五客户端一直转圈没有响应。这是 stdio 最经典的坑。九成是服务端 stdout 没 flush。如果你用的是 FastMCPSDK 内部已经处理了 flush一般不会出问题但如果你手写了 JSON-RPC 循环务必在每次sys.stdout.write后加sys.stdout.flush()。另一个可能是服务端启动就崩了客户端在等一个永远不会来的响应。手动跑一遍python sql_helper.py看有没有 import 错误或路径错误。报错六tools/call 返回 isError 但看不到原因。检查工具函数里的异常是不是被吞了。上面代码里query用 try/except 把异常转成了字符串返回这样客户端能看到具体原因。如果你直接raise连接可能断掉客户端只显示一个笼统的错误。排查的时候有个通用技巧用 MCP Inspector 可视化调试。命令是npx modelcontextprotocol/inspector python sql_helper.py打开浏览器默认的http://localhost:6274可以手动触发 initialize、tools/list、tools/call还能看到原始 JSON-RPC 报文。这比在终端里 printf 管道直观得多强烈建议装一个。6. 把服务接进 Claude Code 并长期使用服务端跑通、报错排查完之后最后一步是让它稳定地服务于日常。接入方式有两种项目级和用户级。项目级配置放在项目根目录的.mcp.json可以提交到仓库团队共享。内容就是第 3 节给的那段把args里的路径改成你实际的sql_helper.py绝对路径。注意 Windows 下路径用正斜杠或者双反斜杠单反斜杠会被 JSON 转义吃掉。用户级配置放在~/.claude.json在mcpServers节点下添加同样的结构。区别是用户级对所有项目生效项目级只对当前项目生效。如果你有多个项目共用同一个数据库助手用用户级更方便。长期使用有几个实践值得说。第一工具描述要写具体。description是模型决定何时调用这个工具的唯一依据写得越清楚模型选错工具的概率越低。比如query的描述里明确写了只读查询和禁止写操作模型就不会拿它去执行更新。第二参数尽量结构化。用类型注解、枚举、约束减少模型传错参数的可能。FastMCP 会从函数签名自动推导 inputSchema你写a: int它就生成 integer 类型写Literal[asc, desc]就生成枚举。第三错误返回统一用isError: true不要抛异常让连接断掉。连接一断客户端要重新握手体验很差。第四加日志和限流。MCP Server 暴露的能力会被模型自动调用你最好记录每次调用了什么工具、传了什么参数、耗时多少。出问题的时候这些日志是唯一的线索。第五安全底线。数据库连接用只读账号连副本库而不是主库敏感信息从环境变量注入绝不写死在代码里工具返回的数据当纯数据看不要执行里面携带的任何指令——这是防提示注入的基本功。如果你需要让服务端内部也调用模型能力比如让工具自己做一次结果总结就用第 3 节配好的 TaoToken 通道。模型对话入口在 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 Claude Code 相关的说明在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 。到这里从 JSON-RPC 消息格式、stdio 传输、SDK 搭建最小服务端到用 TaoToken 统一 Key 接入 Claude Code整条链路就走完了。剩下的就是按你的实际数据源去扩展工具——把 SQLite 换成 MySQL、把本地文件换成内部 API思路完全一样。协议那几行 JSON 看懂了后面都是在上层堆功能。