MCP（模型上下文协议）核心架构解析

MCP（模型上下文协议）核心架构解析

MCP（Model Context Protocol）是一种用于连接客户端、服务器和LLM（大语言模型）的协议。本文将详细解析MCP的核心架构，包括其组件、传输层、消息类型、连接生命周期、错误处理等，并提供具体的实现示例和最佳实践。

MCP核心架构

了解 MCP 如何连接客户端、服务器和 LLM

Model Context Protocol (MCP) 基于灵活且可扩展的架构，能够在 LLM 应用程序和各类集成之间实现无缝通信。本文档概述了 MCP 核心架构的组件及相关概念。

概述

MCP 采用客户端-服务器架构，其中：

• Host（主机）指发起连接的 LLM 应用（如 Claude Desktop 或 IDE）
• Clients（客户端）在主机应用程序内部，维护与服务器的一对一连接
• Servers（服务器）为客户端提供上下文、工具及提示（prompts）

核心组件

协议层（Protocol layer）

协议层负责处理消息的封装、请求/响应关联，以及高级通信模式。

class Protocol<Request, Notification, Result> {
    // 处理传入请求
    setRequestHandler<T>(schema: T, handler: (request: T, extra: RequestHandlerExtra) => Promise<Result>): void
    // 处理传入通知
    setNotificationHandler<T>(schema: T, handler: (notification: T) => Promise<void>): void
    // 发送请求并等待响应
    request<T>(request: Request, schema: T, options?: RequestOptions): Promise<T>
    // 发送单向通知
    notification(notification: Notification): Promise<void>
}

class Session(BaseSession[RequestT, NotificationT, ResultT]):
    async def send_request(
        self,
        request: RequestT,
        result_type: type[Result]
    ) -> Result:
        """
        发送请求并等待响应。如果响应中包含错误，将抛出 McpError。
        """
        # 请求处理的具体实现
    async def send_notification(
        self,
        notification: NotificationT
    ) -> None:
        """发送不需要响应的单向通知。"""
        # 通知处理的具体实现
    async def _received_request(
        self,
        responder: RequestResponder[ReceiveRequestT, ResultT]
    ) -> None:
        """处理对端发来的请求。"""
        # 请求处理的具体实现
    async def _received_notification(
        self,
        notification: ReceiveNotificationT
    ) -> None:
        """处理对端发来的通知。"""
        # 通知处理的具体实现

关键类包括：

• Protocol
• Client
• Server

传输层（Transport layer）

传输层负责在客户端和服务器之间进行实际的数据通信。MCP 支持多种传输机制：

Stdio 传输（Stdio transport）

• 使用标准输入/输出进行通信
• 适用于本地进程通信

基于 HTTP + SSE 的传输（HTTP with SSE transport）

• 通过 Server-Sent Events 实现服务器到客户端的消息传输
• 通过 HTTP POST 实现客户端到服务器的请求

所有传输都使用JSON-RPC 2.0来交换消息。详细的 Model Context Protocol 消息格式请参阅规范。

消息类型

MCP 有以下主要消息类型：

Requests（请求）：期待对端返回响应：

interface Request {
  method: string;
  params?: { ... };
}

Results（结果）：对请求的成功响应：

interface Result {
  [key: string]: unknown;
}

Errors（错误）：表示请求失败：

interface Error {
  code: number;
  message: string;
  data?: unknown;
}

Notifications（通知）：单向消息，不期待响应：

interface Notification {
  method: string;
  params?: { ... };
}

连接生命周期

1. 初始化（Initialization）

1. 客户端通过 initialize 请求发送协议版本和功能列表
2. 服务器返回协议版本和功能列表
3. 客户端通过 initialized 通知进行确认
4. 可以开始正常消息交换

2. 消息交换

初始化完成后，支持以下模式：

• Request-Response（请求-响应）：客户端或服务器发起请求，对方返回响应
• Notifications（通知）：任意一方可以发送单向消息

3. 终止（Termination）

任何一方都可以终止连接：

• 通过 close() 进行正常关闭
• 传输层断开连接
• 遇到错误导致关闭

错误处理

MCP 定义了以下标准错误码：

enum ErrorCode {
  // JSON-RPC 标准错误码
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

SDK 和应用程序可以定义大于 -32000 的自定义错误码。错误会通过以下方式进行传播：

• 请求的错误响应
• 传输层错误事件
• 协议层的错误处理器

实现示例

下面展示了一个实现 MCP 服务器的简单示例：

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const server = new Server({
  name: "example-server",
  version: "1.0.0"
}, {
  capabilities: {
    resources: {}
  }
});
// 处理请求
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: "example://resource",
        name: "Example Resource"
      }
    ]
  };
});
// 连接传输层
const transport = new StdioServerTransport();
await server.connect(transport);

import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
app = Server("example-server")
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="example://resource",
            name="Example Resource"
        )
    ]
async def main():
    async with stdio_server() as streams:
        await app.run(
            streams[0],
            streams[1],
            app.create_initialization_options()
        )
if __name__ == "__main__":
    asyncio.run(main)

最佳实践

传输层选择

本地通信

• 对于本地进程建议使用 Stdio 传输
• 适合同一台机器上的高效通信
• 进程管理简单

远程通信

• 对需要 HTTP 兼容的场景可使用 SSE
• 注意安全性，包括认证和授权

消息处理

请求处理

• 严格验证输入
• 使用类型安全的模式（schema）
• 正确处理错误
• 实现超时机制

进度报告

• 对耗时操作使用进度标识（progress token）
• 按阶段报告进度
• 如果已知总进度，应提供进度总值

错误管理

• 使用合适的错误码
• 提供有用的错误信息
• 在错误时清理资源

安全注意事项

传输安全

• 远程连接使用 TLS
• 验证连接来源
• 必要时实现身份认证

消息验证

• 验证所有传入消息
• 对输入进行清理
• 检查消息大小限制
• 确保符合 JSON-RPC 格式

资源保护

• 实现访问控制
• 校验资源路径
• 监控资源使用情况
• 进行请求速率限制

错误处理

• 不要泄漏敏感信息
• 记录安全相关的错误
• 实现正确的清理工作
• 考虑拒绝服务（DoS）风险

调试和监控

日志记录

• 记录协议事件
• 跟踪消息流向
• 监控性能
• 记录错误

诊断

• 实现健康检查（health checks）
• 监控连接状态
• 跟踪资源使用
• 性能剖析（profiling）

测试

• 测试不同传输方式
• 验证错误处理
• 检查边界情况
• 对服务器进行负载测试