【翻译、转载】MCP 核心架构

核心架构

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

模型上下文协议 (MCP) 构建在一个灵活、可扩展的架构之上，能够实现 LLM 应用程序与集成之间的无缝通信。本文档涵盖了核心的架构组件和概念。

概述
MCP 遵循客户端-服务器 (client-server) 架构，其中：

宿主 (Host) 是发起连接的 LLM 应用程序（如 Claude 桌面版或 IDE）。
客户端 (Client) 在宿主应用程序内部与服务器保持 1:1 连接。
服务器 (Server) 向客户端提供上下文、工具和提示 (prompts)。

在这里插入图片描述

核心组件

协议层
协议层处理消息帧、请求/响应关联以及高级通信模式。

[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 implementationasync def send_notification(self,notification: NotificationT) -> None:"""Send one-way notification that doesn't expect response."""# Notification handling implementationasync def _received_request(self,responder: RequestResponder[ReceiveRequestT, ResultT]) -> None:"""Handle incoming request from other side."""# Request handling implementationasync def _received_notification(self,notification: ReceiveNotificationT) -> None:"""Handle incoming notification from other side."""# Notification handling implementation

关键类包括：

Protocol (协议)
Client (客户端)
Server (服务器)

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

Stdio 传输
- 使用标准输入/输出 (standard input/output) 进行通信。
- 非常适合本地进程。
带 SSE 的 HTTP 传输
- 使用服务器发送事件 (Server-Sent Events, SSE) 处理服务器到客户端的消息。
- 使用 HTTP POST 处理客户端到服务器的消息。

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

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

请求 (Request) 期望从对方获得响应：

interface Request {method: string; // 方法名params?: { ... }; // 参数 (可选)
}

结果 (Result) 是对请求的成功响应：

interface Result {[key: string]: unknown; // 任意键值对
}

错误 (Error) 表明请求失败：

interface Error {code: number; // 错误代码message: string; // 错误消息data?: unknown; // 额外数据 (可选)
}

通知 (Notification) 是不期望响应的单向消息：

interface Notification {method: string; // 方法名params?: { ... }; // 参数 (可选)
}

连接生命周期

1. 初始化
[时序图]
在这里插入图片描述

客户端发送包含协议版本和能力的 initialize 请求。
服务器以其协议版本和能力进行响应。
客户端发送 initialized 通知作为确认。
开始正常的消息交换。

2. 消息交换
初始化后，支持以下模式：

请求-响应 (Request-Response)：客户端或服务器发送请求，另一方响应。
通知 (Notification)：任意一方发送单向消息。

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_serverapp = 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())