MCP协议深度拆解:从手写最小Server到生产实践
2026/9/9 19:06:32 网站建设 项目流程

MCP Server 这波热度,基本是从 AI 工具链里炸出来的。说白了,以前你要让 AI 助手去查数据库、调接口、读文件,每个接入都要定制一套工具调用方式,各家有各家的规范,做起来非常碎。MCP(Model 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 插件、智能助手。它负责和用户交互,也是整个流程的发起方。
  • Client:Host 内置的协议客户端,负责连接 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(通知)。

一个请求消息至少包含jsonrpcidmethodparams四个字段;响应则必须包含jsonrpcid以及resulterror;通知和请求很像,但它不包含id,也不需要任何响应。下面是一个最基本的请求:

{"jsonrpc": "2.0", "id": 1, "method": "ping", "params": {}}

对应的响应:

{"jsonrpc": "2.0", "id": 1, "result": {}}

传输层方面,MCP 目前最常见的两种方式是:

  • stdio:Server 作为本地子进程启动,通过标准输入 stdout/stdin 传输换行分隔的 JSON 消息。
  • Streamable HTTP:Server 作为远程 HTTP 服务,通过 POST 请求发送 JSON-RPC 消息,并可选支持 SSE 流式返回。

对新手来说,最友好的切入点是 stdio。你只需要把一个 Python/Node 进程跑起来,输入输出全部走标准输入输出,不需要考虑端口、Token、鉴权这些东西。

2.2 从握手到工具调用的完整生命周期

MCP 的通信不是上来就随便调,它有严格的握手流程:

  1. Client 发送initialize请求,带上自己的协议版本、能力声明、客户端信息。
  2. Server 返回initialize响应,返回它选定的协议版本、自身能力、服务器信息。
  3. Client 再发送一个notifications/initialized通知,告诉 Server“我已经知道你的能力了,现在可以正常工作了”。
  4. 只有完成前三步之后,Client 才能调用tools/listtools/callresources/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 Server

3.1 为什么新手应该手写一遍

我知道很多教程上来就是“用 FastMCP 三行代码搞定”,这确实快,但对协议的理解会停留在表面。等你遇到客户端连不上、初始化顺序错误、日志污染 stdout 这类问题时,还是会一脸懵。所以我一直建议:至少手写一遍最小实现,再回到 SDK。

手写需要掌握的核心只有三件事:读一行 JSON、处理请求、写一行 JSON。我用 Python 标准库实现,不引入任何依赖。

3.2 最小 Server 的核心代码

创建一个minimal_server.py

import sys import json import logging from datetime import datetime # 日志必须输出到 stderr,stdout 只能用于协议消息 logging.basicConfig(level=logging.INFO, stream=sys.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": f"unknown 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": f"method 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()

这块代码里有一个非常关键的工程意识:所有日志都输出到 stderr,stdout 只输出协议 JSON。因为 stdio 模式靠 stdout 传数据,如果你 print 一行调试信息到 stdout,客户端就会把日志当成 JSON-RPC 消息解析,直接报错。

3.3 用脚本模拟客户端完整走一遍

要验证这个 Server 是否正常,我写了一个测试脚本,模拟 MCP 客户端的完整交互过程:

import subprocess import json import time proc = subprocess.Popen( ["python", "minimal_server.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) 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 Server

4.1 为什么最终要切到 SDK

手写版本适合搞清楚协议,但生产环境不适合长期用。因为一个完整的 MCP Server 还要处理会话生命周期、并发请求、错误边界、资源清理、更多传输方式,这些用 SDK 能省掉大量重复工作。

我推荐 Python 生态的mcp官方 SDK,它提供了FastMCP高层封装。你只要定义函数、加装饰器,就能自动生成tools/listtools/call的协议响应,SDK 底层会帮你做初始化协商、消息分发、参数校验。

安装方式很简单:

pip install mcp

4.2 用 FastMCP 实现同一个 Server

创建一个fast_server.py

from 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.py

SDK 默认使用 stdio 传输,以本地子进程方式运行。如果你希望把它作为一个远程服务,可以用mcp.run(transport="streamable-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 页面,通常自动打开本地服务。界面里能手动发送initializetools/listtools/call等请求,也能直接看到每个工具返回的原始 JSON。这对排查“工具为什么没被调用”“返回结构对不对”非常有帮助。

Inspector 本质上也是一个 MCP Client,你在它的输入框里填入工具参数,它会把请求发给 Server,并把响应展示出来。很多 SDK 能跑通但客户端连不上的问题,用 Inspector 能快速定位是 Server 端的问题,还是客户端的配置问题。

5.2 常见报错和解决办法速查表

现象常见原因解决办法
客户端报“Failed to parse response”stdout 被业务日志污染把所有日志切到 stderr,stdout 只输出 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 支持transport="streamable-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 模式对输出纪律要求极高。从那以后,我对“约定大于配置”这句话有了更真切的理解。希望这篇文章也能让你少走一点弯路。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询