核心架构

Documentation Index

Fetch the complete documentation index at: https://docs.mcp.ximosoft.cn/llms.txt

Use this file to discover all available pages before exploring further.

​

概述

• 主机是发起连接的 LLM 应用程序(如 Claude Desktop 或 IDE)
• 客户端在主机应用程序内部与服务器保持 1:1 连接
• 服务器向客户端提供上下文、工具和提示词

​

核心组件

​

协议层

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

​

传输层

1. 标准输入输出传输
   • 使用标准输入/输出进行通信
   • 适用于本地进程
2. 带 SSE 的 HTTP 传输
   • 使用服务器发送事件进行服务器到客户端的消息传输
   • 使用 HTTP POST 进行客户端到服务器的消息传输

​

消息类型

1. 请求 期望从对方得到响应:
   interface Request {
     method: string;
     params?: { ... };
   }
   
2. 结果 是对请求的成功响应:
   interface Result {
     [key: string]: unknown;
   }
   
3. 错误 表示请求失败:
   interface Error {
     code: number;
     message: string;
     data?: unknown;
   }
   
4. Notifications 是单向消息，不期望响应:
   interface Notification {
     method: string;
     params?: { ... };
   }
   

​

连接生命周期

​

1. 初始化

1. 客户端发送 initialize 请求，包含协议版本和能力
2. 服务器响应其协议版本和能力
3. 客户端发送 initialized 通知作为确认
4. 正常消息交换开始

​

2. 消息交换

• 请求-响应: 客户端或服务器发送请求，另一方响应
• Notifications: 任意一方发送单向消息

​

3. 终止

• 通过 close() 干净关闭
• 传输断开
• 错误条件

​

错误处理

enum ErrorCode {
  // Standard JSON-RPC error codes
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

• 错误响应到请求
• 传输错误事件
• 协议级错误处理程序

​

服务器实现示例

import { Server } from "@modelcontextprotocol/sdk/server.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: "示例资源"
      }
    ]
  };
});

// 连接传输
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="示例资源"
        )
    ]

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)

​

最佳实践

​

传输选择

1. 本地通信
   • 对本地进程使用标准输入输出传输
   • 适用于同一机器上的通信
   • 简单的进程管理
2. 远程通信
   • 在需要 HTTP 兼容性的场景使用 SSE
   • 考虑包括认证和授权在内的安全影响

​

消息处理

1. 请求处理
   • 彻底验证输入
   • 使用类型安全的模式
   • 优雅地处理错误
   • 实现超时机制
2. 进度报告
   • 对长时间操作使用进度令牌
   • 逐步报告进度
   • 在已知时包含总进度
3. 错误管理
   • 使用适当的错误代码
   • 包含有用的错误消息
   • 在出错时清理资源

​

安全考虑

1. 传输安全
   • 对远程连接使用 TLS
   • 验证连接来源
   • 在需要时实现认证
2. 消息验证
   • 验证所有传入消息
   • 净化输入
   • 检查消息大小限制
   • 验证 JSON-RPC 格式
3. 资源保护
   • 实现访问控制
   • 验证资源路径
   • 监控资源使用
   • 限制请求速率
4. 错误处理
   • 不要泄露敏感信息
   • 记录安全相关错误
   • 实现正确的清理
   • 处理 DoS 场景

​

调试和监控

1. 日志记录
   • 记录协议事件
   • 跟踪消息流
   • 监控性能
   • 记录错误
2. 诊断
   • 实现健康检查
   • 监控连接状态
   • 跟踪资源使用
   • 分析性能
3. 测试
   • 测试不同的传输
   • 验证错误处理
   • 检查边缘情况
   • 负载测试服务器