Sign in Subscribe
MCP学习

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

MCP核心架构 了解 MCP 如何连接客户端、服务器和 LLM Model Context Protocol (MCP) 基于灵活且可扩展的架构，能够在 LLM 应用程序和各类集成之间实现无缝通信。本文档概述了 MCP 核心架构的组件及相关概念。

为什么叫QQ？

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

MCP核心架构

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

Model Context Protocol (MCP) 基于灵活且可扩展的架构，能够在 LLM 应用程序和各类集成之间实现无缝通信。本文档概述了 MCP 核心架构的组件及相关概念。

概述

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

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

核心组件

协议层（Protocol layer）

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

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

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

Stdio 传输（Stdio transport）

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

基于 HTTP + SSE 的传输（HTTP with SSE transport）

  • 通过 Server-Sent Events 实现服务器到客户端的消息传输
  • 通过 HTTP POST 实现客户端到服务器的请求

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

消息类型

MCP 有以下主要消息类型：

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

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

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

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

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

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

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

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

连接生命周期

1. 初始化（Initialization）

  1. 客户端通过 initialize 请求发送协议版本和功能列表
  2. 服务器返回协议版本和功能列表
  3. 客户端通过 initialized 通知进行确认
  4. 可以开始正常消息交换

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

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

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
  • 注意安全性，包括认证和授权

消息处理

请求处理

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

进度报告

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

错误管理

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

安全注意事项

传输安全

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

消息验证

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

资源保护

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

错误处理

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

调试和监控

日志记录

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

诊断

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

测试

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

注册以获取更多类似内容。

Enter your email
Subscribe