第10章:MCP协议深度解析——从协议原理到生产级Server实战

admin 2026-06-21 04:58:00 网络安全文章 来源:ZONE.CI 全球网 0 阅读模式

文章总结: 本文系统解析MCP协议的核心原理与生产实践,将其类比为AI领域的USB-C接口,统一连接外部系统。协议基于JSON-RPC实现双向通信,采用三层架构确保安全隔离。关键设计包括Server不可见完整对话、工具动态发现机制及Notification推送功能。详细解析了协议消息格式、错误处理及面试常见问题,为构建生产级Server提供实战指导。 综合评分: 87 文章分类: 技术标准,安全建设,解决方案,安全开发,安全工具


cover_image

第10章:MCP 协议深度解析 —— 从协议原理到生产级 Server 实战

原创

网络安全民工 网络安全民工

网络安全民工

2026年6月19日 17:21 天津

在小说阅读器读本章

去阅读

10.1 MCP 是什么?—— 用类比建立直觉

类比:USB-C 接口

在没有 USB-C 之前:

手机用 Micro-USB、电脑用 USB-A、显示器用 HDMI、硬盘用 Thunderbolt每种设备需要不同的线,接口混乱不堪

USB-C 统一了这一切:

一个接口,连接所有设备协议统一,但每个设备的「功能」不同

MCP 就是 AI 世界的 USB-C:

一个协议,连接所有外部系统协议统一(JSON-RPC),但每个 Server 的「工具」不同

2024年11月由 Anthropic 发布,迅速成为行业标准

MCP 解决了什么核心问题?

旧世界(MCP 之前):

┌──────┐ 自定义协议 ┌──────────┐│ LLM │──────────────→│ Google API││ │ 自定义协议 ├──────────┤│ │──────────────→│ GitHub API││ │ 自定义协议 ├──────────┤│ │──────────────→│ 数据库 │└──────┘ └──────────┘

每连接一个新系统,都要写新的适配代码!

新世界(MCP 之后):

┌──────┐ MCP ┌──────────┐│ LLM │──────────────→│ MCP Server│→ Google│ │ MCP ├──────────┤│ │──────────────→│ MCP Server│→ GitHub│ │ MCP ├──────────┤│ │──────────────→│ MCP Server│→ Database└──────┘ └──────────┘

一个协议,连接所有!新系统只需实现一个 MCP Server。

10.2 MCP 的三层架构 —— Host / Client / Server

│ │

MCP协议 │ MCP协议 │

(JSON-RPC)│ (JSON-RPC)│

▼ ▼

📊 架构示意

  ┌──────────────────────────────────────────────────────────┐  │                     Host Application                      │  │  ┌─────────────────┐  ┌─────────────────┐                │  │  │   MCP Client 1   │  │   MCP Client 2  │  ...          │  │  │  (1:1 连接)     │  │  (1:1 连接)     │                │  │  └────────┬────────┘  └────────┬────────┘                │  │           │                    │                          │  └───────────┼────────────────────┼──────────────────────────┘  ┌──────────────────┐  ┌──────────────────┐  │   MCP Server 1    │  │   MCP Server 2    │  │   文件系统 & Git   │  │     数据库        │  │                   │  │                   │  │  • read_file()    │  │  • query()        │  │  • write_file()   │  │  • insert()       │  │  • git_status()   │  │  • schema()       │  └──────────────────┘  └──────────────────┘

三个角色的职责:

Host(主机)

LLM 所在的应用程序(如 Claude Desktop、Claude Code)

创建和管理多个 Client 实例

控制安全策略和用户授权

聚合多个 Server 提供的上下文

Client(客户端)

Host 内部的组件,每个 Client 连接一个 Server

1:1 关系:一个 Client ↔ 一个 Server

处理协议握手和能力协商

路由协议消息

Server(服务器)

提供具体的上下文的进程/服务

通过 MCP 原语暴露能力

可以是本地进程(stdio)或远程服务(SSE)

关键设计原则(面试常问!):

Server 不能看到完整对话,也不能「看到」其他 Server

→ 安全隔离:对话历史留在 Host,Server 只收到必要信息

Server 应极其简单易构建

→ 复杂逻辑留给 Host,Server 只做一件事

Server 应高度可组合

→ 多个 Server 可以无缝组合使用

10.2.1 MCP 的设计哲学 —— 为什么不是 REST API?

面试官问:「为什么 MCP 不用 REST API,要用 JSON-RPC?」这是在考察

你是否理解协议设计的「为什么」。

▍ REST API 在 Agent 场景下的 3 个根本缺陷

缺陷 1:REST 是「请求-响应」模式,Agent 需要「双向推送」】

REST 只有 Client 主动请求 → Server 被动响应。

但 Agent 场景下 Server 可能主动推送:

工具列表变了(用户安装了新插件)→ Server 需要通知 Host

文件被外部修改了 → Server 需要主动告知

长时间运行的任务完成了 → Server 不能等 Host 来轮询

JSON-RPC 的 Notification(无 id 的消息)天然支持这种推送。

缺陷 2:REST 的 URL 路由是「名词」思维,Agent 需要「动词」思维

REST: GET /tools/weather → 获取天气工具「资源」

MCP: {“method”: “tools/call”, “params”: {“name”: “get_weather”}}

工具调用本质是「动作」,不是「资源」。用 URL 建模动作

导致接口膨胀(每个工具一个 endpoint),而 JSON-RPC 统一 method + params。

缺陷 3:REST 的状态协商靠 Header,JSON-RPC 靠 capabilities 对象

REST: Accept: application/vnd.api.v2+json (版本号藏在 Content-Type)

MCP: {“capabilities”: {“tools”: {…}, “sampling”: {…}}}

能力协商是整个 JSON 对象,可以动态嵌套、版本化、实验性功能标记。

比 HTTP Header 灵活得多。

▍ 为什么 Anthropic 选择了 JSON-RPC 而不是 gRPC?

gRPC 很好,但有 3 个门槛和 MCP 的设计目标冲突:

gRPC 需要 protobuf 编译 → MCP 想让「任何人 30 分钟写一个 Server」

  1. gRPC 强类型 → MCP 需要一定的灵活性(实验性功能、动态工具列表)

  2. gRPC 二进制 → MCP 需要可调试(stdio 下 cat 就能看到消息)

JSON-RPC 的 trade-off:类型安全弱,但开发体验和调试体验极好。

对于「让 100 万开发者快速构建 Server」这个目标,JSON-RPC 是正确选择。

▍ Server 为什么不能看到完整对话?

这是 MCP 最巧妙的安全设计之一。对话历史保留在 Host 内部,Server 只收到:

当前这次工具调用的参数

没有用户之前的提问

没有 LLM 的推理过程

没有其他 Server 的返回结果

为什么?因为如果 Server 能看到完整对话:

GitHub MCP Server 可能读到用户给 LLM 的数据分析请求(包含商业机密)

数据库 Server 可能窥探其他 Server 的查询结果

任何 Server 都可能通过对话历史推断用户意图和上下文

这个设计原则在面试中可以直接引用:

「MCP 的安全隔离是通过 Host 做对话历史过滤实现的,每个 Server 只收到自己需要的最少信息。」

10.3 MCP 的消息格式 —— JSON-RPC 2.0

MCP 使用 JSON-RPC 2.0 作为通信协议。

这是理解 MCP 调用流程的基础。

三种消息类型:

📝 对应的代码实现

REQUEST_EXAMPLERESPONSE_EXAMPLEERROR_RESPONSE_EXAMPLENOTIFICATION_EXAMPLE

# === Request(请求)===REQUEST_EXAMPLE = {    "jsonrpc": "2.0",    "id": 1,                    # 请求ID,用于匹配响应    "method": "tools/call",     # 方法名    "params": {                 # 参数        "name": "get_weather",        "arguments": {"city": "北京"}    }}# === Response(成功响应)===RESPONSE_EXAMPLE = {    "jsonrpc": "2.0",    "id": 1,                    # 对应请求的ID    "result": {                 # 成功结果        "content": [            {"type": "text", "text": "北京天气:晴,25°C"}        ]    }}# === Error Response(错误响应)===ERROR_RESPONSE_EXAMPLE = {    "jsonrpc": "2.0",    "id": 1,    "error": {        "code": -32602,         # 标准 JSON-RPC 错误码        "message": "Invalid params",        "data": {"detail": "缺少必填参数 'city'"}    }}# === Notification(通知 — 无需响应)===NOTIFICATION_EXAMPLE = {    "jsonrpc": "2.0",    "method": "notifications/tools/list_changed",  # 无 id 字段!    "params": {}}

关键区别:

Request/Response 有 id 字段(双向匹配)

Notification 没有 id(不需要响应)

10.3.1 JSON-RPC 生产实战 —— 面试官想知道你真正用过

▍ id 字段的关键作用 —— 不只是「匹配请求」

面试官问「id 字段是干嘛的?」如果你只回答「匹配请求和响应」,那就太浅了。更深层的答案:

并发调用保序 —— Agent 可能同时调用 3 个工具(get_weather、search_news、query_db),3 个请求顺序发、响应乱序回。

id 让你知道「这个返回是哪个请求的」。

幂等性支持 —— 用相同 id 重发请求,Server 应返回相同结果。

这在网络抖动场景下至关重要:Client 发了 tools/call(id=5),Server 执行了但响应丢了,Client 重发 id=5,Server 不应再次执行。

请求链追踪 —— 一个 Agent 动作链的每个子请求可以共享一个trace_id(通过 params 传),但各自有独立 id 做匹配。

▍ JSON-RPC 错误码 —— 不只是 -32601

面试官让你列几个 JSON-RPC 标准错误码:

-32700 Parse error ← JSON 格式无效(写错了花括号)

-32600 Invalid Request ← 不是合法的 Request 对象

-32601 Method not found ← 调了不存在的工具

-32602 Invalid params ← 参数类型不对或缺少必填字段

-32603 Internal error ← Server 内部炸了(兜底用)

MCP 在标准码之上还扩展了:

-32000 ~ -32099: Server 层面错误(如工具执行超时、权限不足)

面试实用技巧:提到错误码时举一个生产中的真实例子——

「线上遇到过:Agent 传了字符串 ‘5’ 而不是数字 5,Server 返回 -32602,我们在 Host 层加了类型自动转换来修复。」

▍ Notification 的生产应用

Notification 不是设计上的点缀,而是 MCP 支持「长连接 Server」的关键:

工具列表变更通知:用户装了新 MCP 插件 → Server 主动通知 Host

资源更新通知:文件被修改 → Server 推送给 Host 刷新上下文

进度通知:长时间工具调用(如大文件分析)→ Server 上报进度

面试中提一句「Notification 机制让 MCP 不只是请求-响应,而是事件驱动的双向通信」会显得你对协议理解很深。

10.4 MCP 的核心方法(Primitives)—— 按调用流程

完整的 MCP 调用生命周期:

📊 架构示意

  ┌────────────────────────────────────────────┐  │  Phase 1: 初始化 (Initialize)              │  │  ┌──────────────────────────────────────┐  │  │  │ Client                    Server     │  │  │  │   │── initialize ──────────→│        │  │  │  │   │←─ capabilities + info ─│        │  │  │  │   │── initialized ─────────→│        │  │  │  └──────────────────────────────────────┘  │  ├────────────────────────────────────────────┤  │  Phase 2: 发现工具 (Tool Discovery)         │  │  ┌──────────────────────────────────────┐  │  │  │   │── tools/list ──────────→│        │  │  │  │   │←─ [工具1, 工具2, ...] ─│        │  │  │  └──────────────────────────────────────┘  │  ├────────────────────────────────────────────┤  │  Phase 3: 调用工具 (Tool Invocation)        │  │  ┌──────────────────────────────────────┐  │  │  │   │── tools/call ──────────→│        │  │  │  │   │←─ tool_result ────────│        │  │  │  └──────────────────────────────────────┘  │  ├────────────────────────────────────────────┤  │  Phase 4: 关闭 (Teardown)                  │  │  ┌──────────────────────────────────────┐  │  │  │   │── 关闭连接 ─────────────→│        │  │  │  └──────────────────────────────────────┘  │  └────────────────────────────────────────────┘

MCP 的三大核心原语(Server 能力):

Tools(工具)

让 LLM 可以「做」事情

示例:查询数据库、调用API、执行计算

对应方法:tools/list, tools/call

Resources(资源)

让 LLM 可以「读」数据

示例:读取文件、获取数据库schema

对应方法:resources/list, resources/read

Prompts(提示模板)

预定义的提示词模板

示例:「代码审查模板」「Bug报告模板」

对应方法:prompts/list, prompts/get

Sampling(采样)—— Client 能力

Server 可以请求 Host 的 LLM 生成内容

示例:Server 需要 LLM 帮助分类或总结

10.5 Capability Negotiation(能力协商)—— 面试重点!

MCP 不是「一刀切」的协议。Client 和 Server 在连接初始化时

互相声明自己的「能力」,只有双方都支持的功能才可用。

Server 声明的能力(capabilities):{"capabilities": {"tools": {"listChanged": true}, // 支持工具,且工具列表可动态变化"resources": {"subscribe": true}, // 支持资源,且支持订阅更新"prompts": {"listChanged": false}, // 支持提示模板,列表静态不变"logging": {} // 支持日志输出}}Client 声明的能力:{"capabilities": {"sampling": {}, // 支持 LLM 采样"roots": {"listChanged": true}, // 支持根目录声明"experimental": {"featureX": {}} // 实验性功能}}

为什么需要能力协商?

向后兼容:新版本可降级到老版本的功能集

渐进增强:功能可以逐步添加,不破坏现有 Server

安全隔离:Host 可以拒绝某些能力(如不启用 sampling)

类比:两人见面先握手说「我会英语、中文、日语」

然后选择双方都会的语言交流。

10.6 MCP 传输层 —— stdio vs SSE

MCP 内置两种传输方式:

stdio(标准输入/输出)

适用:本地进程、命令行工具

原理:Client 启动 Server 作为子进程

通信:通过 stdin/stdout 交换 JSON-RPC 消息

优点:简单、安全(进程隔离)

示例:本地文件系统 Server、本地数据库 Server

SSE(Server-Sent Events)

适用:远程服务、Web 部署

原理:HTTP POST(Client→Server)+ SSE(Server→Client)

优点:支持远程部署、服务端推送

示例:云端 API Server、第三方服务集成

还支持 2025 年新增的 Streamable HTTP 传输方式。

10.6.1 传输层选型决策树 —— 三种方式怎么选?

面试官问「stdio 和 SSE 有什么区别?你项目里用哪个?」

—— 这不是考概念,是在考你有没有做过技术选型。

▍ 决策树(从工程角度)

你的 MCP Server 部署在哪里?

├─ 简单场景,不需要流式

│ └─ Streamable HTTP ← 2025 新增

│ 理由:部署最简单(只需一个 POST endpoint)

│ 适合:简单的 API 封装、无需推送通知的 Server

└─ 复杂场景,需要服务端推送

└─ SSE

理由:支持 Server 主动推送(工具列表变更、进度更新)

需要:反向代理支持(Nginx 关掉 buffering)

注意:SSE 走 HTTP/1.1,不支持 HTTP/2 多路复用

📊 架构示意

  ├─ 本地(和 Host 在同一台机器)  │   └─ stdio ← 唯一正解  │       理由:零网络开销、进程隔离天然安全、不需要额外运维  │       不需要开端口、不需要 HTTPS 证书、不需要鉴权逻辑  │  └─ 远程(Server 和 Host 不在同一台机器)

▍ stdio 的生产陷阱(你可能踩过的坑)

进程管理 —— Client 启动 Server 为子进程,Server 崩溃了谁来重启?

答:Host 负责。Claude Desktop 会自动重启崩溃的 Server。

→ 你实现 MCP Client 时必须处理子进程的生命周期。

大结果传输 —— stdin/stdout 默认有 buffer 限制(通常 64KB)。

如果你的工具返回 10MB 的 JSON,stdout 可能阻塞。

→ MCP 2025-06-18 版本增加了分页支持(tools/call 返回 cursor)。

stdin 被污染 —— 如果 Server 的代码里 print() 了调试日志到 stdout,

Client 会把它当成 JSON-RPC 消息来解析 → 解析失败。

→ 严格规则:日志走 stderr,协议消息走 stdout。

▍ 远程 Server 的安全模型

stdio 下不需要鉴权(信任本地进程),但远程 Server 必须考虑:

传输层安全 → HTTPS(和任何 Web API 一样)

身份认证 → API Key / OAuth Token(SSE 首请求的 Header 带)

工具级授权 → 不是所有的 tools 对所有用户可用

Rate Limiting → 防止 Agent 死循环调用工具

面试中可以提:「简单场景用 API Key,企业场景用 OAuth + 工具级 RBAC。」

📝 对应的代码实现

handle_requestdemo_mcp_full_flowSimulatedMCPServer

# === Request(请求)===class SimulatedMCPServer:    """模拟 MCP Server —— 完整演示 MCP 的核心交互流程。    实现了 Tools 原语的完整调用链路。    """    def __init__(self, name: str, version: str):        self.name = name        self.version = version        self.capabilities = {            "tools": {"listChanged": True},            "resources": {"subscribe": False},            "prompts": {"listChanged": False},        }        # 注册工具        self.tools = {            "get_weather": {                "name": "get_weather",                "description": "查询指定城市的天气信息",                "inputSchema": {                    "type": "object",                    "properties": {                        "city": {                            "type": "string",                            "description": "城市名称"                        }                    },                    "required": ["city"],                },                "handler": self._handle_weather,            },            "search_docs": {                "name": "search_docs",                "description": "在知识库中搜索文档",                "inputSchema": {                    "type": "object",                    "properties": {                        "query": {"type": "string", "description": "搜索关键词"},                        "max_results": {"type": "integer", "default": 5},                    },                    "required": ["query"],                },                "handler": self._handle_search,            },        }    def _handle_weather(self, args: dict) -> dict:        """处理天气查询工具调用。"""        weather = {"北京": "晴 25°C", "上海": "多云 28°C",                   "深圳": "阵雨 30°C"}        city = args.get("city", "")        return {            "content": [{"type": "text",                        "text": weather.get(city, f"未找到{city}的天气数据")}],            "isError": city not in weather,        }    def _handle_search(self, args: dict) -> dict:        """处理文档搜索工具调用。"""        query = args.get("query", "")        max_results = args.get("max_results", 5)        mock_results = [f"文档{i}: 关于'{query}'的内容片段..."                         for i in range(min(max_results, 3))]        return {            "content": [{"type": "text", "text": "\n".join(mock_results)}],            "isError": False,        }    def handle_request(self, request: dict) -> dict:        """MCP Server 的消息路由(核心!)。        Args:            request: JSON-RPC 请求。        Returns:            JSON-RPC 响应。        """        req_id = request.get("id")        method = request.get("method", "")        params = request.get("params", {})        # === Phase 1: initialize ===        if method == "initialize":            return {                "jsonrpc": "2.0",                "id": req_id,                "result": {                    "protocolVersion": "2025-06-18",                    "capabilities": self.capabilities,                    "serverInfo": {                        "name": self.name,                        "version": self.version,                    },                },            }        # === Phase 2: tools/list ===        if method == "tools/list":            tool_list = []            for t in self.tools.values():                tool_list.append({                    "name": t["name"],                    "description": t["description"],                    "inputSchema": t["inputSchema"],                })            return {                "jsonrpc": "2.0",                "id": req_id,                "result": {"tools": tool_list},            }        # === Phase 3: tools/call ===        if method == "tools/call":            tool_name = params.get("name", "")            tool_args = params.get("arguments", {})            tool = self.tools.get(tool_name)            if tool is None:                return {                    "jsonrpc": "2.0",                    "id": req_id,                    "error": {                        "code": -32601,                        "message": f"未知工具: {tool_name}"                    },                }            result = tool["handler"](tool_args)            return {                "jsonrpc": "2.0",                "id": req_id,                "result": result,            }        # 未知方法        return {            "jsonrpc": "2.0",            "id": req_id,            "error": {                "code": -32601,                "message": f"未知方法: {method}"            },        }def demo_mcp_full_flow():    """演示 MCP 的完整调用流程。"""    print("=" * 60)    print("  MCP 完整调用流程演示")    print("=" * 60)    server = SimulatedMCPServer("WeatherServer", "1.0.0")    # Phase 1: 初始化 + 能力协商    print("\n  [Phase 1] 初始化与能力协商")    init_request = {        "jsonrpc": "2.0",        "id": 1,        "method": "initialize",        "params": {            "protocolVersion": "2025-06-18",            "capabilities": {"sampling": {}},            "clientInfo": {"name": "ClaudeCode", "version": "1.0.60"},        },    }    print(f"  Client → Server: {init_request['method']}")    init_response = server.handle_request(init_request)    caps = init_response["result"]["capabilities"]    print(f"  Server → Client: 协议版本 {init_response['result']['protocolVersion']}")    print(f"  Server 能力: {list(caps.keys())}")    # 发送 initialized 通知    print(f"  Client → Server: initialized (通知,无需响应)")    # Phase 2: 发现工具    print("\n  [Phase 2] 发现工具列表")    list_request = {        "jsonrpc": "2.0",        "id": 2,        "method": "tools/list",        "params": {},    }    print(f"  Client → Server: {list_request['method']}")    list_response = server.handle_request(list_request)    tools = list_response["result"]["tools"]    for t in tools:        print(f"    工具: {t['name']} - {t['description']}")    # Phase 3: 调用工具    print("\n  [Phase 3] 调用工具")    call_requests = [        ("tools/call", {"name": "get_weather", "arguments": {"city": "北京"}}),        ("tools/call", {"name": "search_docs", "arguments": {"query": "MCP协议"}}),        ("tools/call", {"name": "get_weather", "arguments": {"city": "火星"}}),    ]    for method, params in call_requests:        req = {"jsonrpc": "2.0", "id": 3, "method": method, "params": params}        print(f"  Client → Server: {params['name']}({params['arguments']})")        response = server.handle_request(req)        if "result" in response:            content = response["result"]["content"][0]["text"]            is_error = response["result"].get("isError", False)            flag = "⚠️ 错误:" if is_error else "  ✅"            print(f"  Server → Client: {flag} {content}")        else:            print(f"  Server → Client: ❌ {response['error']['message']}")

10.7 MCP vs Function Calling vs A2A —— 三者边界在哪?

面试中经常混淆这三个概念,把它们放在一起对比会显得你有系统思维。

📊 架构示意

  ┌──────────────┬──────────────────┬───────────────────┬───────────────────┐  │     维度      │   MCP             │  Function Calling  │      A2A          │  ├──────────────┼──────────────────┼───────────────────┼───────────────────┤  │ 本质          │ 通信协议          │ LLM 能力            │ 通信协议            │  │ 标准化        │ 开放标准          │ 厂商自有实现        │ 开放标准            │  │ 解决的问题     │ LLM ↔ 工具        │ LLM 决策调用工具    │ Agent ↔ Agent      │  │ 连接模式      │ Client↔Server    │ 直接在 API 调用中    │ Agent↔Agent        │  │ 工具发现      │ tools/list 动态   │ 每次请求手动传入     │ AgentCard 发现      │  │ 服务器生态    │ 独立部署 MCP Server│ 无独立服务器概念     │ Agent 作为服务       │  │ 跨模型支持    │ 是(协议层面)     │ 取决于厂商           │ 是(协议层面)       │  └──────────────┴──────────────────┴───────────────────┴───────────────────┘

三者的协作关系(面试高分答案):

“在完整的 Agent 系统中,三者各司其职:

MCP 负责「Agent 怎么连接工具」—— 协议层面的标准化

Function Calling 负责「LLM 怎么决定调哪个工具」—— 模型能力

A2A 负责「Agent 之间怎么协作」—— Agent-to-Agent 通信

它们不是竞争对手,而是一个协议栈的三个层次:

FC (模型层) → MCP (工具层) → A2A (多Agent层)”

实际架构中的配合流程:

MCP Client 通过 tools/list 获取所有工具定义

将工具定义转换为 OpenAI function 格式传给 LLM

LLM(通过 Function Calling)返回 function call

MCP Client 通过 tools/call 执行工具

将结果返回给 LLM 继续推理

(如果是 Multi-Agent 场景,第 2 步的调用可能通过 A2A 来自另一个 Agent)

▍ 面试陷阱:「MCP 和 Function Calling 是竞争关系吗?」

错误回答:「对,MCP 就是要替代 Function Calling」

正确回答:「不是竞争。MCP 是标准化 LLM 和外部工具的连接方式,

而 Function Calling 是 LLM 内部的决策机制。MCP Server 可以

同时对接 OpenAI 的 FC 和 Anthropic 的 Tool Use。」

补一句可以加分:「我做过实验,同一个 MCP Server 给 GPT-4o 和

Claude 都能用——这就是协议标准化的价值。」

10.8 本章总结

核心要点回顾:

MCP 的设计哲学(面试必问!)

为什么是 JSON-RPC 而不是 REST API?

→ REST 是请求-响应、名词路由、Header 协商;MCP 需要双向推送、

动词路由、能力协商对象 → JSON-RPC 更合适

类比:MCP = AI 世界的 USB-C —— 一个协议连接所有外部系统

Client-Host-Server 三层架构

核心安全设计:Server 看不到完整对话 → 对话历史留在 Host

Host 做安全隔离,每个 Server 只收到最少必要信息

JSON-RPC 生产实战

id 字段的 3 个作用:并发保序 / 幂等性 / 请求链追踪

5 个标准错误码:-32700 / -32600 / -32601 / -32602 / -32603

Notification:让 MCP 从请求-响应升级为事件驱动

核心原语

Tools:让 LLM「做事」(tools/list → tools/call)

Resources:让 LLM「读数据」

Prompts:预定义提示模板

Sampling:Server 请求 Host 的 LLM 生成

能力协商(Capability Negotiation)

初始化时双方声明 capabilities 对象

只有双方都支持的功能才可用

面试答法:「像两个人见面先报自己会的语言,选交集交流」

传输层选型

本地 → stdio(零网络开销、进程隔离、无需鉴权)

远程简单 → Streamable HTTP(一个 POST endpoint)

远程复杂 → SSE(支持服务端推送,需 Nginx 关 buffering)

stdio 三大陷阱:进程管理 / 大结果传输 / stdin 污染

MCP vs Function Calling vs A2A

不是竞争,是协议栈三层:

FC (模型层) → MCP (工具层) → A2A (多Agent层)

同一 MCP Server 可同时对接 OpenAI FC 和 Anthropic Tool Use

面试速记(完整版):

“请解释 MCP 的工作原理”

→ 三层架构(Host/Client/Server) + JSON-RPC 通信 + 能力协商

→ 生命周期:Initialize → Capability Negotiation →Tools/Resources/Prompts → Teardown

→ 设计哲学:不是 REST API 因为需要双向推送和动词路由

→ 安全:Server 看不到完整对话,Host 做信息过滤

→ 传输:本地 stdio,远程 SSE/Streamable HTTP

→ MCP + Function Calling + A2A 是互补关系,不是竞争

📝 对应的代码实现

=== Request(请求)===

if __name__ == "__main__":    print("╔══════════════════════════════════════════════════════╗")    print("║  第10章:MCP 协议原理与调用全流程                      ║")    print("║  JSON-RPC · 原语 · 能力协商 · stdio/SSE             ║")    print("╚══════════════════════════════════════════════════════╝")    print("\n▶ 10.1-10.3 MCP 架构与消息格式")    print("-" * 50)    print("MCP = Client-Host-Server 架构")    print("通信协议: JSON-RPC 2.0")    print("消息类型: Request / Response / Notification")    print()    print("Request 示例:")    import pprint    pprint.pprint(REQUEST_EXAMPLE, width=60)    print()    print("Response 示例:")    pprint.pprint(RESPONSE_EXAMPLE, width=60)    print("\n▶ 10.6 MCP 完整调用流程演示")    demo_mcp_full_flow()    print("\n▶ 10.7 MCP vs Function Calling 对比")    print("-" * 50)    comparisons = [        ("本质", "MCP: 通信协议", "FC: LLM 能力"),        ("标准化", "MCP: 开放标准", "FC: 厂商自有实现"),        ("工具发现", "MCP: tools/list 动态发现", "FC: 每次手动传入"),        ("可组合性", "MCP: 多 Server 组合", "FC: 依赖应用代码"),        ("关系", "MCP 管理连接", "FC 管理执行决策"),    ]    for dim, mcp, fc in comparisons:        print(f"  {dim:12s}  {mcp:30s}  {fc}")    print("\n✅ 第10章完成!")


免责声明:

本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。

任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。

本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我

本文转载自:网络安全民工 网络安全民工 网络安全民工《第10章:MCP 协议深度解析 —— 从协议原理到生产级 Server 实战》

评论:0   参与:  0