核心架构
了解 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
关键类包括：
ProtocolClientServer
传输层
传输层处理客户端和服务端之间的实际通信。MCP 支持多种传输机制：
-
Stdio 传输
- 使用标准输入/输出进行通信
- 适用于本地进程
-
HTTP 与 SSE 传输
- 使用服务端发送事件（ SSE ）进行服务端到客户端的消息传递
- 使用 HTTP POST 进行客户端到服务端的消息传递
所有传输都使用JSON-RPC 2.0 来交换消息。有关模型上下文协议消息格式的详细信息，请参阅规范。
消息类型
MCP 有以下主要类型的消息：
- 请求期望从另一方获得响应：
interface Request {
method: string;
params?: { ... };
}
- 结果是对请求的成功响应：
interface Result {
[key: string]: unknown;
}
- 错误表示请求失败：
interface Error {
code: number;
message: string;
data?: unknown;
}
- 通知是单向消息，不期望响应：
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)
最佳实践
传输选择
-
本地通信
- 对本地进程使用 stdio 传输
- 适用于同一台机器的通信
- 简单的进程管理
-
远程通信
- 对需要 HTTP 兼容性的场景使用 SSE
- 考虑安全影响，包括身份验证和授权
消息处理
-
请求处理
- 彻底验证输入
- 使用类型安全的模式
- 优雅地处理错误
- 实现超时
-
进度报告
- 对长时间操作使用进度令牌
- 逐步报告进度
- 在已知时包括总进度
-
错误管理
- 使用适当的错误代码
- 包括有用的错误消息
- 在错误时清理资源
安全注意事项
-
传输安全
- 对远程连接使用 TLS
- 验证连接来源
- 在需要时实现身份验证
-
消息验证
- 验证所有传入消息
- 清理输入
- 检查消息大小限制
- 验证 JSON-RPC 格式
-
资源保护
- 实现访问控制
- 验证资源路径
- 监控资源使用情况
- 限制请求速率
-
错误处理
- 不要泄露敏感信息
- 记录与安全相关的错误
- 实现适当的清理
- 处理 DoS 场景
调试和监控
-
日志记录
- 记录协议事件
- 跟踪消息流
- 监控性能
- 记录错误
-
诊断
- 实现健康检查
- 监控连接状态
- 跟踪资源使用情况
- 分析性能
-
测试
- 测试不同的传输
- 验证错误处理
- 检查边缘情况
- 负载测试服务端
Select Language