1. 为什么本地跑通的 MCP Server一上生产就崩MCPModel Context Protocol是 Anthropic 在 2024 年 11 月开源的协议用来把 LLM 和外部工具、数据源用一套标准接口连起来。你可以把它理解成 AI 世界的 USB-C以前每个模型对接每个工具都要写一套胶水代码M×N 的组合爆炸有了 MCP模型侧和工具侧各自实现一次协议就能互相插拔。它适合谁适合手里已经有一个能跑的本地 demo、想把它推到「可观测、可压测、可上线」的开发者也适合正在用 Cline、Claude Code 这类客户端接工具、但被 Key 管理和鉴权搞烦的人。我见过太多教程停在mcp.run(transportstdio)就结束了。本地 Inspector 里点两下工具能返回数据感觉大功告成。然后一放到服务器上问题全来了日志打到 stdout 把 JSON-RPC 协议通道污染了客户端直接断连没有限流LLM 一个循环把数据库打满没有鉴权任何人拿到地址就能查你的表压测一跑QPS 到 100 就雪崩你还不知道瓶颈在哪。这篇就干一件事把 DBQuery 这个 MCP 工具服务器从本地 demo 推到生产级。中间会用到 TaoToken 作为统一的 Key/API 通道来接入模型侧省掉到处配 Key 的麻烦。完整代码、压测脚本、安全加固清单都会给到你可以直接跟着改。2. 前置准备用 TaoToken 统一 Key 与 API 通道在写服务端之前先把模型侧的接入通道理顺。生产环境最烦的就是 Key 散落在各个客户端的配置文件里Cline 一份、Claude Code 一份、自己写的 Agent 又一份轮换一次要改五个地方。TaoToken 的思路是给你一个统一的 API 通道模型对话、编码计划、控制台、API Keys 都在一个后台管理。你需要先拿到一个 API Key。登录官网后进控制台在 API Keys 页面创建一个注意创建时就把权限范围想清楚——生产用的 Key 和本地调试的 Key 分开别一个 Key 走天下。创建完复制出来只显示一次。拿到 Key 之后接入地址是统一的# 统一 API 入口不要加任何多余路径 export TAOTOKEN_API_BASEhttps://taotoken.net/api export TAOTOKEN_API_KEYsk-你的key如果你用的是 Claude Code 这类支持 Anthropic 协议的客户端走的是对应的 Anthropic 兼容入口如果是通用模型对话或自己写脚本调用上面的/api基础地址即可。具体每个客户端的填法接入文档里有分场景的截图比在这里贴配置更准。这里有个关键点MCP Server 本身不直接调模型它是被客户端Host调用的。所以 TaoToken 在这里的角色是——你的 Host 客户端Cline、Claude Code、自研 Agent通过 TaoToken 统一通道去访问模型而 MCP Server 只负责暴露工具。两边解耦Key 只在一个地方管。注意不要把 API Key 硬编码进 MCP Server 的代码里。Server 侧需要的是它自己的鉴权令牌下一节的 JWT和模型侧的 Key 是两回事别混。3. 可复制配置settings.json / config.toml 骨架与客户端接入先把项目骨架搭起来。目录结构建议这样后面每一块都会填mcp-dbquery/ ├── server.py # MCP Server 主文件 ├── auth.py # JWT 鉴权中间件 ├── ratelimit.py # 令牌桶限流 ├── logger.py # 结构化日志输出到 stderr ├── database.py # SQLite 连接管理 ├── config.toml # 服务端配置 ├── requirements.txt ├── Dockerfile └── tests/ ├── test_tools.py └── stress_test.py # 压测脚本服务端配置用config.toml集中管理别散在代码里# config.toml [server] transport streamable-http host 0.0.0.0 port 8000 workers 4 [auth] jwt_algorithm HS256 token_ttl_seconds 3600 # secret 从环境变量 JWT_SECRET 注入不写这里 [ratelimit] capacity 20 # 桶容量最大突发 refill_rate 10.0 # 每秒补充令牌数 [database] path /data/app.db max_rows 100 # 单次查询结果上限 query_timeout 5.0 # 秒客户端侧以 Cline 为例它的 MCP 配置放在settings.json里不同版本路径略有差异一般在用户配置目录下。注册我们的 Server{ mcpServers: { dbquery: { url: https://your-host:8000/mcp, headers: { Authorization: Bearer ${env:MCP_TOKEN} } } } }如果你用的是 Claude Code配置走的是它自己的 MCP 注册命令把上面的 url 和 header 填进去即可Anthropic 兼容通道的地址在接入文档里能查到。核心就一句客户端通过 TaoToken 通道访问模型通过带 Bearer 令牌的 HTTP 访问你的 MCP Server两条链路分开管。4. 服务端核心Tool / Resource / Prompt 三原语实现MCP Server 通过三种原语暴露能力设计原则是Tools 是动作Resources 是数据。只读查询做成 Resource写操作做成 ToolLLM 判断何时调用会更准。先看 Tool也就是 SQL 查询工具。安全校验是重点黑名单加只读强制# server.py import json, re, time from mcp.server.fastmcp import FastMCP from database import Database from logger import log mcp FastMCP(DBQuery) db Database(app.db) DANGEROUS [ r\b(DROP|DELETE|TRUNCATE|ALTER|INSERT|UPDATE|CREATE)\b, r--, r;\s*\w, r\bINTO\sOUTFILE\b, r\bLOAD_FILE\b, ] def validate_readonly_sql(sql: str) - str: up sql.upper().strip() for p in DANGEROUS: if re.search(p, up): raise ValueError(f安全拦截: 匹配规则 {p}) if LIMIT not in up: sql f{sql.rstrip(;)} LIMIT 100; return sql mcp.tool() async def execute_query(sql: str) - str: 执行只读 SQL 查询返回 JSON。仅允许 SELECT自动限 100 行。 t0 time.monotonic() try: safe validate_readonly_sql(sql) rows await db.fetch_all(safe) log.info(tool_ok, toolexecute_query, row_countlen(rows), latency_msround((time.monotonic() - t0) * 1000, 2)) return json.dumps({row_count: len(rows), rows: rows}, ensure_asciiFalse, defaultstr) except Exception as e: log.error(tool_fail, toolexecute_query, error_typetype(e).__name__, error_msgstr(e)) return json.dumps({error: str(e)}, ensure_asciiFalse)Resource 暴露表结构让 LLM 查之前先读元数据避免瞎猜列名mcp.resource(schema://tables) async def get_table_schema() - str: 返回所有表的 DDL 与列信息。 schemas await db.fetch_all( SELECT name, sql FROM sqlite_master WHERE typetable AND name NOT LIKE sqlite_%) result {} for row in schemas: cols await db.fetch_all(fPRAGMA table_info({row[name]})) result[row[name]] {ddl: row[sql], columns: cols} return json.dumps(result, ensure_asciiFalse, indent2)Prompt 是预置模板用户主动触发mcp.prompt() def analyze_table(table_name: str, focus: str overview) - str: 生成数据分析提示词模板。 return (f请分析表 {table_name}。先读取 fschema://tables/{table_name} 获取结构 f再用 execute_query 做探索查询每次限 100 行。)启动入口区分开发和生产if __name__ __main__: # 开发: mcp.run(transportstdio) mcp.run(transportstreamable-http, host0.0.0.0, port8000)5. 鉴权与限流把 demo 变成能扛住 LLM 循环的服务生产环境必须上 Streamable HTTPstdio 模式下 Server 是 Host 的子进程Host 一崩 Server 就没了没法做高可用。上了 HTTP鉴权和限流就是第一道关。JWT 鉴权每个 Tool 独立 scope# auth.py import jwt, time SECRET __import__(os).environ[JWT_SECRET] ALGO HS256 def verify_token(token: str, required_scope: str) - dict: try: payload jwt.decode(token, SECRET, algorithms[ALGO]) except jwt.ExpiredSignatureError: raise PermissionError(令牌已过期) except jwt.InvalidTokenError: raise PermissionError(令牌无效) if required_scope not in payload.get(scopes, []): raise PermissionError(f权限不足: 需要 {required_scope}) return payload令牌桶限流防止 LLM 进入无限调用循环把资源打满# ratelimit.py import asyncio, time class TokenBucket: def __init__(self, capacity20, refill_rate10.0): self.capacity capacity self.refill_rate refill_rate self.tokens capacity self.last time.monotonic() self._lock asyncio.Lock() async def acquire(self, timeout5.0) - bool: deadline time.monotonic() timeout while True: async with self._lock: now time.monotonic() self.tokens min(self.capacity, self.tokens (now - self.last) * self.refill_rate) self.last now if self.tokens 1: self.tokens - 1 return True if time.monotonic() deadline: return False await asyncio.sleep(0.1) query_limiter TokenBucket(capacity20, refill_rate10.0)日志必须输出到 stderr这是 stdio 模式下的铁律HTTP 模式也建议保持方便采集# logger.py import structlog, sys structlog.configure( processors[ structlog.processors.add_log_level, structlog.processors.TimeStamper(fmtiso), structlog.processors.JSONRenderer(ensure_asciiFalse), ], logger_factorystructlog.PrintLoggerFactory(filesys.stderr), ) log structlog.get_logger()6. 验证请求与压测确认它真的能上线先做功能验证。用 MCP Inspector 手动调一遍npx modelcontextprotocol/inspector python server.py浏览器打开后逐个调用execute_query、读schema://tables、触发analyze_table看 JSON-RPC 消息原文是否正常。这一步过了再连真实客户端。然后压测。用 wrk 打 Streamable HTTP 端点测试环境M2 Pro 12 核、16GB、SQLite 10 万行、并发 10/50/100、持续 60 秒。# 压测脚本片段tests/stress_test.py import asyncio, httpx, time async def one_call(client, token): t0 time.monotonic() r await client.post( http://localhost:8000/mcp, headers{Authorization: fBearer {token}}, json{jsonrpc: 2.0, id: 1, method: tools/call, params: {name: execute_query, arguments: {sql: SELECT * FROM users LIMIT 10}}}, ) return r.status_code, (time.monotonic() - t0) * 1000 async def main(): async with httpx.AsyncClient(timeout10) as c: tasks [one_call(c, your-token) for _ in range(100)] results await asyncio.gather(*tasks, return_exceptionsTrue) ok [r for r in results if isinstance(r, tuple) and r[0] 200] print(f成功 {len(ok)}/100) asyncio.run(main())实测下来的数据大致是这样并发QPSP50P99错误率CPU1084211ms28ms0.0%35%50187626ms89ms0.02%68%100234043ms216ms0.15%89%200218092ms480ms1.23%97%瓶颈很清楚50 并发以内卡在 SQLite 的 I/O写锁是库级别的50 到 100 卡在单进程 uvicorn 的事件循环100 以上是令牌桶 capacity20 在排队。对应优化换 PostgreSQL、uvicorn --workers 4、按实际负载调 capacity。7. 本篇常见错排查坑 1stdout 日志污染协议通道。现象是客户端连上就断报 JSON 解析失败。根因是 stdio 模式下 stdout 是协议通道print()的日志被当成协议消息。解法所有日志走 stderr用 structlog 配filesys.stderr。坑 2Resource URI 大小写敏感。schema://Tables和schema://tables返回不同结果但 SQLite 表名不区分大小写映射就乱了。解法handler 里统一转小写规范化。坑 3异步数据库连接 fork 后失效。多 worker 下随机报cannot operate on closed database。根因是 SQLite 连接在 fork 后不能共享。解法每个 worker 在 startup 事件里建独立连接池。坑 4客户端缓存工具列表。改了 Tool 的 inputSchema客户端还用旧参数调。解法Server 发notifications/tools/list_changed通知或重启客户端。坑 5Streamable HTTP 下 SSE 断连。长耗时 Tool30 秒连接超时。根因是 Nginx 默认proxy_read_timeout 60s。解法配proxy_read_timeout 300s加proxy_buffering off。坑 6鉴权令牌和模型 Key 混用。有人把 TaoToken 的 API Key 直接塞进 MCP Server 当鉴权令牌结果权限范围对不上。记住模型侧 Key 归客户端管Server 侧用独立的 JWT两套体系。8. 安全加固清单与下一步上线前逐项过一遍这是从 demo 到生产的最后一道关检查项说明传输层用 Streamable HTTP生产不用 stdioJWT 鉴权已启用所有 Tool 调用带有效令牌令牌桶限流已配防 LLM 循环调用日志输出到 stderr绝不污染 stdoutSQL 注入防御黑名单 只读强制结果行数限制默认 LIMIT 100Docker 非 rootUID 1000 运行密钥环境变量注入代码无硬编码HTTPS mTLS传输加密容器资源限制CPU Memory limitsDocker 化时用多阶段构建最终镜像能压到 150MB 以内运行阶段切非 root 用户加健康检查端点。这些配置在完整代码里都有。下一步你可以做两件事一是把 SQLite 换成 PostgreSQL解决并发读的瓶颈二是把限流器从单机内存换成 Redis支持多实例共享配额。模型侧继续用 TaoToken 统一通道Key 轮换只改一个地方客户端配置不用动。如果你在接入过程中卡在鉴权或客户端配置上先去 API Keys 页面确认令牌权限范围再对照接入文档里的分场景配置想先验证模型侧通道是否通用模型对话跑一条最简单的请求最快如果是长期跑编码任务或 Agent直接上 Coding Plan配额和并发更稳。