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）

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

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）

测试

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