资讯

历史

科技

环境与自然

成长

游戏

财经

文学与艺术

美食

健康

家居

文化

情感

汽车

三农

军事

旅行

运动

教育

生活

星座命理

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

创作时间:

作者:

@小白创作中心

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

引用

来源

https://www.bilibili.com/opus/1022254559229116469

MCP（Model Context Protocol）是一种基于灵活且可扩展架构的通信协议，能够在LLM应用程序和各类集成之间实现无缝通信。本文将详细介绍MCP的核心架构组件及相关概念，包括协议层、传输层、消息类型、连接生命周期、错误处理等内容，并提供具体的实现示例和最佳实践建议。

概述

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

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

核心组件

协议层（Protocol layer）

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

TypeScript实现

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

Python实现

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

tdio传输（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?: { ... };
}

连接生命周期

初始化（Initialization）

消息交换

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

Request-Response（请求-响应）：客户端或服务器发起请求，对方返回响应
Notifications（通知）：任意一方可以发送单向消息

终止（Termination）

任何一方都可以终止连接：

通过close()进行正常关闭
传输层断开连接
遇到错误导致关闭

错误处理

MCP定义了以下标准错误码：

enum ErrorCode {
  // JSON-RPC标准错误码
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

SDK和应用程序可以定义大于-32000的自定义错误码。

错误会通过以下方式进行传播：

请求的错误响应
传输层错误事件
协议层的错误处理器

实现示例

下面展示了一个实现MCP服务器的简单示例：

TypeScript实现

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

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)