MCP协议-焦点架构

焦点架构

了解MCP怎样毗连客户端、服务端和LLM
模型上下文协议（MCP）基于一个机动、可扩展的架构，使LLM应用步伐和集成之间的通讯无缝衔接。本文档涵盖了焦点架构组件和概念。
概述

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

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

     焦点组件

协议层

协议层处置惩罚消息框架、哀求/响应链接和高级通讯模式。
Python 示例代码

1. class Session(BaseSession[RequestT, NotificationT, ResultT]):
2.     async def send_request(
3.         self,
4.         request: RequestT,
5.         result_type: type[Result]
6.     ) -> Result:
7.         """
8.         Send request and wait for response. Raises McpError if response contains error.
9.         """
10.         # Request handling implementation
12.     async def send_notification(
13.         self,
14.         notification: NotificationT
15.     ) -> None:
16.         """Send one-way notification that doesn't expect response."""
17.         # Notification handling implementation
19.     async def _received_request(
20.         self,
21.         responder: RequestResponder[ReceiveRequestT, ResultT]
22.     ) -> None:
23.         """Handle incoming request from other side."""
24.         # Request handling implementation
26.     async def _received_notification(
27.         self,
28.         notification: ReceiveNotificationT
29.     ) -> None:
30.         """Handle incoming notification from other side."""
31.         # Notification handling implementation

复制代码

关键类包括：

• Protocol
  
• Client
  
• Server
  

传输层

传输层处置惩罚客户端和服务端之间的现实通讯。MCP支持多种传输机制：

• Stdio传输
  
  
  
  • 利用标准输入/输出进行通讯
    
  • 适用于本地进程
    
  
  
• HTTP与SSE传输
  
  
  
  • 利用服务端发送事件（SSE）进行服务端到客户端的消息传递
    
  • 利用HTTP POST进行客户端到服务端的消息传递
    
  
  

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

MCP有以下主要范例的消息：

• 哀求期望从另一方获得响应：
  

1.             interface Request {
2.               method: string;
3.               params?: { ... };
4.             }

复制代码

• 效果是对哀求的成功响应：
  

1.             interface Result {
2.               [key: string]: unknown;
3.             }

复制代码

• 错误表现哀求失败：
  

1.            interface Error {
2.                code: number;
3.                message: string;
4.                data?: unknown;
5.              }

复制代码

• 关照是单向消息，不期望响应：
  

1.             interface Notification {
2.               method: string;
3.               params?: { ... };
4.             }

复制代码

毗连生命周期

1. 初始化

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

复制代码

2. 消息互换

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

• 哀求-响应：客户端或服务端发送哀求，另一方响应
  
• 关照：任一方发送单向消息
  

3. 终止

任一方都可以终止毗连：

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

错误处置惩罚

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

2.     enum ErrorCode {
3.       // 标准JSON-RPC错误代码
4.       ParseError = -32700,
5.       InvalidRequest = -32600,
6.       MethodNotFound = -32601,
7.       InvalidParams = -32602,
8.       InternalError = -32603
9.     }

复制代码

SDK和应用步伐可以在-32000以上定义自己的错误代码。
错误通过以下方式传播：

• 对哀求的错误响应
  
• 传输上的错误事件
  
• 协议级错误处置惩罚步伐
  

实现示例

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

1. import asyncio
2. import mcp.types as types
3. from mcp.server import Server
4. from mcp.server.stdio import stdio_server
6. app = Server("example-server")
8. @app.list_resources()
9. async def list_resources() -> list[types.Resource]:
10.     return [
11.         types.Resource(
12.             uri="example://resource",
13.             name="Example Resource"
14.         )
15.     ]
17. async def main():
18.     async with stdio_server() as streams:
19.         await app.run(
20.             streams[0],
21.             streams[1],
22.             app.create_initialization_options()
23.         )
25. if __name__ == "__main__":
26.     asyncio.run(main)

复制代码

最佳实践

传输选择

• 本地通讯
  
  
  
  • 对本地进程利用stdio传输
    
  • 适用于同一台机器的通讯
    
  • 简单的进程管理
    
  
  
• 远程通讯
  
  
  
  • 对必要HTTP兼容性的场景利用SSE
    
  • 考虑安全影响，包括身份验证和授权
    
  
  

消息处置惩罚

• 哀求处置惩罚
  
  
  
  • 彻底验证输入
    
  • 利用范例安全的模式
    
  • 优雅地处置惩罚错误
    
  • 实现超时
    
  
  
• 进度报告
  
  
  
  • 对长时间操纵利用进度令牌
    
  • 逐步报告进度
    
  • 在已知时包括总进度
    
  
  
• 错误管理
  
  
  
  • 利用适当的错误代码
    
  • 包括有用的错误消息
    
  • 在错误时清理资源
    
  
  

安全注意事项

• 传输安全
  
  
  
  • 对远程毗连利用TLS
    
  • 验证毗连来源
    
  • 在必要时实现身份验证
    
  
  
• 消息验证
  
  
  
  • 验证全部传入消息
    
  • 清理输入
    
  • 检查消息巨细限制
    
  • 验证JSON-RPC格式
    
  
  
• 资源掩护
  
  
  
  • 实现访问控制
    
  • 验证资源路径
    
  • 监控资源利用环境
    
  • 限制哀求速率
    
  
  
• 错误处置惩罚
  
  
  
  • 不要泄露敏感信息
    
  • 记载与安全相关的错误
    
  • 实现适当的清理
    
  • 处置惩罚DoS场景
    
  
  

调试和监控

• 日志记载
  
  
  
  • 记载协议事件
    
  • 跟踪消息流
    
  • 监控性能
    
  • 记载错误
    
  
  
• 诊断
  
  
  
  • 实现健康检查
    
  • 监控毗连状态
    
  • 跟踪资源利用环境
    
  • 分析性能
    
  
  
• 测试
  
  
  
  • 测试不同的传输
    
  • 验证错误处置惩罚
    
  • 检查边沿环境
    
  • 负载测试服务端
    
  
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。