☰
从零实现一个 MCP 服务:用 TaoToken 统一 Key 打通 JSON-RPC 与 stdio 落地
2026/10/8 11:32:20 网站建设 项目流程

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 服务是什么、能做什么、适合谁。MCP(Model 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数,配置的时候别把查询串带进去,否则某些客户端会报 URL 解析错误。

先拿 Key。进控制台创建 API Key,路径是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,创建完在 API Keys 页面能看到完整 Key,形如sk-开头的一串。这个 Key 只显示一次,复制下来存好。API Keys 页面地址是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。

接下来是配置。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 mcp

Python 版本要求 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_ascii=False) 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 type='table'" ).fetchall() conn.close() return json.dumps([r[0] for r in rows], ensure_ascii=False) @mcp.resource("database://schema") def get_schema() -> str: """数据库表结构说明""" conn = sqlite3.connect(DB_PATH) rows = conn.execute( "SELECT name FROM sqlite_master WHERE type='table'" ).fetchall() schema = [] for (t,) in rows: cols = conn.execute(f"PRAGMA 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。如果你用的是 FastMCP,SDK 内部已经处理了 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_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite ,编码和 Agent 场景用 Coding Plan,地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,Claude Code 相关的说明在 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite 。

到这里,从 JSON-RPC 消息格式、stdio 传输、SDK 搭建最小服务端,到用 TaoToken 统一 Key 接入 Claude Code,整条链路就走完了。剩下的就是按你的实际数据源去扩展工具——把 SQLite 换成 MySQL、把本地文件换成内部 API,思路完全一样。协议那几行 JSON 看懂了,后面都是在上层堆功能。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询