本文翻译自 MCP 官方文档

核心架构

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

概述

MCP 遵循客户端-服务器架构,其中:

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

xxx

核心组件

协议层

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

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 implementation

    async def send_notification(
        self,
        notification: NotificationT
    ) -> None:
        """Send one-way notification that doesn't expect response."""
        # Notification handling implementation

    async def _received_request(
        self,
        responder: RequestResponder[ReceiveRequestT, ResultT]
    ) -> None:
        """Handle incoming request from other side."""
        # Request handling implementation

    async def _received_notification(
        self,
        notification: ReceiveNotificationT
    ) -> None:
        """Handle incoming notification from other side."""
        # Notification handling implementation

关键类包括:

  • Protocol
  • Client
  • Server

传输层

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

  1. 标准输入输出 (Stdio) 传输
    • 使用标准输入/输出进行通信
    • 适用于本地进程
  2. 使用 SSE 的 HTTP 传输
    • 使用服务器发送事件 (Server-Sent Events) 处理服务器到客户端的消息
    • 使用 HTTP POST 处理客户端到服务器的消息

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

消息类型

MCP 主要有以下几种消息类型:

  1. 请求 (Requests) 需要对方响应:
interface Request {
  method: string;
  params?: { ... };
}
  1. 结果 (Results) 是对请求的成功响应:
interface Result {
  [key: string]: unknown;
}
  1. 错误 (Errors) 表示请求失败:
interface Error {
  code: number;
  message: string;
  data?: unknown;
}
  1. 通知 (Notifications) 是不需要响应的单向消息:
interface Notification {
  method: string;
  params?: { ... };
}

连接生命周期

1. 初始化

xxx

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

2. 消息交换

初始化后,支持以下模式:

  • 请求-响应 (Request-Response):客户端或服务器发送请求,另一方响应
  • 通知 (Notifications):任何一方发送单向消息

3. 终止

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

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

错误处理

MCP 定义了以下标准错误代码:

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

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

错误通过以下方式传播:

  • 对请求的错误响应
  • 传输上的错误事件
  • 协议级别的错误处理程序

实现示例

以下是实现 MCP 服务器的基本示例:

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)

最佳实践

传输选择

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

消息处理

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

安全注意事项

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

调试和监控

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

资源

资源是模型上下文协议 (MCP) 中的一个核心原语,它允许服务器暴露数据和内容,供客户端读取并用作 LLM 交互的上下文。

概述

资源代表 MCP 服务器希望提供给客户端的任何类型的数据。这可以包括:

  • 文件内容
  • 数据库记录
  • API 响应
  • 实时系统数据
  • 屏幕截图和图像
  • 日志文件
  • 以及更多

每个资源都由一个唯一的 URI 标识,并且可以包含文本或二进制数据。

资源 URI

资源使用遵循以下格式的 URI 进行标识:

[protocol]://[host]/[path]

例如:

  • file:///home/user/documents/report.pdf
  • postgres://database/customers/schema
  • screen://localhost/display1

协议和路径结构由 MCP 服务器实现定义。服务器可以定义自己的自定义 URI 方案。

资源类型

资源可以包含两种类型的内容:

文本资源

文本资源包含 UTF-8 编码的文本数据。这些适用于:

  • 源代码
  • 配置文件
  • 日志文件
  • JSON/XML 数据
  • 纯文本

二进制资源

二进制资源包含以 Base64 编码的原始二进制数据。这些适用于:

  • 图像
  • PDF 文件
  • 音频文件
  • 视频文件
  • 其他非文本格式

资源发现

客户端可以通过两种主要方法发现可用资源:

直接资源

服务器通过 resources/list 端点暴露一个具体的资源列表。每个资源包括:

{
  uri: string;           // Unique identifier for the resource
  name: string;          // Human-readable name
  description?: string;  // Optional description
  mimeType?: string;     // Optional MIME type
}

资源模板

对于动态资源,服务器可以暴露 URI 模板,客户端可以使用这些模板来构建有效的资源 URI:

{
  uriTemplate: string;   // URI template following RFC 6570
  name: string;          // Human-readable name for this type
  description?: string;  // Optional description
  mimeType?: string;     // Optional MIME type for all matching resources
}

读取资源

要读取资源,客户端需要发送一个带有资源 URI 的 resources/read 请求。

服务器响应包含资源内容列表:

{
  contents: [
    {
      uri: string;        // The URI of the resource
      mimeType?: string;  // Optional MIME type

      // One of:
      text?: string;      // For text resources
      blob?: string;      // For binary resources (base64 encoded)
    }
  ]
}

资源更新

MCP 通过两种机制支持资源的实时更新:

列表变更

当可用资源列表发生变化时,服务器可以通过 notifications/resources/list_changed 通知来告知客户端。

内容变更

客户端可以订阅特定资源的更新:

  1. 客户端发送带有资源 URI 的 resources/subscribe 请求
  2. 当资源发生变化时,服务器发送 notifications/resources/updated 通知
  3. 客户端可以使用 resources/read 获取最新内容
  4. 客户端可以使用 resources/unsubscribe 取消订阅

实现示例

以下是在 MCP 服务器中实现资源支持的一个简单示例:

app = Server("example-server")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="file:///logs/app.log",
            name="Application Logs",
            mimeType="text/plain"
        )
    ]

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    if str(uri) == "file:///logs/app.log":
        log_contents = await read_log_file()
        return log_contents

    raise ValueError("Resource not found")

# Start server
async with stdio_server() as streams:
    await app.run(
        streams[0],
        streams[1],
        app.create_initialization_options()
    )

最佳实践

在实现资源支持时:

  1. 使用清晰、描述性的资源名称和 URI
  2. 包含有用的描述以指导 LLM 理解
  3. 在已知时设置适当的 MIME 类型
  4. 为动态内容实现资源模板
  5. 对频繁变更的资源使用订阅
  6. 通过清晰的错误消息优雅地处理错误
  7. 考虑对大型资源列表进行分页
  8. 在适当的时候缓存资源内容
  9. 在处理前验证 URI
  10. 为您的自定义 URI 方案编写文档

安全考虑

在暴露资源时:

  • 验证所有资源 URI
  • 实施适当的访问控制
  • 清理文件路径以防止目录遍历
  • 谨慎处理二进制数据
  • 考虑对资源读取进行速率限制
  • 审计资源访问
  • 加密传输中的敏感数据
  • 验证 MIME 类型
  • 为长时间运行的读取操作设置超时
  • 妥善处理资源清理

好的,这是您提供的 HTML 内容转换成的 Markdown 格式:

提示 (Prompts)

提示 (Prompts) 使服务器能够定义可重用的提示模板和工作流,客户端可以轻松地将其呈现给用户和大型语言模型 (LLM)。它们提供了一种强大的方式来标准化和共享常见的 LLM 交互。

概述

MCP 中的提示是预定义的模板,可以:

  • 接受动态参数
  • 包含来自资源的上下文
  • 链接多个交互
  • 指导特定工作流
  • 呈现为 UI 元素(如斜杠命令)

提示结构

每个提示都通过以下内容定义:

{
  name: string;              // Unique identifier for the prompt
  description?: string;      // Human-readable description
  arguments?: [              // Optional list of arguments
    {
      name: string;          // Argument identifier
      description?: string;  // Argument description
      required?: boolean;    // Whether argument is required
    }
  ]
}

发现提示

客户端可以通过 prompts/list 端点发现可用的提示:

// Request
{
  method: "prompts/list"
}

// Response
{
  prompts: [
    {
      name: "analyze-code",
      description: "Analyze code for potential improvements",
      arguments: [
        {
          name: "language",
          description: "Programming language",
          required: true
        }
      ]
    }
  ]
}

使用提示

要使用提示,客户端需要发起 prompts/get 请求:

// Request
{
  method: "prompts/get",
  params: {
    name: "analyze-code",
    arguments: {
      language: "python"
    }
  }
}

// Response
{
  description: "Analyze Python code for potential improvements",
  messages: [
    {
      role: "user",
      content: {
        type: "text",
        text: "Please analyze the following Python code for potential improvements:\n\n```python\ndef calculate_sum(numbers):\n    total = 0\n    for num in numbers:\n        total = total + num\n    return total\n\nresult = calculate_sum([1, 2, 3, 4, 5])\nprint(result)\n```"
      }
    }
  ]
}

动态提示

提示可以是动态的,并包含:

嵌入的资源上下文

{
  "name": "analyze-project",
  "description": "Analyze project logs and code",
  "arguments": [
    {
      "name": "timeframe",
      "description": "Time period to analyze logs",
      "required": true
    },
    {
      "name": "fileUri",
      "description": "URI of code file to review",
      "required": true
    }
  ]
}

在处理 prompts/get 请求时:

{
  "messages": [
    {
      "role": "user",
      "content": {
        "type": "text",
        "text": "Analyze these system logs and the code file for any issues:"
      }
    },
    {
      "role": "user",
      "content": {
        "type": "resource",
        "resource": {
          "uri": "logs://recent?timeframe=1h",
          "text": "[2024-03-14 15:32:11] ERROR: Connection timeout in network.py:127\n[2024-03-14 15:32:15] WARN: Retrying connection (attempt 2/3)\n[2024-03-14 15:32:20] ERROR: Max retries exceeded",
          "mimeType": "text/plain"
        }
      }
    },
    {
      "role": "user",
      "content": {
        "type": "resource",
        "resource": {
          "uri": "file:///path/to/code.py",
          "text": "def connect_to_service(timeout=30):\n    retries = 3\n    for attempt in range(retries):\n        try:\n            return establish_connection(timeout)\n        except TimeoutError:\n            if attempt == retries - 1:\n                raise\n            time.sleep(5)\n\ndef establish_connection(timeout):\n    # Connection implementation\n    pass",
          "mimeType": "text/x-python"
        }
      }
    }
  ]
}

多步骤工作流

const debugWorkflow = {
  name: "debug-error",
  async getMessages(error: string) {
    return [
      {
        role: "user",
        content: {
          type: "text",
          text: `Here's an error I'm seeing: ${error}`
        }
      },
      {
        role: "assistant",
        content: {
          type: "text",
          text: "I'll help analyze this error. What have you tried so far?"
        }
      },
      {
        role: "user",
        content: {
          type: "text",
          text: "I've tried restarting the service, but the error persists."
        }
      }
    ];
  }
};

实现示例

以下是在 MCP 服务器中实现提示的完整示例:

from mcp.server import Server
import mcp.types as types

# Define available prompts
PROMPTS = {
    "git-commit": types.Prompt(
        name="git-commit",
        description="Generate a Git commit message",
        arguments=[
            types.PromptArgument(
                name="changes",
                description="Git diff or description of changes",
                required=True
            )
        ],
    ),
    "explain-code": types.Prompt(
        name="explain-code",
        description="Explain how code works",
        arguments=[
            types.PromptArgument(
                name="code",
                description="Code to explain",
                required=True
            ),
            types.PromptArgument(
                name="language",
                description="Programming language",
                required=False
            )
        ],
    )
}

# Initialize server
app = Server("example-prompts-server")

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return list(PROMPTS.values())

@app.get_prompt()
async def get_prompt(
    name: str, arguments: dict[str, str] | None = None
) -> types.GetPromptResult:
    if name not in PROMPTS:
        raise ValueError(f"Prompt not found: {name}")

    if name == "git-commit":
        changes = arguments.get("changes") if arguments else ""
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"Generate a concise but descriptive commit message "
                        f"for these changes:\n\n{changes}"
                    )
                )
            ]
        )

    if name == "explain-code":
        code = arguments.get("code") if arguments else ""
        language = arguments.get("language", "Unknown") if arguments else "Unknown"
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"Explain how this {language} code works:\n\n{code}"
                    )
                )
            ]
        )

    raise ValueError("Prompt implementation not found")

最佳实践

在实现提示时:

  1. 使用清晰、描述性的提示名称
  2. 为提示和参数提供详细描述
  3. 验证所有必需的参数
  4. 优雅地处理缺失的参数
  5. 考虑提示模板的版本控制
  6. 在适当时缓存动态内容
  7. 实现错误处理
  8. 文档化预期的参数格式
  9. 考虑提示的可组合性
  10. 使用各种输入测试提示

UI 集成

提示可以在客户端 UI 中呈现为:

  • 斜杠命令
  • 快捷操作
  • 上下文菜单项
  • 命令面板条目
  • 引导式工作流
  • 交互式表单

更新与变更

服务器可以通知客户端有关提示的变更:

  1. 服务器能力:prompts.listChanged
  2. 通知:notifications/prompts/list_changed
  3. 客户端重新获取提示列表

安全注意事项

在实现提示时:

  • 验证所有参数
  • 清理用户输入
  • 考虑速率限制
  • 实施访问控制
  • 审计提示使用情况
  • 适当处理敏感数据
  • 验证生成的内容
  • 实现超时机制
  • 考虑提示注入风险
  • 文档化安全要求

工具

工具是模型上下文协议 (MCP) 中的一个强大原语,它使服务器能够向客户端公开可执行的功能。通过工具,大型语言模型 (LLM) 可以与外部系统交互、执行计算并在现实世界中采取行动。

工具被设计为由模型控制,这意味着服务器向客户端公开这些工具,目的是让 AI 模型能够自动调用它们,但调用过程需要人类用户批准。

概述

MCP 中的工具允许服务器公开可执行函数,这些函数可以被客户端调用,并被 LLM 用于执行操作。工具的关键方面包括:

  • 发现:客户端可以通过 tools/list 端点列出可用工具。
  • 调用:使用 tools/call 端点调用工具,服务器执行请求的操作并返回结果。
  • 灵活性:工具的范围可以从简单的计算到复杂的 API 交互。

资源类似,工具通过唯一的名称进行标识,并可以包含描述以指导其使用。然而,与资源不同,工具代表可以修改状态或与外部系统交互的动态操作。

工具定义

每个工具都使用以下结构进行定义:

{
  name: string;          // Unique identifier for the tool
  description?: string;  // Human-readable description
  inputSchema: {         // JSON Schema for the tool's parameters
    type: "object",
    properties: { ... }  // Tool-specific parameters
  }
}

实现工具

以下是在 MCP 服务器中实现一个基本工具的示例:

app = Server("example-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="calculate_sum",
            description="Add two numbers together",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@app.call_tool()
async def call_tool(
    name: str,
    arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name == "calculate_sum":
        a = arguments["a"]
        b = arguments["b"]
        result = a + b
        return [types.TextContent(type="text", text=str(result))]
    raise ValueError(f"Tool not found: {name}")

工具模式示例

以下是服务器可以提供的工具类型的一些示例:

系统操作

与本地系统交互的工具:

{
  name: "execute_command",
  description: "Run a shell command",
  inputSchema: {
    type: "object",
    properties: {
      command: { type: "string" },
      args: { type: "array", items: { type: "string" } }
    }
  }
}

API 集成

封装外部 API 的工具:

{
  name: "github_create_issue",
  description: "Create a GitHub issue",
  inputSchema: {
    type: "object",
    properties: {
      title: { type: "string" },
      body: { type: "string" },
      labels: { type: "array", items: { type: "string" } }
    }
  }
}

数据处理

转换或分析数据的工具:

{
  name: "analyze_csv",
  description: "Analyze a CSV file",
  inputSchema: {
    type: "object",
    properties: {
      filepath: { type: "string" },
      operations: {
        type: "array",
        items: {
          enum: ["sum", "average", "count"]
        }
      }
    }
  }
}

最佳实践

实现工具时的最佳实践:

  1. 提供清晰、描述性的名称和说明。
  2. 为参数使用详细的 JSON Schema 定义。
  3. 在工具描述中包含示例,以演示模型应如何使用它们。
  4. 实现正确的错误处理和验证。
  5. 对长时间运行的操作使用进度报告。
  6. 保持工具操作的专注性和原子性。
  7. 记录预期的返回值结构。
  8. 实现适当的超时机制。
  9. 考虑对资源密集型操作进行速率限制。
  10. 记录工具使用情况以进行调试和监控。

安全注意事项

公开工具时的安全注意事项:

输入验证

  • 根据模式验证所有参数。
  • 清理(净化)文件路径和系统命令。
  • 验证 URL 和外部标识符。
  • 检查参数大小和范围。
  • 防止命令注入。

访问控制

  • 在需要时实施身份验证。
  • 使用适当的授权检查。
  • 审计工具使用情况。
  • 限制请求速率。
  • 监控滥用行为。

错误处理

  • 不要向客户端暴露内部错误。
  • 记录与安全相关的错误。
  • 适当地处理超时。
  • 错误发生后清理资源。
  • 验证返回值。

MCP 支持动态工具发现:

  1. 客户端可以随时列出可用工具。
  2. 服务器可以在工具发生变化时使用 notifications/tools/list_changed 通知客户端。
  3. 可以在运行时添加或删除工具。
  4. 可以更新工具定义(但这应谨慎进行)。

错误处理

工具错误应在结果对象内报告,而不是作为 MCP 协议级别的错误。这使得 LLM 能够看到并可能处理该错误。当工具遇到错误时:

  1. 在结果中将 isError 设置为 true
  2. content 数组中包含错误详细信息。

以下是工具的正确错误处理示例:

try:
   # Tool operation
   result = perform_operation()
   return types.CallToolResult(
       content=[
           types.TextContent(
               type="text",
               text=f"Operation successful: {result}"
           )
       ]
   )
except Exception as error:
   return types.CallToolResult(
       isError=True,
       content=[
           types.TextContent(
               type="text",
               text=f"Error: {str(error)}"
           )
       ]
   )

这种方法允许 LLM 看到发生了错误,并可能采取纠正措施或请求人工干预。

MCP 工具的全面测试策略应涵盖:

  • 功能测试:验证工具在使用有效输入时能正确执行,并能适当地处理无效输入。
  • 集成测试:使用真实和模拟的依赖项测试工具与外部系统的交互。
  • 安全测试:验证身份验证、授权、输入清理(净化)和速率限制。
  • 性能测试:检查负载下的行为、超时处理和资源清理。
  • 错误处理测试:确保工具通过 MCP 协议正确报告错误并清理资源。

好的,这是将您提供的 HTML 内容(包含中文翻译)转换为 Markdown 格式的结果:

采样(Sampling)

采样是 MCP (模型上下文协议) 的一项强大功能,它允许服务器通过客户端请求 LLM (大语言模型) 的补全 (completion),从而在维护安全和隐私的同时实现复杂的代理行为 (agentic behaviors)。

MCP 的这项功能在 Claude Desktop 客户端中尚未得到支持。

采样工作原理

采样流程遵循以下步骤:

  1. 服务器向客户端发送一个 sampling/createMessage 请求。
  2. 客户端审查该请求并可对其进行修改。
  3. 客户端从 LLM 进行采样。
  4. 客户端审查补全结果。
  5. 客户端将结果返回给服务器。

这种“人在环路”(human-in-the-loop) 的设计确保用户能够控制 LLM 看到和生成的内容。

消息格式

采样请求使用标准化的消息格式:

{
  messages: [
    {
      role: "user" | "assistant",
      content: {
        type: "text" | "image",

        // For text:
        text?: string,

        // For images:
        data?: string,             // base64 encoded
        mimeType?: string
      }
    }
  ],
  modelPreferences?: {
    hints?: [{
      name?: string                // Suggested model name/family
    }],
    costPriority?: number,         // 0-1, importance of minimizing cost
    speedPriority?: number,        // 0-1, importance of low latency
    intelligencePriority?: number  // 0-1, importance of capabilities
  },
  systemPrompt?: string,
  includeContext?: "none" | "thisServer" | "allServers",
  temperature?: number,
  maxTokens: number,
  stopSequences?: string[],
  metadata?: Record<string, unknown>
}

请求参数

消息 (Messages)

messages 数组包含要发送给 LLM 的对话历史记录。每条消息包含:

  • role:角色,为 “user” (用户) 或 “assistant” (助手)。
  • content:消息内容,可以是:
    • 包含 text 字段的文本内容。
    • 包含 data (base64 编码) 和 mimeType 字段的图像内容。

模型偏好 (Model preferences)

modelPreferences 对象允许服务器指定其模型选择偏好:

  • hints:模型名称建议数组,客户端可用其选择合适的模型:
    • name:可匹配完整或部分模型名称的字符串 (例如 “claude-3”, “sonnet”)。
    • 客户端可以将建议映射到来自不同提供商的等效模型。
    • 多个建议按优先顺序进行评估。
  • 优先级值 (0-1 归一化):
    • costPriority:最小化成本的重要性。
    • speedPriority:低延迟响应的重要性。
    • intelligencePriority:高级模型能力的重要性。

客户端根据这些偏好及其可用的模型做出最终的模型选择。

系统提示 (System prompt)

可选的 systemPrompt 字段允许服务器请求特定的系统提示。客户端可以修改或忽略此提示。

上下文包含 (Context inclusion)

includeContext 参数指定要包含哪些 MCP 上下文:

  • "none":不包含额外上下文。
  • "thisServer":包含来自请求服务器的上下文。
  • "allServers":包含来自所有已连接 MCP 服务器的上下文。

客户端控制实际包含哪些上下文。

采样参数 (Sampling parameters)

使用以下参数微调 LLM 采样:

  • temperature:控制随机性 (0.0 到 1.0)。
  • maxTokens:要生成的最大令牌 (token) 数。
  • stopSequences:停止生成的序列数组。
  • metadata:额外的、特定于提供商的参数。

响应格式

客户端返回一个补全结果:

{
  model: string,  // Name of the model used
  stopReason?: "endTurn" | "stopSequence" | "maxTokens" | string,
  role: "user" | "assistant",
  content: {
    type: "text" | "image",
    text?: string,
    data?: string,
    mimeType?: string
  }
}

请求示例

以下是一个向客户端请求采样的示例:

{
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "What files are in the current directory?"
        }
      }
    ],
    "systemPrompt": "You are a helpful file system assistant.",
    "includeContext": "thisServer",
    "maxTokens": 100
  }
}

最佳实践

在实现采样时:

  1. 始终提供清晰、结构良好的提示。
  2. 恰当处理文本和图像内容。
  3. 设置合理的令牌限制。
  4. 通过 includeContext 包含相关上下文。
  5. 在使用响应之前对其进行验证。
  6. 优雅地处理错误。
  7. 考虑对采样请求进行速率限制。
  8. 记录预期的采样行为。
  9. 使用各种模型参数进行测试。
  10. 监控采样成本。

人在环路控制

采样在设计时就考虑了人工监督:

对于提示

  • 客户端应向用户展示建议的提示。
  • 用户应能够修改或拒绝提示。
  • 系统提示可以被过滤或修改。
  • 上下文包含由客户端控制。

对于补全结果

  • 客户端应向用户展示补全结果。
  • 用户应能够修改或拒绝补全结果。
  • 客户端可以过滤或修改补全结果。
  • 用户控制使用哪个模型。

安全注意事项

在实现采样时:

  • 验证所有消息内容。
  • 清理敏感信息。
  • 实施适当的速率限制。
  • 监控采样使用情况。
  • 加密传输中的数据。
  • 处理用户数据隐私。
  • 审计采样请求。
  • 控制成本风险。
  • 实施超时机制。
  • 优雅地处理模型错误。

常见模式

代理工作流 (Agentic workflows)

采样支持以下代理模式:

  • 读取和分析资源。
  • 基于上下文做出决策。
  • 生成结构化数据。
  • 处理多步骤任务。
  • 提供交互式协助。

上下文管理

上下文的最佳实践:

  • 请求最少的必要上下文。
  • 清晰地组织上下文结构。
  • 处理上下文大小限制。
  • 根据需要更新上下文。
  • 清理过时的上下文。

错误处理

健壮的错误处理应:

  • 捕获采样失败。
  • 处理超时错误。
  • 管理速率限制。
  • 验证响应。
  • 提供回退行为 (fallback behaviors)。
  • 适当地记录错误。

局限性

请注意以下局限性:

  • 采样依赖于客户端的能力。
  • 用户控制采样行为。
  • 上下文大小有限制。
  • 可能存在速率限制。
  • 应考虑成本。
  • 模型可用性各不相同。
  • 响应时间各不相同。
  • 并非所有内容类型都受支持。

根 (Roots)

根 (Roots) 是 MCP 中的一个概念,用于定义服务器可以操作的边界。它们为客户端提供了一种方式,告知服务器相关的资源及其位置。

什么是根 (Roots)?

根是一个 URI,客户端建议服务器应关注该 URI。当客户端连接到服务器时,它会声明服务器应使用哪些根。虽然主要用于文件系统路径,但根可以是任何有效的 URI,包括 HTTP URL。

例如,根可以是:

file:///home/user/projects/myapp
https://api.example.com/v1

为什么使用根?

根有几个重要的目的:

  1. 指导:告知服务器相关的资源和位置。
  2. 明确性:根明确了哪些资源是您工作区的一部分。
  3. 组织:多个根允许您同时处理不同的资源。

根如何工作

当客户端支持根时,它会:

  1. 在连接期间声明 roots 能力。
  2. 向服务器提供建议的根列表。
  3. 当根发生变化时通知服务器(如果支持)。

虽然根是信息性的,并非严格强制,但服务器应:

  1. 遵守提供的根。
  2. 使用根 URI 定位和访问资源。
  3. 优先处理根边界内的操作。

常见用例

根通常用于定义:

  • 项目目录
  • 仓库位置
  • API 端点
  • 配置位置
  • 资源边界

最佳实践

使用根时:

  1. 仅建议必要的资源。
  2. 为根使用清晰、描述性的名称 (URI)。
  3. 监控根的可访问性。
  4. 优雅地处理根的变化。

示例

以下是一个典型的 MCP 客户端可能公开根的方式:

{
  "roots": [
    {
      "uri": "file:///home/user/projects/frontend",
      "name": "Frontend Repository"
    },
    {
      "uri": "https://api.example.com/v1",
      "name": "API Endpoint"
    }
  ]
}

此配置建议服务器同时关注本地仓库和一个 API 端点,并保持它们在逻辑上分离。

传输(Transports)

模型上下文协议 (MCP) 中的传输为客户端和服务器之间的通信奠定了基础。传输负责处理消息发送和接收的底层机制。

消息格式

MCP 使用 JSON-RPC 2.0 作为其传输格式。传输层负责将 MCP 协议消息转换为 JSON-RPC 格式以进行传输,并将接收到的 JSON-RPC 消息转换回 MCP 协议消息。

使用了三种类型的 JSON-RPC 消息:

请求 (Requests)

{
  jsonrpc: "2.0",
  id: number | string,
  method: string,
  params?: object
}

响应 (Responses)

{
  jsonrpc: "2.0",
  id: number | string,
  result?: object,
  error?: {
    code: number,
    message: string,
    data?: unknown
  }
}

通知 (Notifications)

{
  jsonrpc: "2.0",
  method: string,
  params?: object
}

内置传输类型

MCP 包含两种标准的传输实现:

标准输入/输出 (stdio)

stdio 传输通过标准输入和输出流实现通信。这对于本地集成和命令行工具特别有用。

在以下情况使用 stdio:

  • 构建命令行工具
  • 实现本地集成
  • 需要简单的进程间通信
  • 使用 shell 脚本时
  • python(Server)
app = Server("example-server")

async with stdio_server() as streams:
    await app.run(
        streams[0],
        streams[1],
        app.create_initialization_options()
    )
  • Python(Client)
params = StdioServerParameters(
    command="./server",
    args=["--option", "value"]
)

async with stdio_client(params) as streams:
    async with ClientSession(streams[0], streams[1]) as session:
        await session.initialize()

服务器发送事件 (SSE)

SSE 传输支持服务器到客户端的流式传输,并使用 HTTP POST 请求进行客户端到服务器的通信。

在以下情况使用 SSE:

  • 仅需要服务器到客户端的流式传输时
  • 在受限网络中工作时
  • 实现简单更新时
  • python(Server)
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

app = Server("example-server")
sse = SseServerTransport("/messages")

async def handle_sse(scope, receive, send):
    async with sse.connect_sse(scope, receive, send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

async def handle_messages(scope, receive, send):
    await sse.handle_post_message(scope, receive, send)

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages", endpoint=handle_messages, methods=["POST"]),
    ]
)
  • Python(Client)
async with sse_client("http://localhost:8000/sse") as streams:
    async with ClientSession(streams[0], streams[1]) as session:
        await session.initialize()

自定义传输

MCP 使得为特定需求实现自定义传输变得容易。任何传输实现只需要符合 Transport 接口:

您可以为以下目的实现自定义传输:

  • 自定义网络协议
  • 专用通信渠道
  • 与现有系统集成
  • 性能优化
  • Python

请注意:虽然 MCP 服务器通常使用 asyncio 来实现,但对于像传输(Transports)这样的底层接口,我们建议使用 anyio 来实现,以获得更广泛的兼容性。

@contextmanager
async def create_transport(
    read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
    write_stream: MemoryObjectSendStream[JSONRPCMessage]
):
    """
    Transport interface for MCP.

    Args:
        read_stream: Stream to read incoming messages from
        write_stream: Stream to write outgoing messages to
    """
    async with anyio.create_task_group() as tg:
        try:
            # Start processing messages
            tg.start_soon(lambda: process_messages(read_stream))

            # Send messages
            async with write_stream:
                yield write_stream

        except Exception as exc:
            # Handle errors
            raise exc
        finally:
            # Clean up
            tg.cancel_scope.cancel()
            await write_stream.aclose()
            await read_stream.aclose()

错误处理

传输实现应处理各种错误场景:

  1. 连接错误
  2. 消息解析错误
  3. 协议错误
  4. 网络超时
  5. 资源清理

错误处理示例:

  • Python

请注意:虽然 MCP 服务器通常使用 asyncio 来实现,但对于像传输(Transports)这样的底层接口,我们建议使用 anyio 来实现,以获得更广泛的兼容性。

@contextmanager
async def example_transport(scope: Scope, receive: Receive, send: Send):
    try:
        # Create streams for bidirectional communication
        read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
        write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

        async def message_handler():
            try:
                async with read_stream_writer:
                    # Message handling logic
                    pass
            except Exception as exc:
                logger.error(f"Failed to handle message: {exc}")
                raise exc

        async with anyio.create_task_group() as tg:
            tg.start_soon(message_handler)
            try:
                # Yield streams for communication
                yield read_stream, write_stream
            except Exception as exc:
                logger.error(f"Transport error: {exc}")
                raise exc
            finally:
                tg.cancel_scope.cancel()
                await write_stream.aclose()
                await read_stream.aclose()
    except Exception as exc:
        logger.error(f"Failed to initialize transport: {exc}")
        raise exc

最佳实践

在实现或使用 MCP 传输时:

  1. 正确处理连接生命周期
  2. 实现适当的错误处理
  3. 在连接关闭时清理资源
  4. 使用适当的超时设置
  5. 发送前验证消息
  6. 记录传输事件以供调试
  7. 在适当时实现重连逻辑
  8. 处理消息队列中的背压 (backpressure)
  9. 监控连接健康状况
  10. 实施适当的安全措施

安全注意事项

在实现传输时:

  • 实施适当的身份验证机制
  • 验证客户端凭据
  • 使用安全的令牌 (token) 处理
  • 实施授权检查

数据安全

  • 对网络传输使用 TLS
  • 加密敏感数据
  • 验证消息完整性
  • 实施消息大小限制
  • 清理 (sanitize) 输入数据

网络安全

  • 实施速率限制
  • 使用适当的超时设置
  • 处理拒绝服务 (DoS) 场景
  • 监控异常模式
  • 实施适当的防火墙规则

调试传输

调试传输问题的技巧:

  1. 启用调试日志记录
  2. 监控消息流
  3. 检查连接状态
  4. 验证消息格式
  5. 测试错误场景
  6. 使用网络分析工具
  7. 实施健康检查
  8. 监控资源使用情况
  9. 测试边缘情况
  10. 使用适当的错误跟踪