核心架构

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

模型上下文协议（ MCP ）基于一个灵活、可扩展的架构，使 LLM 应用程序和集成之间的通信无缝衔接。本文档涵盖了核心架构组件和概念。

文档地址

概述

MCP 遵循客户端-服务端架构，其中：

• 主机是启动连接的 LLM 应用程序（如 Claude 桌面或 IDE ）
• 客户端与服务端保持 1:1 连接，位于主机应用程序内部
• 服务端向客户端提供上下文、工具和提示

flowchart LR subgraph Host a1[MCP Client] a2[MCP Client] end subgraph Server Process b1[MCP Server] end subgraph Server Process b2[MCP Server] end a1 <-- Transport Layer --> b1 a2 <-- Transport Layer --> b2 

核心组件

协议层

协议层处理消息框架、请求/响应链接和高级通信模式。

Python 示例代码

class Session(BaseSession[RequestT, NotificationT, ResultT]): async def send_request( self, request: RequestT, result_type: type[Result] ) -> Result: """ Send request and wait for response. Raises McpError if response contains error. """ # Request handling implementation async def send_notification( self, notification: NotificationT ) -> None: """Send one-way notification that doesn't expect response.""" # Notification handling implementation async def _received_request( self, responder: RequestResponder[ReceiveRequestT, ResultT] ) -> None: """Handle incoming request from other side.""" # Request handling implementation async def _received_notification( self, notification: ReceiveNotificationT ) -> None: """Handle incoming notification from other side.""" # Notification handling implementation 

关键类包括：

• Protocol
• Client
• Server

传输层

传输层处理客户端和服务端之间的实际通信。MCP 支持多种传输机制：

1. Stdio 传输

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

2. HTTP 与 SSE 传输

   • 使用服务端发送事件（ SSE ）进行服务端到客户端的消息传递
   • 使用 HTTP POST 进行客户端到服务端的消息传递

所有传输都使用JSON-RPC 2.0 来交换消息。有关模型上下文协议消息格式的详细信息，请参阅规范。

消息类型

MCP 有以下主要类型的消息：

1. 请求期望从另一方获得响应：

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

1. 结果是对请求的成功响应：

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

1. 错误表示请求失败：

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

1. 通知是单向消息，不期望响应：

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

连接生命周期

1. 初始化

1. 客户端发送`initialize`请求，包含协议版本和能力 2. 服务端响应其协议版本和能力 3. 客户端发送`initialized`通知作为确认 4. 正常消息交换开始 

2. 消息交换

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

• 请求-响应：客户端或服务端发送请求，另一方响应
• 通知：任一方发送单向消息

3. 终止

任一方都可以终止连接：

• 通过close()进行干净关闭
• 传输断开
• 错误条件

错误处理

MCP 定义了这些标准错误代码：

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

SDK 和应用程序可以在-32000 以上定义自己的错误代码。

错误通过以下方式传播：

• 对请求的错误响应
• 传输上的错误事件
• 协议级错误处理程序

实现示例

以下是一个实现 MCP 服务端的基本示例：

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. 请求处理

   • 彻底验证输入
   • 使用类型安全的模式
   • 优雅地处理错误
   • 实现超时

2. 进度报告

   • 对长时间操作使用进度令牌
   • 逐步报告进度
   • 在已知时包括总进度

3. 错误管理

   • 使用适当的错误代码
   • 包括有用的错误消息
   • 在错误时清理资源

安全注意事项

1. 传输安全

   • 对远程连接使用 TLS
   • 验证连接来源
   • 在需要时实现身份验证

2. 消息验证

   • 验证所有传入消息
   • 清理输入
   • 检查消息大小限制
   • 验证 JSON-RPC 格式

3. 资源保护

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

4. 错误处理

   • 不要泄露敏感信息
   • 记录与安全相关的错误
   • 实现适当的清理
   • 处理 DoS 场景

调试和监控

1. 日志记录

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

2. 诊断

   • 实现健康检查
   • 监控连接状态
   • 跟踪资源使用情况
   • 分析性能

3. 测试

   • 测试不同的传输
   • 验证错误处理
   • 检查边缘情况
   • 负载测试服务端