MCP 协议到底是什么？传输层 × 消息格式 × 工具注册 逐层拆解

MCP 是什么

2024 年 11 月，Anthropic 发布了 Model Context Protocol（MCP）——一个开放的、标准化的协议，旨在统一 AI 模型与外部工具、数据源之间的通信方式。MCP 充当了 "AI 的 USB-C"——让任何一个符合标准的服务器插入后，AI 就可以立刻使用它的工具。截至 2026 年，MCP 已经拥有 9700 万+ 月下载量和 10000+ 个注册服务器。

协议的核心哲学是 "一次对接，处处可用"。开发者只需要编写一次 MCP 服务器，所有兼容的 AI 应用（Claude Desktop、Cline、OpenCode 等）都可以无缝复用。这解决了此前 Agent 生态中 "每个框架都要重写一遍工具" 的碎片化痛点。MCP 严格遵循 Host-Client-Server 模型：Host 负责用户的交互环境、授权决策和连接管理；Client 是与 Server 建立 1:1 通道的协议客户端；Server 是工具、资源和提示模板的实际提供者。

在 OpenCode 这样的 AI 编码工具中，MCP 的配置以声明式 JSON 文件管理。开发者可以在 opencode.json 中定义多个 MCP 服务器：本地服务器通过命令和参数启动子进程，远程服务器通过 URL 建立 HTTP 连接。OpenCode 的 MCP 配置系统（packages/core/src/config/mcp.ts）使用 Effect Schema 对服务器配置进行类型安全的校验，包括 command、cwd、environment、timeout 等字段的完整约束。

查看代码 ▸ (\"ConfigV2.MCP.Remote\")({\n type: Schema.Literal(\"remote\"),\n url: Schema.String,\n headers: Schema.Record(Schema.String, Schema.String).pipe(Schema.optional),\n oauth: Schema.Union([OAuth, Schema.Literal(false)]).pipe(Schema.optional),\n disabled: Schema.Boolean.pipe(Schema.optional),\n timeout: PositiveInt.pipe(Schema.optional),\n}) {}\n\nexport const Server = Schema.Union([Local, Remote]).pipe(Schema.toTaggedUnion(\"type\"))','OpenCode MCP 配置 Schema')">查看代码 ▸

三层架构

MCP 协议采用经典的三层架构：Transport（传输层） → Session（会话层） → Application（应用层）。核心是 mcp/shared/session.py 中的 BaseSession，它定义了 MCP 会话的完整生命周期。Host 是用户环境（如 Claude Desktop），Client 是与 Server 建立 1:1 连接的协议客户端，Server 提供工具、资源和提示模板。

在三层架构的底部，Transport 层负责原始字节流的读写。传输层之上是 Session 层，它管理消息的路由、请求 ID 的分配、挂起请求的追踪以及通知的分发。最顶部是 Application 层，处理具体的业务逻辑：工具调用、资源读取、提示模板填充。这种分层设计让每一层都可以独立替换——你可以更换传输方式而无需修改工具逻辑，也可以扩展应用功能而不影响底层通信。

值得注意的是，MCP 的 Capability Negotiation（能力协商） 机制发生在会话初始化阶段。Client 和 Server 在握手时交换各自支持的特性（工具、资源、提示、日志、采样等），然后才能进行后续通信。这确保了即使 Server 和 Client 版本不同，也能在双方都支持的子集上正常工作。在 OpenCode 中，MCP 服务器的状态管理通过 packages/tui/src/feature-plugins/sidebar/mcp.tsx 的侧边栏组件实时展示，每种状态（connected、disabled、failed、needs_auth）使用不同颜色的指示点呈现。

查看代码 ▸ \n {(item) => (\n \n •\n \n {item.name}{\" \"}\n \n \n Connected\n \n {item.error}\n \n \n \n \n \n )}\n','OpenCode MCP 侧边栏')">查看代码 ▸

传输层

MCP 的传输层支持三种模式：stdio（子进程标准 I/O，适合本地）、SSE（Server-Sent Events，适合远程 HTTP）、Streamable HTTP（新标准，更灵活）。在 Python SDK 中，传输层实现在 mcp/client/stdio.py、mcp/client/sse.py 和 mcp/client/streamable_http.py。服务端对应 mcp/server/stdio.py 和 mcp/server/sse.py。

Stdio 传输是最常用的本地传输方式。Python 客户端通过 asyncio.create_subprocess_exec 启动 TypeScript 子进程，将 JSON-RPC 消息写入 stdin，从 stdout 读取响应。这种方式的优势在于零网络开销、低延迟，适合单机部署。关键在于子进程管理——使用 async with 上下文管理器确保进程正确清理，且 stderr 需要单独捕获以排查错误。

SSE（Server-Sent Events） 传输适合远程场景。客户端先通过 GET /sse 建立单向事件流（接收服务端的 session ID 和通知），然后通过 POST /messages 发送请求。2025 年引入的 Streamable HTTP 是 SSE 的进化版，它使用更标准的 HTTP 流式响应，不需要维持长连接 SSE 通道，在网络代理和负载均衡场景下兼容性更好。OpenCode 的远程 MCP 调用实现（packages/opencode/src/tool/mcp-websearch.ts）展示了如何通过 HTTP POST 发送 JSON-RPC tools/call 请求并解析 SSE 或 JSON 响应流。

查看代码 ▸ (\n http: HttpClient.HttpClient,\n url: string,\n tool: string,\n args: Schema.Struct,\n value: Schema.Struct.Type,\n timeout: Duration.Input,\n) => Effect.gen(function* () {\n const request = yield* HttpClientRequest.post(url).pipe(\n HttpClientRequest.accept("application/json, text/event-stream"),\n HttpClientRequest.schemaBodyJson(McpRequest(args))({\n jsonrpc: "2.0" as const,\n id: 1 as const,\n method: "tools/call" as const,\n params: { name: tool, arguments: value },\n }),\n )\n const response = yield* HttpClient.filterStatusOk(http).execute(request).pipe(\n Effect.timeoutOrElse({\n duration: timeout,\n orElse: () => Effect.die(new Error(`${tool} request timed out`)),\n }),\n )\n const body = yield* response.text\n return yield* parseResponse(body)\n})','OpenCode 远程 MCP 调用')">查看代码 ▸

消息层

MCP 的消息层基于 JSON-RPC 2.0 协议，定义了三种消息类型：Request（请求）、Response（响应） 和 Notification（通知）。每条消息都有 jsonrpc 版本、id、method 和 params 字段。核心方法包括 tools/list（列出工具）、tools/call（调用工具）、resources/list（列出资源）和 prompts/get（获取提示模板）。

Request 需要对方回响应（有 id），Notification 不需要（无 id）。这种设计参考了经典的 RPC 模式，但增加了异步通知的能力——Server 可以主动向 Client 发送资源变更通知（notifications/resources/list_changed），Client 也可以向 Server 发送日志消息（notifications/message）。响应中的 error 字段遵循 JSON-RPC 2.0 标准错误码（-32700 解析错误、-32600 无效请求、-32601 方法未找到等），外加 MCP 自定义的错误码。

在 OpenCode 的 TypeScript 实现中，JSON-RPC 消息通过 Effect Schema 进行类型安全校验。例如 packages/opencode/src/tool/mcp-websearch.ts 中的 McpRequest 函数就定义了一个严格的 Schema：要求 jsonrpc 必须是字面量 "2.0"、id 是字面量 1、method 是字面量 "tools/call"，params 需要包含 name 和类型化的 arguments。这种编译时校验在运行时通过 Schema 解码器自动转换成 JSON 序列化，避免了手写序列化代码中的类型错误。

查看代码 ▸ 查看代码 ▸

工具注册与发现

服务端通过 ToolManager（mcp/server/mcpserver/tools/tool_manager.py）管理工具的注册与发现。工具注册使用装饰器模式：@server.tool("name", description="...")。每个工具包含 name、description 和 inputSchema（JSON Schema 格式的参数描述）。客户端通过 tools/list 发现可用工具，通过 tools/call 调用具体工具。

Input Schema 使用 JSON Schema 格式定义参数约束。例如一个文件读取工具可能要求 path 参数是字符串且以 / 开头，一个计算器工具可能要求 expression 参数是数学表达式字符串。JSON Schema 提供了类型检查、枚举值、正则表达式校验等能力，让 LLM 可以理解每个参数的预期格式和取值范围，从而生成正确的调用参数。

MCP 协议还定义了 Resources（资源） 和 Prompts（提示模板） 两种补充能力。资源是只读的数据源（文件内容、数据库记录、API 响应），通过 URI 定位（如 file:///path/to/doc.md 或 db://users/123）。提示模板是预定义的 Prompt 模板，可带参数填充。这三者共同构成了 MCP 服务器的完整能力面：工具用于执行动作，资源用于提供上下文，提示用于规范化交互模式。

list[Tool]:\n return list(self._tools.values())\n\n def get_tool(self, name: str) -> Tool | None:\n return self._tools.get(name)','ToolManager')">查看代码 ▸ 查看代码 ▸

会话管理与安全

BaseSession 是 MCP 会话的核心异步上下文管理器，处理请求/响应的路由、通知的分发和进度追踪。安全模型采用 TransportSecurity 层，在传输层之上提供认证和加密。会话生命周期包括初始化（initialize）、能力协商（negotiate_capabilities）和正常关闭。

会话初始化时，Client 发送 initialize 请求，包含 协议版本、客户端信息（名称和版本）以及 能力声明。Server 返回其能力声明和服务器信息，然后 Client 发送 initialized 通知表示完成。这个握手过程确保了会话建立在双方共识的基础上。如果版本不兼容，Server 可以返回错误并拒绝连接。

MCP 的安全模型在 v1.x 中引入了增强的 OAuth 2.0 支持。远程 MCP 服务器可以通过 OAuth 2.0 Device Authorization Grant 或 Authorization Code Grant 进行认证。OpenCode 的 CLI 实现（packages/opencode/src/cli/cmd/mcp.ts）展示了完整的认证流程：通过 opencode mcp auth 命令触发 OAuth 流程，支持浏览器自动打开认证页面，若无法自动打开则显示 URL 供手动访问，完成后展示认证状态（authenticated/expired/not_authenticated）。此外还有 opencode mcp logout 命令管理 OAuth 凭据的生命周期。

dict:\n method = message["method"]\n handler = self._handlers.get(method)\n if not handler:\n return {"jsonrpc":"2.0","id":message.get("id"),"error":{"code":-32601,"message":"Method not found"}}\n return await handler(message)','Dispatcher')">查看代码 ▸ [name, mcp.getAuthStatus(name)])),\n { concurrency: "unbounded" },\n )\n // 启动 OAuth 流程\n yield* MCP.Service.use((mcp) => mcp.authenticate(serverName))\n})\n\n// 浏览器打开失败的降级处理\nconst events = yield* EventV2Bridge.Service\nconst unsubscribe = yield* events.listen((event) => {\n if (event.type !== MCP.BrowserOpenFailed.type) return Effect.void\n spinner.stop("Could not open browser automatically")\n prompts.log.warn("Please open this URL in your browser to authenticate:")\n prompts.log.info(data.url)\n})\n\n// 认证状态回显\nif (status.status === "connected") {\n spinner.stop("Authentication successful!")\n} else if (status.status === "needs_client_registration") {\n spinner.stop("Authentication failed", 1)\n prompts.log.error(status.error)\n}','OpenCode OAuth 认证')">查看代码 ▸