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

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

MCP（Model Context Protocol）是一种基于灵活且可扩展架构的通信协议，能够在LLM应用程序和各类集成之间实现无缝通信。本文将详细介绍MCP的核心架构组件及相关概念，包括协议层、传输层、消息类型、连接生命周期、错误处理等内容，并提供具体的实现示例和最佳实践建议。

概述

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

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

核心组件

协议层（Protocol layer）

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

TypeScript实现

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>
}

Python实现

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支持多种传输机制：

• tdio传输（Stdio transport）
• 使用标准输入/输出进行通信
• 适用于本地进程通信
• 基于HTTP + SSE的传输（HTTP with SSE transport）
• 通过Server-Sent Events实现服务器到客户端的消息传输
• 通过HTTP POST实现客户端到服务器的请求

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

消息类型

MCP有以下主要消息类型：

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

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

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

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

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

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

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

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

连接生命周期

1. 初始化（Initialization）

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服务器的简单示例：

TypeScript实现

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);

Python实现

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)

最佳实践

传输层选择

1. 本地通信

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

2. 远程通信

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

消息处理

1. 请求处理

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

2. 进度报告

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

3. 错误管理

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

安全注意事项

1. 传输安全

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

2. 消息验证

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

3. 资源保护

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

4. 错误处理

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

调试和监控

1. 日志记录

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

2. 诊断

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

3. 测试

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