ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

MCP协议深度拆解:从手写最小Server到生产实践

MCP协议深度拆解:从手写最小Server到生产实践 MCP Server 这波热度基本是从 AI 工具链里炸出来的。说白了以前你要让 AI 助手去查数据库、调接口、读文件每个接入都要定制一套工具调用方式各家有各家的规范做起来非常碎。MCPModel Context Protocol模型上下文协议想做的事情就是把“AI 应用怎么连外部数据源和工具”这件事统一成一套标准协议相当于给 AI 生态做一个“USB-C 接口”。这篇文章我打算从协议原理开始讲然后带你不依赖任何高级封装手写一个最小 MCP Server最后再切换到官方 SDK 做生产可用版本。无论你是刚开始接触 MCP、还是已经能跑通 demo 但没搞懂内部时序这篇文章应该都能帮你把最后一层窗户纸捅破。1. MCP 到底解决了什么问题1.1 为什么 AI 应用需要一套“万能插座”在过去很长一段时间里给 AI 模型接外部工具是典型的“点对点”模式。你的应用接一个数据库就要写一个数据库插件接一个办公软件就要再写一个办公软件插件。插件越多维护成本越高而且每个插件都要自己定义参数格式、返回值结构、错误处理方式。模型服务商、应用开发者、工具提供方三方各搞各的生态非常碎片化。MCP 的思路是定义一个公共的“插座”模型应用是 Host它通过内部的 MCP Client 去连接各种 MCP Server每个 Server 只需要按照协议暴露自己的能力不关心对面到底是谁。这样一来工具提供方只需要实现一次 MCP Server就能被任何支持 MCP 的客户端复用。这和网络里的 mesh 组网、应用里的单点登录 SSO 本质上是同一类思想先定一套大家都遵守的协议再把点对点的对接成本降下来。1.2 MCP 架构中的三个角色MCP 架构里最核心的是三个角色Host模型应用本身比如桌面客户端、IDE 插件、智能助手。它负责和用户交互也是整个流程的发起方。ClientHost 内置的协议客户端负责连接 Server、发送请求、接收响应。Server实现协议的服务端暴露 Tools、Resources、Prompts 三类能力给客户端调用。很多刚接触的人会把 Client 和 Server 搞混。简单记法谁被启动、谁提供服务谁就是 Server谁去连它、谁去调用它谁就是 Client。Host 只是个更上层的容器里面可以同时管理多个 Client 和多个 Server。1.3 能力模型工具、资源、提示词MCP 的 Server 可以暴露三类能力这也是协议层面对“功能”做的抽象Tools可执行的函数比如查天气、发邮件、计算表达式。AI 模型根据用户需求决定是否调用。Resources可读取的数据用 URI 标识比如一个文件内容、一行数据库记录、一张文档截图。Prompts模板化的提示词帮助用户或模型按固定结构发起任务。这三类能力分别对应协议里的tools/*、resources/*、prompts/*方法。理解这个分类非常重要因为后面所有代码都围绕着“如何注册能力、如何响应请求”展开。2. 协议原理深度拆解消息、生命周期与调用模型2.1 MCP 的消息格式和传输层MCP 在应用层使用的是 JSON-RPC 2.0 协议。所有消息都是 JSON 对象并且分为三类Request请求、Response响应、Notification通知。一个请求消息至少包含jsonrpc、id、method、params四个字段响应则必须包含jsonrpc、id以及result或error通知和请求很像但它不包含id也不需要任何响应。下面是一个最基本的请求{jsonrpc: 2.0, id: 1, method: ping, params: {}}对应的响应{jsonrpc: 2.0, id: 1, result: {}}传输层方面MCP 目前最常见的两种方式是stdioServer 作为本地子进程启动通过标准输入 stdout/stdin 传输换行分隔的 JSON 消息。Streamable HTTPServer 作为远程 HTTP 服务通过 POST 请求发送 JSON-RPC 消息并可选支持 SSE 流式返回。对新手来说最友好的切入点是 stdio。你只需要把一个 Python/Node 进程跑起来输入输出全部走标准输入输出不需要考虑端口、Token、鉴权这些东西。2.2 从握手到工具调用的完整生命周期MCP 的通信不是上来就随便调它有严格的握手流程Client 发送initialize请求带上自己的协议版本、能力声明、客户端信息。Server 返回initialize响应返回它选定的协议版本、自身能力、服务器信息。Client 再发送一个notifications/initialized通知告诉 Server“我已经知道你的能力了现在可以正常工作了”。只有完成前三步之后Client 才能调用tools/list、tools/call、resources/read等方法。这个顺序很容易被忽视很多人拿官方 SDK 写mcp.tool()能跑通但一旦自己手写协议层就容易在初始化没完成时就处理业务请求导致各种诡异问题。initialize请求的简化结构如下{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: {name: my-client, version: 1.0.0} } }Server 的响应要包含自己能支持的协议版本和能力声明{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: {name: minimal-server, version: 0.1.0} } }这里有一个容易被忽略的细节protocolVersion不一定要和 Client 完全一致但 Server 必须选择一个自己能支持的版本返回后续所有消息都按协商后的版本处理。2.3 Tools 的发现与调用过程MCP 中工具能力通过两个方法暴露tools/list客户端启动时调用获取全部工具列表包括工具名称、描述、输入参数 JSON Schema。tools/call客户端根据模型决策调用指定工具传入参数获取执行结果。工具调用流程完成后Server 返回一个结构化结果其中最核心的是content数组。每个 content item 可以是一个文本块也可以是图像块或资源链接。如果工具执行出错不要返回 JSON-RPC error而是把isError置为true并把错误信息放到 content 里。这个设计很反直觉但实际排查时非常有用——因为 AI 模型可以读取 content 中的错误信息决定下一步动作而 JSON-RPC error 更多表示协议层错误。下面是一个典型的tools/call响应{ jsonrpc: 2.0, id: 3, result: { content: [ {type: text, text: 5} ] } }3. 不依赖 SDK手写一个最小 MCP Server3.1 为什么新手应该手写一遍我知道很多教程上来就是“用 FastMCP 三行代码搞定”这确实快但对协议的理解会停留在表面。等你遇到客户端连不上、初始化顺序错误、日志污染 stdout 这类问题时还是会一脸懵。所以我一直建议至少手写一遍最小实现再回到 SDK。手写需要掌握的核心只有三件事读一行 JSON、处理请求、写一行 JSON。我用 Python 标准库实现不引入任何依赖。3.2 最小 Server 的核心代码创建一个minimal_server.pyimport sys import json import logging from datetime import datetime # 日志必须输出到 stderrstdout 只能用于协议消息 logging.basicConfig(levellogging.INFO, streamsys.stderr) def read_message(): line sys.stdin.readline() if not line: return None try: return json.loads(line) except json.JSONDecodeError as exc: logging.error(invalid json: %s, exc) return None def write_message(obj): sys.stdout.write(json.dumps(obj) \n) sys.stdout.flush() def handle_tools_list(): return { tools: [ { name: get_current_time, description: 返回当前时间, inputSchema: { type: object, properties: {} }, }, { name: add, description: 计算两个数字之和, inputSchema: { type: object, properties: { a: {type: number}, b: {type: number} }, required: [a, b] } } ] } def handle_tools_call(params): name params.get(name, ) arguments params.get(arguments, {}) or {} if name get_current_time: return { content: [ {type: text, text: datetime.now().isoformat()} ] } if name add: try: a float(arguments.get(a)) b float(arguments.get(b)) except (TypeError, ValueError): return { isError: True, content: [ {type: text, text: 参数必须都是数字} ] } return { content: [ {type: text, text: str(a b)} ] } return { isError: True, content: [ {type: text, text: funknown tool: {name}} ] } def handle_message(msg): # 没有 id 的是通知比如 notifications/initialized if id not in msg: logging.info(received notification: %s, msg.get(method)) return None method msg.get(method) params msg.get(params, {}) or {} if method initialize: return { jsonrpc: 2.0, id: msg[id], result: { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: { name: minimal-mcp-server, version: 0.1.0 } } } if method tools/list: return { jsonrpc: 2.0, id: msg[id], result: handle_tools_list() } if method tools/call: return { jsonrpc: 2.0, id: msg[id], result: handle_tools_call(params) } if method ping: return {jsonrpc: 2.0, id: msg[id], result: {}} return { jsonrpc: 2.0, id: msg[id], error: { code: -32601, message: fmethod not found: {method} } } def main(): while True: msg read_message() if msg is None: break response handle_message(msg) if response is not None: write_message(response) if __name__ __main__: main()这块代码里有一个非常关键的工程意识所有日志都输出到 stderrstdout 只输出协议 JSON。因为 stdio 模式靠 stdout 传数据如果你 print 一行调试信息到 stdout客户端就会把日志当成 JSON-RPC 消息解析直接报错。3.3 用脚本模拟客户端完整走一遍要验证这个 Server 是否正常我写了一个测试脚本模拟 MCP 客户端的完整交互过程import subprocess import json import time proc subprocess.Popen( [python, minimal_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) def send(obj): proc.stdin.write(json.dumps(obj) \n) proc.stdin.flush() def recv(): return json.loads(proc.stdout.readline()) # 1. 握手 send({ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: {name: test-client, version: 1.0.0} } }) print(initialize:, recv()) # 2. 初始化完成通知 send({ jsonrpc: 2.0, method: notifications/initialized }) # 3. 列出工具 send({ jsonrpc: 2.0, id: 2, method: tools/list, params: {} }) print(tools/list:, recv()) # 4. 调用工具 send({ jsonrpc: 2.0, id: 3, method: tools/call, params: { name: add, arguments: {a: 1, b: 2} } }) print(tools/call:, recv()) proc.terminate()运行后会看到initialize: {jsonrpc: 2.0, id: 1, result: {protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: {name: minimal-mcp-server, version: 0.1.0}}} tools/list: {jsonrpc: 2.0, id: 2, result: {tools: [...]}} tools/call: {jsonrpc: 2.0, id: 3, result: {content: [{type: text, text: 3.0}]}}看到这串输出说明你已经握住 MCP 的通信本质了。整个过程不需要任何第三方包就是“读一行 JSON、处理、写一行 JSON”的循环。3.4 手动实现时容易踩的坑手写版本虽然简单但有几个点值得单独拎出来说readline()会阻塞等待输入如果进程没有被客户端正常退出会一直卡住。你在本机测试时记得用terminate()结束子进程。参数解析时arguments可能是null或空对象。不能直接arguments.get(a)就完事要做空值兜底。协议错误码要遵循 JSON-RPC 2.0 规范-32700解析错误、-32600无效请求、-32601方法不存在、-32602参数无效、-32603内部错误。不要在生产环境的 Server 里用print做日志所有日志都要进 stderr。调试时可以2 server.log重定向查看。4. 用官方 SDK 快速构建生产可用的 MCP Server4.1 为什么最终要切到 SDK手写版本适合搞清楚协议但生产环境不适合长期用。因为一个完整的 MCP Server 还要处理会话生命周期、并发请求、错误边界、资源清理、更多传输方式这些用 SDK 能省掉大量重复工作。我推荐 Python 生态的mcp官方 SDK它提供了FastMCP高层封装。你只要定义函数、加装饰器就能自动生成tools/list、tools/call的协议响应SDK 底层会帮你做初始化协商、消息分发、参数校验。安装方式很简单pip install mcp4.2 用 FastMCP 实现同一个 Server创建一个fast_server.pyfrom datetime import datetime from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-server) mcp.tool() def get_current_time() - str: 返回当前时间ISO 格式。 return datetime.now().isoformat() mcp.tool() def add(a: float, b: float) - float: 计算两个数字之和。 return a b if __name__ __main__: mcp.run()就是这么简单。mcp.tool()会根据函数签名、类型注解、docstring 自动生成工具描述和 JSON Schema。你不需要手动维护tools/list的返回结构也不需要处理initialize握手。启动方式直接写python fast_server.pySDK 默认使用 stdio 传输以本地子进程方式运行。如果你希望把它作为一个远程服务可以用mcp.run(transportstreamable-http)但需要注意额外依赖和鉴权配置这里先不展开。4.3 配置到 MCP 客户端现在各种 MCP 客户端基本都支持通过 JSON 配置本地服务。以桌面客户端为例通常是在全局配置文件的mcpServers字段里加一个条目{ mcpServers: { demo-server: { command: python, args: [/absolute/path/to/fast_server.py] } } }配置完成后重启客户端它就会自动启动脚本并建立 stdio 连接。你在对话里让 AI “查看当前时间”或“计算 1 2”客户端就会调用对应工具。如果你之前手写过 Server再对比 SDK 版会明显感受到封装带来的效率提升同样功能代码量从上百行缩到十几行。5. 调试、排查与避坑指南5.1 用官方 Inspector 做交互式调试当我们手写协议层时可以用标准库测试但用 SDK 开发时我更推荐直接用官方 Inspector。运行方式是在项目目录里执行mcp dev fast_server.py它会启动一个 Web 页面通常自动打开本地服务。界面里能手动发送initialize、tools/list、tools/call等请求也能直接看到每个工具返回的原始 JSON。这对排查“工具为什么没被调用”“返回结构对不对”非常有帮助。Inspector 本质上也是一个 MCP Client你在它的输入框里填入工具参数它会把请求发给 Server并把响应展示出来。很多 SDK 能跑通但客户端连不上的问题用 Inspector 能快速定位是 Server 端的问题还是客户端的配置问题。5.2 常见报错和解决办法速查表现象常见原因解决办法客户端报“Failed to parse response”stdout 被业务日志污染把所有日志切到 stderrstdout 只输出 JSON客户端提示协议版本不支持Client 和 ServerprotocolVersion不匹配在initialize响应中返回客户端能接受的版本工具列表为空工具函数没有登录或者没有添加mcp.tool()检查装饰器是否生效函数是否在实例化之后注册调用工具时参数总是缺字段AI 客户端拿不到准确的 JSON Schema完善函数类型注解和 docstring必要时手写inputSchema回调返回错误但模型看不到信息工具内部抛异常SDK 可能封装成通用错误在函数内捕获异常返回isError: true和可读文本重启服务后客户端不生效客户端缓存了旧的 Server 连接断开重连或完全退出客户端再启动5.3 几个容易忽略的细节第一个细节是tools/call的返回值。MCP 客户端通常不会把异常等同于“工具执行失败”如果你想告诉模型“这个操作没成功”要把isError置为true。不设置isError时即使content里写了“失败”模型也可能当作正常结果。第二个细节是参数类型。AI 客户端有可能会传入字符串形式的数字比如{a: 1, b: 2}。在 FastMCP 中类型注解为float时SDK 一般会做转换但如果你的函数逻辑复杂还是建议在函数体内显式校验一次。第三个细节是超时。MCP 的 stdio 模式一般不会遇到网络超时但如果工具本身执行很长时间客户端可能会出现等待超时。耗时的任务建议拆成“提交任务 查询结果”两个工具或者以异步方式执行避免长时间阻塞消息循环。6. 如果我要上生产还需要考虑什么6.1 安全边界最重要MCP Server 本质上是“把系统能力开放给 AI 模型”。一个大模型不是值得信任的内部程序它可能被提示词注入影响可能产生错误参数。所有工具都要遵守最小权限原则能只读就不要给写权限能限定范围就不要给全量访问。之前有人喜欢搞“万能执行工具”让模型直接跑 shell 命令这种设计在本地 demo 里很酷但一旦暴露到公网或共享环境风险极高。建议对工具做白名单控制比如只允许操作指定目录、只允许操作指定数据库表所有危险操作都要有审计日志。6.2 远程传输与部署模式本地 stdio 模式适合个人开发和使用但如果你要把能力开放给团队或线上服务就要考虑 Streamable HTTP。官方 SDK 支持transportstreamable-http但远程部署还需要考虑鉴权、限流、CORS 等。一种常见做法是在 MCP Server 外面套一层 API 网关用 Token 认证再转发到内部服务。部署时建议单独拉起进程不要让 MCP Server 和其他 Web 服务混在一个进程里。因为 stdio 模式下 Server 的生命周期由客户端管理如果客户端崩溃子进程可能变成孤儿进程造成资源泄漏。6.3 协议版本演进MCP 协议还在快速迭代版本号变化比较频繁。你在落地时最好做一个“协议版本白名单”只支持自己验证过的版本。不同版本的客户端可能发送不同的消息结构如果 Server 无条件接受所有请求很容易在协议升级后踩坑。我的经验是在initialize响应里返回一个固定的、你充分测试过的protocolVersion而不是盲目跟随最新版本。客户端如果版本太旧就让它升级如果版本太新也先保持现有版本稳定运行。这比不断追新要可靠得多。最后再分享一点个人体会MCP 绝不复杂它的核心就是 JSON-RPC 加一套能力抽象最难的部分反而是工程细节——日志别污染 stdout、初始化顺序要对、错误返回要规范、危险操作要管控。如果你正在做 AI 工具链我建议先花两小时手写一遍最小 Server再切到官方 SDK。这短短两小时能帮你把以后遇到的所有连接问题都变成“可解释问题”而不是靠玄学改配置重启。我自己当初手写第一版的时候踩得最深的就是日志污染问题明明逻辑全对客户端就是解析失败。后来用重定向把 stderr 和 stdout 分开才明白 MCP 的 stdio 模式对输出纪律要求极高。从那以后我对“约定大于配置”这句话有了更真切的理解。希望这篇文章也能让你少走一点弯路。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进