1. 为什么本地跑通的 MCP Server，一上生产就崩

MCP（Model Context Protocol）是 Anthropic 在 2024 年 11 月开源的协议，用来把 LLM 和外部工具、数据源用一套标准接口连起来。你可以把它理解成 AI 世界的 USB-C：以前每个模型对接每个工具都要写一套胶水代码，M×N 的组合爆炸；有了 MCP，模型侧和工具侧各自实现一次协议，就能互相插拔。它适合谁？适合手里已经有一个能跑的本地 demo、想把它推到「可观测、可压测、可上线」的开发者，也适合正在用 Cline、Claude Code 这类客户端接工具、但被 Key 管理和鉴权搞烦的人。

我见过太多教程停在mcp.run(transport="stdio")就结束了。本地 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_BASE="https://taotoken.net/api" export TAOTOKEN_API_KEY="sk-你的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，写操作做成 Tool，LLM 判断何时调用会更准。

先看 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\s+OUTFILE\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", tool="execute_query", row_count=len(rows), latency_ms=round((time.monotonic() - t0) * 1000, 2)) return json.dumps({"row_count": len(rows), "rows": rows}, ensure_ascii=False, default=str) except Exception as e: log.error("tool_fail", tool="execute_query", error_type=type(e).__name__, error_msg=str(e)) return json.dumps({"error": str(e)}, ensure_ascii=False)

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 type='table' AND name NOT LIKE 'sqlite_%'") result = {} for row in schemas: cols = await db.fetch_all(f"PRAGMA table_info({row['name']})") result[row["name"]] = {"ddl": row["sql"], "columns": cols} return json.dumps(result, ensure_ascii=False, indent=2)

Prompt 是预置模板，用户主动触发：

@mcp.prompt() def analyze_table(table_name: str, focus: str = "overview") -> str: """生成数据分析提示词模板。""" return (f"请分析表 `{table_name}`。先读取 " f"schema://tables/{table_name} 获取结构，" f"再用 execute_query 做探索查询，每次限 100 行。")

启动入口区分开发和生产：

if __name__ == "__main__": # 开发: mcp.run(transport="stdio") mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

5. 鉴权与限流：把 demo 变成能扛住 LLM 循环的服务

生产环境必须上 Streamable HTTP，stdio 模式下 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, capacity=20, refill_rate=10.0): self.capacity = capacity self.refill_rate = refill_rate self.tokens = capacity self.last = time.monotonic() self._lock = asyncio.Lock() async def acquire(self, timeout=5.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(capacity=20, refill_rate=10.0)

日志必须输出到 stderr，这是 stdio 模式下的铁律，HTTP 模式也建议保持，方便采集：

# logger.py import structlog, sys structlog.configure( processors=[ structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(ensure_ascii=False), ], logger_factory=structlog.PrintLoggerFactory(file=sys.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": f"Bearer {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(timeout=10) as c: tasks = [one_call(c, "your-token") for _ in range(100)] results = await asyncio.gather(*tasks, return_exceptions=True) ok = [r for r in results if isinstance(r, tuple) and r[0] == 200] print(f"成功 {len(ok)}/100") asyncio.run(main())

实测下来的数据大致是这样：

瓶颈很清楚：50 并发以内卡在 SQLite 的 I/O，写锁是库级别的；50 到 100 卡在单进程 uvicorn 的事件循环；100 以上是令牌桶 capacity=20 在排队。对应优化：换 PostgreSQL、uvicorn --workers 4、按实际负载调 capacity。

7. 本篇常见错排查

坑 1：stdout 日志污染协议通道。现象是客户端连上就断，报 JSON 解析失败。根因是 stdio 模式下 stdout 是协议通道，print()的日志被当成协议消息。解法：所有日志走 stderr，用 structlog 配file=sys.stderr。

坑 2：Resource 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通知，或重启客户端。

坑 5：Streamable HTTP 下 SSE 断连。长耗时 Tool（>30 秒）连接超时。根因是 Nginx 默认proxy_read_timeout 60s。解法：配proxy_read_timeout 300s加proxy_buffering off。

坑 6：鉴权令牌和模型 Key 混用。有人把 TaoToken 的 API Key 直接塞进 MCP Server 当鉴权令牌，结果权限范围对不上。记住：模型侧 Key 归客户端管，Server 侧用独立的 JWT，两套体系。

8. 安全加固清单与下一步

上线前逐项过一遍，这是从 demo 到生产的最后一道关：

Docker 化时用多阶段构建，最终镜像能压到 150MB 以内，运行阶段切非 root 用户，加健康检查端点。这些配置在完整代码里都有。

下一步你可以做两件事：一是把 SQLite 换成 PostgreSQL，解决并发读的瓶颈；二是把限流器从单机内存换成 Redis，支持多实例共享配额。模型侧继续用 TaoToken 统一通道，Key 轮换只改一个地方，客户端配置不用动。

如果你在接入过程中卡在鉴权或客户端配置上，先去 API Keys 页面确认令牌权限范围，再对照接入文档里的分场景配置；想先验证模型侧通道是否通，用模型对话跑一条最简单的请求最快；如果是长期跑编码任务或 Agent，直接上 Coding Plan，配额和并发更稳。