“002MCP”这个代号说白了是我个人折腾了大半个月的一个小项目一个基于 MCPModel Context Protocol模型上下文协议的统一上下文处理中间层。起初只是想在多个 AI 工具之间共享同一份记忆和工具集后来发现能做的东西越来越多索性重构成了一套独立的服务。MCP 这个名字在圈内这两年越来越响但真正上手做过完整闭环的人不算多所以我打算把这半个月踩过的坑、理过的思路、敲过的代码原原本本写出来。这篇东西适合刚接触 MCP 的开发者也适合已经在跑单机 demo、想往服务化方向挪一挪的团队。你会看到它是怎么解决“上下文碎片化”这个痛点的不同模型、不同工具、不同会话之间怎么统一管理上下文、怎么暴露工具调用、怎么控 token、怎么扛并发。我也会把项目里最核心的“上下文窗口管理”和“工具动态注册”这两块拆开来讲最后附上我实际调试时遇到的几个经典问题。建议读完的人可以直接对着代码捋一遍很多细节只可意会写出来反而啰嗦。1. 为什么叫 002MCP需求拆解与方案选型1.1 项目背景上下文碎片化到底有多痛先说一个场景。本地同时跑着好几个 AI 客户端有的走第三方模型接口有的连本地模型还有的是我用脚本临时调用的 API。每个客户端各自维护一套会话历史A 工具里聊过的结论B 工具完全不知道。一旦想跨会话汇总、让多个智能体协作就只能在各个终端之间手动复制粘贴历史记录效率低到让人崩溃。更深层的问题是工具调用。早期项目集成每个 AI 应用都要单独写一套插件或工具封装给 A 写的天气查询换个模型又得重写一遍给 B 写的数据库查询换个框架根本没法复用。等于每接一个新产品就重复造一遍轮子。这种碎片化本质上是上下文和工具没有从业务中解耦出来。所以我给这个项目起了个代号“002”意思是第二版思路第一版我用的是自定义 Socket 协议能跑但扩展性太差第二版彻底转向 MCP 标准把上下文和工具统一成独立服务于是就叫 002MCP。1.2 设计目标把上下文变成可插拔的基础设施立项时我给这个服务定了三个目标。第一统一入口所有 AI 应用都通过 MCP 协议连接这同一个服务不再各自维护状态第二工具复用服务端注册一次工具所有客户端都能调用新增能力只改服务端不动客户端第三上下文可控每轮对话的上下文按策略自动裁剪、按需注入避免“越聊越贵、越聊越乱”。三个目标对应到架构上就是三层接入层负责协议转换和客户端鉴权管理层负责上下文存储、检索和窗口控制执行层负责工具注册、参数校验和实际调用。分层之后好处是每层都能独立替换比如协议层今天支持 MCP明天想加一个自定义 HTTP 接口只需要在接入层加一个适配器后面两层完全不用动。1.3 为什么选 MCP 而不是自定义协议第一版我走了自定义协议的老路用 JSON 串加一个简单路由表跑单机 demo 没问题但多客户端接入、权限控制、流式响应这些问题全部得自己造。MCP 作为一个开放的、基于 JSON-RPC 的标准协议把这些问题都定义清楚了客户端与服务端之间的会话初始化、工具列表获取、工具调用、资源订阅、取消请求都有现成规范。协议标准化带来的直接收益就是生态互通今天用一个客户端明天换另一个只要支持 MCP 就能直接接入我的服务。当然 MCP 也有学习成本它的规范里有不少概念比如 protocol version、capability negotiation、content block初次接触会觉得抽象。我的建议是别急着啃规范全文先跑通一个最小闭环再回头理解概念效率会高很多。2. 架构设计与核心机制2.1 整体模块怎么切整个服务我拆成四个模块MCP Protocol Adapter、Context Manager、Tool Registry、Executor。它们的协作关系我用大白话讲客户端发来一条 JSON-RPC 消息Adapter 负责解析并把消息转成内部结构体Context Manager 负责从会话存储中加载历史、拼接当前请求的上下文并保存新产生的消息Tool Registry 放着所有已注册工具的定义包括参数 schemaExecutor 根据客户端请求的工具名和参数动态执行对应函数并把结果转成 MCP 格式返回。这四个模块里最容易被低估的是 Context Manager。很多人以为 MCP 只是把工具暴露出去其实上下文管理才是真正拉开体验差距的地方。我在 Context Manager 里实现了三个策略滑动窗口、摘要压缩、关键信息提取。滑动窗口保留最近 N 轮完整对话摘要压缩在会话过长时自动触发把早期对话总结成摘要块关键信息提取则是正则加规则把对话里的数字参数、文件名、目标地址这类高价值信息单独抽出来长期保存在会话记忆里。2.2 一条消息完整的生命周期我不喜欢空谈架构直接写一条消息从进入到返回的全过程这样你能直观看到每层做了什么。客户端发送tools/call { name: query_order, arguments: { order_id: A1001 } }服务端处理链Adapter 校验 JSON-RPC 格式和协议版本鉴权通过后转换为内部请求结构体。Context Manager 先从会话存储中取出该 session 的最近上下文摘要附带关键信息表。Tool Registry 按工具名查到 query_order 的定义用 JSON Schema 校验 arguments缺字段直接报参数错误。Executor 执行真实函数查询订单库返回结果给 Adapter。Adapter 把结果包装成 MCP 的 call_tool_result同时 Context Manager 把这次的调用记录追加到会话尾部。整个过程里只有第 4 步是业务相关的前 3 步和最后一步都是通用逻辑换任何工具都一样。这也是我强烈推荐做一层的核心原因把业务逻辑和协议逻辑分干净后续加工具就是往 Registry 里塞一条定义和对应函数完事。2.3 上下文窗口控制怎么防止 token 爆掉这是我在项目里花时间最多的地方。会话一长如果总是把完整历史丢给模型很快 token 上限就会被打爆。我采用的策略组合是按角色分段裁剪 按 token 估算动态调度。具体实现如下MAX_CONTEXT_TOKENS 8000 RECENT_LINES_TO_KEEP 20 def build_context(session): recent session.history[-RECENT_LINES_TO_KEEP:] older session.history[:-RECENT_LINES_TO_KEEP] if estimate_tokens(older) MAX_CONTEXT_TOKENS * 0.5: summary summarize(older) return summary recent return session.history先用一个简单的 tokenizer按字符数和关键词命中做估算算出历史总量如果早期部分太大则触发摘要压缩。摘要模型可以用轻量级模型也可以直接调上下文管理服务里内置的压缩接口。实测下来8000 token 的窗口压到一半以内对话质量几乎没有下降响应速度提升明显。提示token 估算别用精确 tokenizer太慢直接用字节数乘系数误差在 15% 以内完全够用。3. 实操过程从零搭起 002MCP 核心服务3.1 搭建服务端骨架我用的语言是 Python主要原因有三MCP 官方 SDK 对 Python 支持最好异步生态成熟我的工具函数也多是 Python 写的复用成本最低。如果你团队主要技术栈是 TypeScript官方同样有 SDK实现思路基本相同代码层面稍作翻译即可。一个最小的 MCP 服务端核心代码如下from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(002mcp) app.list_tools() async def list_tools(): return [ Tool( nameecho, description回传一段文本用来测试连通性, inputSchema{ type: object, properties: { text: {type: string, description: 要回传的文本} }, required: [text] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name echo: text arguments.get(text, ) return [TextContent(typetext, textfecho: {text})] async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options())这段代码里list_tools是给客户端看的“菜单”call_tool是真正的执行入口。inputSchema直接用 JSON Schema 语法客户端会根据这个 schema 自动生成参数表单或做输入校验。先把这段跑通mcp 客户端能连上说明最底层已经通了。3.2 动态注册与参数校验让工具变成配置每个工具都手动写一遍 list 和 call 里的分支项目一多代码就会膨胀。我改成声明式注册每个工具就是一个字典加一个函数自动生成列表和分发逻辑。TOOL_DEFINITIONS [ { name: query_order, description: 按订单号查询状态, inputSchema: { type: object, properties: { order_id: {type: string} }, required: [order_id] }, handler: query_order_function }, ] TOOL_MAP {t[name]: t for t in TOOL_DEFINITIONS} app.list_tools() async def list_tools(): return [Tool(namet[name], descriptiont[description], inputSchemat[inputSchema]) for t in TOOL_DEFINITIONS] app.call_tool() async def call_tool(name: str, arguments: dict): tool TOOL_MAP.get(name) if not tool: raise ValueError(funknown tool: {name}) validate_arguments(tool[inputSchema], arguments) result await tool[handler](arguments) return [TextContent(typetext, textstr(result))]新增一个工具只需要在TOOL_DEFINITIONS里加一条记录list 和 call 自动就会带上它。validate_arguments我直接用了现成的 JSON Schema 校验库校验失败会返回标准错误客户端收到的错误信息也能知道具体是哪个字段不合法。3.3 客户端接入多端复用的关键一步服务端通了客户端接入反而往往是新手最容易卡住的地方。我用官方 SDK 写一个最小客户端示例import asyncio from mcp.client.stdio import stdio_client from mcp.client import ClientSession async def main(): async with stdio_client(server_command[python, server.py]) as (read, write): async with ClientSession(read, write) as session: init_result await session.initialize() print(协议版本:, init_result.protocolVersion) tools await session.list_tools() print(可用工具:, [t.name for t in tools]) result await session.call_tool(query_order, {order_id: A1001}) print(调用结果:, result) asyncio.run(main())这里用的传输方式是 stdio适合本机进程间通信。如果客户端在另一台机器建议换成 SSE 或 Streamable HTTP 传输官方 SDK 都支持只是配置略有不同。我的实际项目里既有本地 CLI 客户端也有远程 Web 服务二者都通过这个同样的 MCP 协议接入客户端代码主体几乎一模一样区别只在传输层。4. 常见问题与排查技巧实录4.1 排查表这些坑我几乎全踩过我不太信“一次成功”这种事这个表是我调项目时亲自遇到并解决的大概率你也会碰到。现象根因解决办法客户端 connect 后马上断开服务端初始化响应格式不对或协议版本不匹配检查 initialize 返回里是否包含 protocolVersion 和 capabilities工具列表拉不到list_tools 里返回的对象不是 Tool 类型确认用了 mcp.types.Tool而不是普通 dict调用工具报 invalid paramsJSON Schema 写错如 required 字段写成字符串用 JSON Schema 校验器先单独验证 schema 合法性上下文永远只有最后几轮会话存储没有正确追加新消息在 call_tool 处理结束后记得调用 context_manager.append并发一高就报错全局变量存了会话状态导致串话把会话状态改成按 session_id 隔离的字典加锁长文本返回被截断MCP 协议层对单条消息大小有限制改用多块 content 返回或启用流式输出其中会话串话是最容易隐藏的问题。一开始我图省事把上下文对象定义为模块级全局变量两个客户端同时连接A 的对话记录被 B 覆盖排查了半天才发现是这个原因。后来所有状态一律按 session_id 为 key 存到一个带锁的上下文字典里问题彻底消失。4.2 请求超时与重试机制工具调用如果内部逻辑很慢客户端等太久就会超时。MCP 规范里没有强制超时机制但这恰恰是坑。我测试真实数据库查询时一个慢查询可能跑两三秒前端已经超时。我的处理方案是三层配合第一层客户端侧设置合理的超时时间比如默认 15 秒慢工具单独加长第二层服务端把慢工具的执行放到独立线程池中不让它阻塞其他请求第三层对于超时但可能正在执行的任务提供标准错误返回并支持客户端二次轮询查询状态。工具执行我使用 asyncio 的wait_for包装代码如下try: result await asyncio.wait_for(handler(arguments), timeouttimeout) except asyncio.TimeoutError: return [TextContent(typetext, textTool execution timed out)]千万别把超时设得过于激进。我之前设 5 秒结果本地模型加载权重都要 4 秒多导致工具永远“超时”。后面改成按工具独立设置超时时间慢查询类工具配置 30 秒普通查询 10 秒才稳定下来。4.3 上下文压缩token 开销降了一半最后说说效果。在开启上下文压缩前后我统计了同一批对话的平均 token 消耗。未开启压缩时20 轮对话后单次请求约消耗 9000 token部分模型直接爆上限开启后会话历史超过阈值后早期内容被压缩成摘要单次请求降到 4000 token响应速度也从平均 4.2 秒降到 2.8 秒。压缩的代价是会丢失一部分细节但如果只保留“最终结论”和“关键数字”对绝大多数任务足够。我的建议是压到原体积 30%-50% 是比较甜点的区间压太狠后续对话容易答非所问压太少又起不到省 token 的作用。5. 工具选型与后续扩展方向5.1 SDK 选型对比Python 与 TypeScriptMCP 生态目前最成熟的是 TypeScript 和 Python 两套官方 SDK。Python 的异步风格更贴近我的习惯类型定义也够清晰TypeScript 的客户端集成度更高如果前端项目直接用起来会更顺。我个人的建议是服务端优先选 Python理由有三异步生态好、科学计算和数据处理库多、把业务函数封装成工具的成本最低前端或 Node 环境为主的团队选 TypeScript 也完全合理。对于传输方式stdio 适合本地进程SSE 适合局域网内多客户端Streamable HTTP 适合跨网络场景。我目前主要跑 stdio 加一个 HTTP 网关后续打算把两个传输方式都加上让同一份工具定义同时暴露给本机和远程客户端。5.2 后续扩展从单机到多机、多模型002MCP 现在跑在单机上后续有两个明确的扩展方向。第一个方向是横向拆分工具域把不同业务域的工具拆成多个 MCP 服务再通过一个聚合网关对外暴露统一接口。好处是不同的工具服务可以独立部署、独立升级一个服务故障不影响其他。第二个方向是支持多模型路由。现在一个客户端只接一个模型实际场景里复杂推理用大模型简单查询用小模型成本可以省下一大截。我计划在 Context Manager 之上加一个 router 层根据当前请求的估算难度和 token 量动态选择底层模型。因为上层已经是统一协议加这一层不会动到现有工具注册和调用逻辑。我还想提一下“熟悉”与“标准化”的取舍。MCP 规范本身还在快速演进不同版本之间会有一些细微差异这很正常。我的态度是拥抱标准但标准变动时留好适配层别把协议细节散落在业务代码里。002MCP 里所有协议相关的代码都集中在 adapter 目录里将来协议升级改动面积能控制在很小范围。动手跑一遍比看十遍文章有用得多。你可以先从上面的最小服务端开始把 echo 跑通再慢慢往里填自己的工具函数。等你能做到“在服务端加一个工具客户端零改动就能调用”你就真正体会到 002MCP 这类统一上下文层的价值了。
