MCP协议深度解析:AI Agent如何与真实世界连接
强烈推荐收藏MCP 协议深度解析AI 工具的「USB-C」是怎么设计的——JSON-RPC选型、四层架构、stdio vs HTTP、安全攻防给 Claude 写一套工具给 GPT 写一套适配给 DeepSeek 再写一套——这就是 MCP 要消灭的噩梦。Anthropic 在 2024 年底推出 Model Context Protocol号称 AI 领域的 USB-C。本文扒开协议层从 JSON-RPC 的选型逻辑讲到 Streamable HTTP 的协议升级再给出生产级 MCP Server 的完整实现。一、MCP 要解决什么问题1.1 没有 MCP 的世界# 给 Claude 写工具claude_tool(get_weather)defget_weather_for_claude(city):...# 给 GPT 写工具不同的格式openai_tool{type:function,function:{name:get_weather,parameters:{...}}}# 给 DeepSeek 写工具又是一种格式deepseek_tool{...}三个 AI三套代码。更糟的是新出一个 AI 框架 → 再写一套。这个「N×M 适配问题」在软件工程里被各种标准协议解决过HTTP、SQL、USBAI 工具接入是最后一个没有标准的领域。MCP 的解决方案很直接工具开发者写一个 MCP Server所有支持 MCP 的 AI 应用都能用。1.2 MCP vs Function Calling vs API维度Function Calling直接调 APIMCP标准化❌ 厂商锁定❌ 每次定制✅ 开放协议复用❌ 一个模型一套❌ 硬编码✅ 写一次处处用发现机制❌ 手动定义❌ 无✅ 自描述安全 代码内控制 代码内控制✅ 协议级隔离二、原理为什么选 JSON-RPC 2.0MCP 的通信层选择了 JSON-RPC 2.0 —— 一个 2010 年的老协议。为什么不是 gRPC不是 REST2.1 JSON-RPC vs gRPC# JSON-RPC 2.0 请求人类可读{jsonrpc:2.0,id:1,method:tools/call,params:{name:get_weather,arguments:{city:北京}}}# gRPC 请求需要先定义 .proto 文件# syntax proto3;# service WeatherService {# rpc GetWeather(WeatherRequest) returns (WeatherResponse);# }# message WeatherRequest { string city 1; }维度JSON-RPC 2.0gRPC可读性✅ 纯文本日志直接看❌ 二进制需要工具开发成本✅ 不需要 .proto❌ 需要定义→生成桩代码类型安全❌ 运行时校验✅ 编译期检查性能 文本序列化✅ Protobuf 二进制生态✅ 几乎所有语言都有库 需要 protocMCP 选 JSON-RPC 的根本原因降低接入门槛。一个 Python 开发者不需要学 Protobuf 就能写 MCP Server。性能在这里不是瓶颈——工具调用的频率远低于模型推理。2.2 JSON-RPC 2.0 消息结构# 四种消息类型# 1. 请求Request—— 客户端 → 服务端{jsonrpc:2.0,method:tools/list,id:1}# 2. 响应Response—— 服务端 → 客户端{jsonrpc:2.0,id:1,result:{tools:[...]}}# 3. 错误Error{jsonrpc:2.0,id:1,error:{code:-32601,message:Method not found}}# 4. 通知Notification—— 不需要响应{jsonrpc:2.0,method:notifications/initialized}三、机制MCP 四层架构逐层拆解3.1 完整架构图┌─────────────────────────────────────────┐ │ MCP Host │ │ Claude Desktop / Cursor / VS Code │ │ │ │ ┌─────────────────────────────────┐ │ │ │ MCP Client │ │ │ │ 1 个 Client ↔ 1 个 Server │ │ │ │ 管理连接生命周期 │ │ │ └─────────────┬───────────────────┘ │ └────────────────┼────────────────────────┘ │ JSON-RPC 2.0 │ ┌────────────────┼────────────────────────┐ │ ▼ │ │ MCP Server │ │ │ │ ┌──────────┐ ┌────────┐ ┌─────────┐ │ │ │ Resources│ │ Tools │ │ Prompts │ │ │ │(只读数据) │ │(可执行)│ │(指令模板)│ │ │ └──────────┘ └────────┘ └─────────┘ │ │ │ │ │ ┌───────▼───────┐ │ │ │ Data Source │ │ │ │(文件/DB/API) │ │ │ └───────────────┘ │ └─────────────────────────────────────────┘3.2 四大能力详解能力类型方向示例Resources只读数据Server → Client文件内容、数据库查询结果Tools可执行Client → Server发邮件、创建文件、调 APIPrompts指令模板Server → Client「审阅这段代码」、「翻译成中文」Sampling反向推理Server → ClientServer 请求 Host 端的 LLM 做推理3.3 生命周期① 启动阶段 Server 启动 → 等待 Client 连接 Client 连接 → 发送 initialize 请求 Server 返回能力列表tools/resources/prompts Client 发送 initialized 通知 → 进入运行态 ② 运行阶段 Client 发送 tools/call → Server 执行并返回 Client 发送 resources/read → Server 返回数据 ③ 关闭阶段 Client 断开连接 → Server 清理资源四、传输层stdio vs Streamable HTTP4.1 stdio本地进程通信Host 启动 MCP Server 作为子进程 ↓ 通过 stdin → 发 JSON-RPC 请求 通过 stdout ← 收 JSON-RPC 响应// Claude Desktop 配置本地 stdio 模式{mcpServers:{weather:{command:uv,args:[run,weather_server.py]}}}优点零网络开销、不需要认证、极简部署。缺点只能本地用、Host 崩溃 Server 一起死、无法远程访问。4.2 Streamable HTTP2025 年 3 月的协议升级原来的 HTTP SSE 模式有问题两个端点/sse监听 /messages发送负载均衡器不兼容。Streamable HTTP 把两者合并旧GET /sse持久连接 POST /messages发消息 新POST /mcp所有请求走同一个端点# 启动 HTTP 模式# mcp run weather_server.py --transport streamable-http --port 8080# 客户端连接{mcpServers:{weather:{url:http://your-server:8080/mcp,transport:streamable-http}}}关键改进每条请求可以独立鉴权Bearer Token天然兼容负载均衡器。五、实战生产级 MCP Serverfrommcp.server.fastmcpimportFastMCPimporthttpx mcpFastMCP(dev-assistant)# ── Tools ──mcp.tool()asyncdefsearch_stackoverflow(query:str,limit:int3)-str:搜索 Stack Overflow 相关问题asyncwithhttpx.AsyncClient()asclient:respawaitclient.get(https://api.stackexchange.com/2.3/search,params{order:desc,sort:relevance,intitle:query,site:stackoverflow,pagesize:limit},timeout10)dataresp.json()results[]foritemindata.get(items,[]):results.append(f【{item[score]}票】{item[title]}\n{item.get(link,)})return\n\n.join(results)ifresultselse未找到相关结果mcp.tool()defgenerate_commit_message(diff:str)-str:根据 git diff 生成规范的 commit message# 实际项目中这里调 LLMreturnf根据 diff 生成 commit message:{diff[:50]}...# ── Security Eggs ──importosmcp.tool()defread_file_safe(path:str)-str:安全读取文件限制目录和大小BASE_DIRos.path.expanduser(~/projects)full_pathos.path.normpath(os.path.join(BASE_DIR,path))# 路径遍历防护ifnotfull_path.startswith(BASE_DIR):return拒绝访问路径越界# 文件大小限制ifos.path.getsize(full_path)5*1024*1024:# 5MBreturn拒绝访问文件过大withopen(full_path,r)asf:returnf.read()if__name____main__:mcp.run()# 默认 stdio开发用# mcp.run(transportstreamable-http, port8080) # 生产用六、安全MCP Server 的防线6.1 路径遍历防护# ❌ 危险LLM 可以传 ../../etc/passwdmcp.tool()defread_file(path:str)-str:returnopen(path).read()# ✅ 安全限制在指定目录BASE/safe/datamcp.tool()defread_file_safe(path:str)-str:resolvedos.path.normpath(os.path.join(BASE,path))ifnotresolved.startswith(BASE):return拒绝returnopen(resolved).read()6.2 生产部署Docker 隔离FROM python:3.12-slim RUN pip install mcp httpx COPY server.py /app/ WORKDIR /app CMD [python, server.py]# 只读文件系统 无网络访问 → 就算 LLM 发疯也炸不了宿主机dockerrun--rm--read-only--networknone my-mcp-server6.3 速率限制importtimefromcollectionsimportdefaultdictclassRateLimiter:def__init__(self,max_calls:int60,window:int60):self.maxmax_calls self.windowwindow self.callsdefaultdict(list)defallow(self,tool_name:str)-bool:nowtime.time()self.calls[tool_name][tfortinself.calls[tool_name]iftnow-self.window]iflen(self.calls[tool_name])self.max:returnFalseself.calls[tool_name].append(now)returnTruelimiterRateLimiter(max_calls30)# 每个工具每分钟最多 30 次mcp.tool()asyncdefsearch_web(query:str)-str:ifnotlimiter.allow(search_web):return请求过于频繁请稍后再试# ...七、总结层级选型原因应用层MCP vs 自定义标准化 定制通信层JSON-RPC 2.0 vs gRPC可读性 低门槛 性能传输层stdio vs Streamable HTTP开发用 stdio生产用 HTTP安全层Docker 隔离 速率限制永远不信任 LLM 的输入MCP 的核心洞察AI 工具接入需要标准化就像 HTTP 标准化了网页传输。写一次 MCP Server所有 AI 都能用——这就是它的全部价值。 下一篇【AI Skill机制从工作流引擎到自动化执行框架】——从 CLAUDE.md 到社区 Skill 生态。标签#MCP #ModelContextProtocol #Agent #AI工具 #安全 #程序员必读