直接给结论把一个 MCP Server 跑起来很快快则半小时慢则一个下午。但真到了生产环境你会发现“能跑”和“能扛”之间隔着一条很宽的河。鉴权怎么防绕过、日志怎么不打进密钥、流式输出怎么处理客户端中途断开、多用户同时使用时会话状态怎么不串……这些 SD K 的 demo 不会替你考虑的问题才是生产级 MCP Server 的真正难点。这篇文章是我从零手写一个生产级 MCP Server 的完整记录重点落在这四件事上鉴权、流式传输、状态管理以及配套的日志与稳定性调优。适合两类人看一类是正在从 demo 走向生产的开发者另一类是想理解 MCP 协议底层实现细节、准备自己造轮子的同学。我会直接给工程方案、代码片段和踩坑过程尽量让文章能当一份 checklist 用。1. 从“能跑”到“敢上生产”MCP Server 中间藏了哪些硬需求MCPModel Context Protocol的核心定位很简单让 LLM 应用通过标准化的 JSON-RPC 2.0 接口去调用外部工具、数据源和资源。协议层面Server 最常见要处理的方法无非这几类initialize建连、tools/list暴露工具列表、tools/call触发工具执行再加上notifications/initialized这类通知。但从工程架构来说这些都只是“业务逻辑层”生产环境的 MCP Server 还必须叠加额外的一层又一层。我用过不少团队自己写的 MCP Server功能角度看都正常但一部署到生产立刻暴露出一堆问题没有统一的鉴权入口token 散落在各个工具函数里校验SSE 流式输出时用户一刷新页面后端任务还在继续跑产生一堆无效调用多人同时使用同一个服务互相看到对方的对话上下文日志里明晃晃地打印着Authorization: Bearer sk-xxx…拿出去做故障排查时自己先被吓一跳。这里我整理了一个“生产级最低要求清单”也是我在设计服务时反复对照的关注点demo 阶段生产阶段身份认证不做或每个工具各自判断网关/传输层统一鉴权支持多租户密钥管理硬编码在代码里环境变量或密钥管理系统日志脱敏请求追踪无request_id session_id 贯穿日志超时控制依赖上游默认超时连接超时/总超时/空闲超时分开设置并发控制来多少执行多少信号量限流防上游被打爆会话状态全局 dict一塞了事会话隔离 TTL 可恢复日志管理print 默认 logging结构化 JSON 轮转 敏感字段过滤优雅关闭杀进程就完事SIGTERM 接后台任务取消资源回收这个清单并不是要求你一次性全做完而是先形成框架再逐步补齐。下面我按重要性逐个拆开讲先从鉴权开始——因为一旦密钥泄露其他所有安全性都失去意义。2. 鉴权不能只验一次Token 校验、密钥落盘与日志脱敏的实操拆解2.1 为什么鉴权必须收敛到传输层入口很多人写 MCP Server 时习惯在每个 tool 里自己判断 token# 不要这样写每个工具都重复鉴权逻辑迟早漏一个 server.tool() async def query_order(order_id: str, token: str): if not valid_token(token): return {error: unauthorized} ...这样做的坏处很明显业务代码和鉴权逻辑耦合新增一个工具忘写校验就是漏洞token 以参数形式出现在请求体里日志一打就泄露。正确的做法是在传输层做一个“门卫”所有请求先过鉴权再进业务逻辑。我用的是 FastAPI 的依赖注入# auth.py from fastapi import Depends, HTTPException, Header import jwt from datetime import datetime, timezone async def verify_token(authorization: str Header(default)): if not authorization.startswith(Bearer ): raise HTTPException(status_code401, detailmissing token) token authorization.removeprefix(Bearer ).strip() try: payload jwt.decode(token, SECRET_KEY, algorithms[HS256]) except jwt.PyJWTError: raise HTTPException(status_code401, detailinvalid token) return { user_id: payload[sub], tenant_id: payload.get(tenant, default), exp: payload.get(exp), }然后在所有 MCP 路由上统一加Depends(verify_token)。这样 token 只出现在 Header 里不会进入业务逻辑层也方便后续接入限流、审计。2.2 密钥泄露的最大入口不是代码是日志在讲“鉴权绕过”这个问题之前我得先说一个更隐蔽的坑大部分密钥泄露不是被攻击者破解的而是自己打印出去的。第三方库在 debug 级别会打印完整请求头很多框架默认 access log 会把路径参数完整输出如果你用logger.debug(headers)那Authorization里的 token 就进了日志。所以我自己实现了一个日志过滤器对所有敏感字段做统一的脱敏处理# log_filter.py import logging, re SENSITIVE_PATTERNS [ re.compile(r(Authorization:\s*Bearer\s)[A-Za-z0-9._\-], re.I), re.compile(r(\access_token\\s*:\s*\)[^\], re.I), re.compile(r(\api_key\\s*:\s*\)[^\], re.I), ] class SensitiveDataFilter(logging.Filter): def filter(self, record: logging.LogRecord) - bool: msg record.getMessage() for pattern in SENSITIVE_PATTERNS: msg pattern.sub(lambda m: m.group(1) ***, msg) record.msg msg return True这一步做完之后即使某个库把 headers 打到日志里敏感信息也已经被替换成了***。我在部署完第一个版本后专门 grep 了一遍线上日志里的Authorization和sk-确认清干净了才敢对外接流量。2.3 防止鉴权绕过的几个细节鉴权绕过通常不是高深的黑客技巧而是工程上的“漏洞三连”第一白名单和黑名单混淆。有人喜欢在路由里写“只需要给这几个接口加鉴权”这是最危险的。正确姿势是默认全鉴权只有明确允许的接口比如/healthz走白名单app FastAPI(dependencies[Depends(verify_token)]) app.get(/healthz, dependencies[]) # 显式开放健康检查 async def healthz(): return {status: ok}第二JWT 的校验只做了“能解码”就放行。必须同时校验exp和iss否则一个过期 token 照样能通过。第三限流缺失导致的暴力尝试。鉴权接口一般都有防护但工具调用接口反而常常没有限流攻击者可以拿着一个低权限 token 反复调同一个工具制造消耗。我后面会在第 6 章专门讲并发限制但这属于同一类问题生产级的 MCP Server 必须默认任何入口都可能被滥用。3. 流式传输没你想的那么“流”SSE 断连、背压和增量推送的工程化处理3.1 MCP 流式传输在解决什么问题MCP 协议里tools/call的响应既可以是一条完整 JSON也可以拆成多个 JSON-RPC 消息通过 SSEServer-Sent Events逐条推给客户端。真实的业务需求很直接工具执行很慢需要不断推送notifications/progress让用户看到进度工具返回的数据量很大比如查询日志、生成报告一次性拼一个超大 JSON内存会先撑不住LLM 应用侧希望边生成边展示延迟越低越好。所以流式传输的工程重点不是“能不能用 SSE 推数据”而是“推的时候怎么保证稳定”。3.2 一个带断连检测和背压控制的 SSE 实现先看最基础的 SSE 写法很多人是这么开的from fastapi import FastAPI from fastapi.responses import StreamingResponse import json, asyncio app FastAPI() app.post(/mcp/stream) async def mcp_stream(request: Request, payload: dict): async def event_generator(): # 模拟长耗时工具的多次返回 for i in range(10): chunk {jsonrpc: 2.0, method: notifications/progress, params: {progress: i * 10}} yield fdata: {json.dumps(chunk)}\n\n await asyncio.sleep(1) # 最终结果 result {jsonrpc: 2.0, id: 1, result: {content: [{type: text, text: done}]}} yield fdata: {json.dumps(result)}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这段代码 demo 没问题上线就有三个隐患第一个隐患客户端断开后生成器还在跑。用户刷新页面或断网后FastAPI 的StreamingResponse会尝试往已断开的连接写数据但我们已经开启的任务不会自动取消。解决方法是监听request.is_disconnected()async def event_generator(): for i in range(10): if await request.is_disconnected(): break # 客户端已断开停止任务 yield fdata: ...\n\n await asyncio.sleep(1)第二个隐患背压问题。如果客户端消费速度慢于服务端生产速度数据会在缓冲区堆积内存一路涨。解决方式是用有界队列做中间缓冲queue: asyncio.Queue asyncio.Queue(maxsize32) async def event_generator(): while True: try: # 放入生产结果队列满了就丢弃最旧的消息 queue.put_nowait(chunk) except asyncio.QueueFull: await queue.get() # 丢弃最旧 queue.put_nowait(chunk) yield fdata: {json.dumps(chunk)}\n\n await asyncio.sleep(0.1)队列的maxsize就是“允许的最大积压量”超过就丢弃旧消息保证内存不会无限膨胀。生产上长任务可以保留最后一条进度 最终结果中间过程丢弃完全没问题。第三个隐患心跳缺失。SSE 连接经过 Nginx 或负载均衡时如果长时间没有数据中间设备会主动切断连接。所以必须周期性发送注释行作为心跳yield : keepalive\n\n每 15 秒发一次就可以避免“什么都正常但连接莫名断了”的问题。3.3 为什么我建议把大响应拆成“进度事件 最终结果”实测下来流式传输最稳的模式是先用notifications/progress事件分段推送进度最后推一条完整的results事件。不要试图把一个超大 JSON 拆成多个半截 JSON 分开传输客户端解析会很难受。进度事件只是“提示”最终结果才是“事实”协议两侧都要按这个约定来实现。4. 会话状态不是字典里塞个值多租户隔离、过期回收与状态重建4.1 状态管理不只是“存对话记录”MCP 协议里有一个Mcp-Session-IdHeader用于标识客户端与服务器之间的一次逻辑会话。它的管理维度包括连接建立状态initialize完成、客户端能力声明对话上下文多轮工具调用之间的关联数据工具执行中的临时状态批量任务做到第几步、分页游标到哪个位置鉴权后的用户/租户信息避免每个工具重新解析 token。这类状态如果在多用户环境里不做隔离后果是会后端串号——A 用户看到 B 用户的对话内容。我在压测环境里真实遇到过因为全局 dict 里用了session_id以外的字段做 key两个并发请求同时写入上下文互相覆盖。排查到凌晨才反应过来。4.2 内存 dict 方案为什么不能直接上生产很多人第一步都会用字典sessions {}这个方案在单体 demo 下完全没问题但生产环境至少有四个问题维度全局 dict 的问题隔离性不同用户/租户的数据容易互相覆盖过期回收没有 TTLsession 只增不减内存泄漏多实例两台实例各自维护一份请求哈希到 B 实例时上下文丢失异常恢复进程一重启所有会话全灭4.3 单机场景的最优解TTL Cache 会话 ID如果服务还没到分布式程度我建议先用带过期时间的缓存来兜底比如cachetools.TTLCachefrom cachetools import TTLCache # 每个租户 1000 个会话过期时间 30 分钟 session_cache: TTLCache TTLCache(maxsize10_000, ttl1800) def get_session(session_id: str, user_id: str): key f{user_id}:{session_id} # 天然隔离不同租户 return session_cache.get(key) def set_session(session_id: str, user_id: str, state: dict): key f{user_id}:{session_id} session_cache[key] state这里有两个设计细节值得注意一个是key 里带 user_id。即使两个会话 ID 相同不同用户的数据也不会串另一个是TTL 不能设成永久。对话状态本身是“短期热数据”30 分钟没有活跃就丢掉节省了内存也避免了垃圾堆积。4.4 多实例部署后必走的一步存 Redis如果服务要横向扩容TTL Cache 就不够了。必须把会话状态放到共享存储我用 Redis 的主要数据模型是import redis, json, time r redis.Redis(hostredis, port6379, decode_responsesTrue) def save_session(session_id: str, user_id: str, state: dict, ttl: int 1800): key fmcp:session:{user_id}:{session_id} r.setex(key, ttl, json.dumps(state)) def load_session(session_id: str, user_id: str): key fmcp:session:{user_id}:{session_id} data r.get(key) return json.loads(data) if data else None4.5 状态“丢失”时的降级方案不管用哪种存储都要接受一个现实会话状态会丢网络会抖进程会重启。所以生产级设计里一定要有无状态恢复的能力。具体做法是每个请求必须是幂等的至少对“创建型”工具调用加上去重。客户端在建立连接时带上断点续传所需的上下文摘要。Server 端如果发现 session 不存在返回特定错误码比如-32603并提示客户端重新执行initialize。不要幻想“状态永远在”。把状态当作可随时重建的缓存来设计才是生产级的正确心态。5. 日志系统决定你深夜能被叫醒几次结构化日志与自定义管理的落地细节5.1 默认 logging 在生产环境基本不可用默认的 Python logging 输出你是分不清“哪个请求产生这条日志”的。比如多用户并发时INFO tool query_order called ERROR failed to call upstream你完全不知道这是谁的请求、哪个会话、花了多长时间。生产故障排查靠这种日志等于抓瞎凌晨两点你会被一条“服务异常”的告警叫醒然后花一小时看日志也定位不了问题。5.2 打通 request_id 和 session_id 是关键一步我的做法是在中间件里生成request_id放进contextvars然后所有日志输出时都带上它。同时把 MCP 层已经解析好的session_id和user_id也注入进去。# context.py from contextvars import ContextVar request_id_var: ContextVar[str] ContextVar(request_id, default-) session_id_var: ContextVar[str] ContextVar(session_id, default-) user_id_var: ContextVar[str] ContextVar(user_id, default-)# middleware.py import uuid from .context import request_id_var, session_id_var, user_id_var app.middleware(http) async def request_context(request: Request, call_next): rid uuid.uuid4().hex[:12] request_id_var.set(rid) session_id_var.set(request.headers.get(Mcp-Session-Id, -)) user_id_var.set(getattr(request.state, user_id, -)) response await call_next(request) response.headers[X-Request-Id] rid return response5.3 自定义一个结构化的 JSON Formatter有了contextvars接下来就是把日志输出格式改成 JSON。这是我最推荐的自定义日志管理方式每条日志本身就是一行结构化 JSON方便直接接入日志平台、按字段过滤。# json_logger.py import json, logging from .context import request_id_var, session_id_var, user_id_var class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) - str: base { ts: self.formatTime(record, %Y-%m-%d %H:%M:%S), level: record.levelname, logger: record.name, msg: record.getMessage(), } # 注入链路上下文 base[request_id] request_id_var.get() base[session_id] session_id_var.get() base[user_id] user_id_var.get() if hasattr(record, extra_fields): base.update(record.extra_fields) return json.dumps(base, ensure_asciiFalse)然后配置 RotatingFileHandler 做日志轮转避免磁盘被日志打满import logging.handlers handler logging.handlers.RotatingFileHandler( mcp-server.log, maxBytes50 * 1024 * 1024, backupCount7, encodingutf-8 ) handler.setFormatter(JsonFormatter()) root_logger logging.getLogger() root_logger.addHandler(handler)5.4 日志脱敏和鉴权章节里做过的过滤结合第 2 章那个SensitiveDataFilter在这里要挂到根 logger 上root_logger.addFilter(SensitiveDataFilter())这样不管日志是哪个第三方库打印的只要是经过根 logger都会先脱敏。实测下来这行代码救过我很多次——有些底层库的 debug 日志会把完整 URL 和 query 参数带出来里面的 token 要是不脱敏日志平台一公开就是事故。5.5 一条理想日志长什么样接入这套日志方案之后线上日志大概长这样{ts: 2025-06-01 12:00:01, level: INFO, logger: mcp.tool, msg: query_order called, request_id: a3f2b9c1e4d5, session_id: sess_9d2k, user_id: user_77, extra: {tool: query_order, elapsed_ms: 234}}看到这条日志我能直接回答哪个用户、哪个会话、哪个工具、花了多久。再加上注解里的elapsed_ms我甚至不需要额外写性能监控日志就已经是基础 APM 了。6. 生产环境压测暴露的问题超时、并发与资源回收的最终防线6.1 上游工具调用必须显式设置超时MCP Server 本质是个“中间人”LLM 调用你你去调上游 API/数据库/内部服务。上游一旦慢你这里就会积压大量请求。Nginx 默认 60s 断开但你的服务如果不去主动掐断慢调用线程和连接会被拖死。我给所有 upstream 调用都加了超时控制import asyncio timeout asyncio.timeout(15) # 15 秒总超时 try: async with timeout: result await call_upstream_http(...) except TimeoutError: logger.warning(upstream call timed out, exc_infoTrue) raise HTTPException(status_code504, detailupstream timeout)asyncio.timeout是 3.11 的写法旧版本可以用asyncio.wait_for。要区分“连接建立”和“总耗时”两个超时参数不能只设一个。6.2 用信号量挡住并发洪峰压测时我发现一个现象并发 50 个请求同时调一个需要外部 API 的工具上游 API 直接被我们自己打挂了。LLM 应用侧并发很高如果工具执行不加限制一个 MCP Server 足以变成 DDoS 放大器。我用的方案是按工具类型加asyncio.Semaphoretool_semaphores: dict[str, asyncio.Semaphore] {} DEFAULT_CONCURRENCY 20 def get_semaphore(tool_name: str) - asyncio.Semaphore: if tool_name not in tool_semaphores: tool_semaphores[tool_name] asyncio.Semaphore(DEFAULT_CONCURRENCY) return tool_semaphores[tool_name] async def call_with_limit(tool_name: str, func, *args, **kwargs): sem get_semaphore(tool_name) async with sem: return await func(*args, **kwargs)注意信号量的释放必须用async with否则finally写漏一次信号量就永久少一个服务很快进入“假死”状态。这个错误我确实遇到过因为忘了释放一个信号量槽位20 个并发槽被占满后后续所有请求全部卡死。6.3 压测数据参考我自己用一个 2C4G 的容器做了压测模拟的是“每个工具调用需要访问一个上游 HTTP 接口耗时 200ms~500ms”的场景指标未加超时信号量加了超时信号量并发 100QPS崩溃前 QPS 峰值约 80内存持续增长稳定在 60 左右内存波动 10%慢上游占比 10%线程池被打满响应最大延迟 30s延迟控制在 1s 内内存持续爬升最后 OOM稳定GC 回收正常结论很明确牺牲一点点吞吐换来稳定性和可用性这在生产环境是完全值得的。6.4 优雅关闭与资源回收最后一个容易被忽略的细节是优雅关闭。你用CtrlC杀掉进程时如果有正在执行的长任务最好能给它一个取消信号让它在 coder 里收尾。用 uvicorn 的话可以监听lifespanfrom contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动时初始化连接池、redis、信号量等 yield # 关闭时取消所有后台任务关闭连接池 pending [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] for t in pending: t.cancel() await asyncio.gather(*pending, return_exceptionsTrue) redis_client.close()这样服务发布时可以保证不留下半通不亮的任务和孤儿连接。虽然看起来是“最后一步”但真到了发布群里喊“谁的服务还在占用连接池”你会感谢这段代码的。我在实际开发中的体会是生产级 MCP Server 最难的不是协议本身而是那些协议之外的东西日志打通了问题定位效率能提升一个量级鉴权收敛到传输层后面接多少新工具都不会漏状态管理先想好隔离和过期再谈分布式扩展。如果你现在正准备把 MCP Server 推到生产建议先从日志和超时控制这两个“最不性感”的部分做起——这两个做好了其他细节才有机会被看见。