本文翻译自 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 支持多种传输机制：

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

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

消息类型

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. 初始化

xxx

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

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)

最佳实践

传输选择

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

消息处理

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

安全注意事项

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

调试和监控

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

资源

资源是模型上下文协议 (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 通知来告知客户端。

内容变更

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

客户端发送带有资源 URI 的 resources/subscribe 请求
当资源发生变化时，服务器发送 notifications/resources/updated 通知
客户端可以使用 resources/read 获取最新内容
客户端可以使用 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()
    )

最佳实践

在实现资源支持时：

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

安全考虑

在暴露资源时：

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

提示 (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")

最佳实践

在实现提示时：

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

UI 集成

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

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

更新与变更

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

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

安全注意事项

在实现提示时：

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

工具

工具是模型上下文协议 (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"]
        }
      }
    }
  }
}

最佳实践

实现工具时的最佳实践：

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

安全注意事项

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

输入验证

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

访问控制

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

错误处理

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

MCP 支持动态工具发现：

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

错误处理

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

在结果中将 isError 设置为 true。
在 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 客户端中尚未得到支持。

采样工作原理

采样流程遵循以下步骤：

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

这种“人在环路”(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
  }
}

最佳实践

在实现采样时：

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

人在环路控制

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

对于提示

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

对于补全结果

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

安全注意事项

在实现采样时：

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

常见模式

代理工作流 (Agentic workflows)

采样支持以下代理模式：

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

上下文管理

上下文的最佳实践：

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

错误处理

健壮的错误处理应：

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

局限性

请注意以下局限性：

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

根 (Roots)

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

什么是根 (Roots)？

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

例如，根可以是：

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

为什么使用根？

根有几个重要的目的：

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

根如何工作

当客户端支持根时，它会：

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

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

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

常见用例

根通常用于定义：

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

最佳实践

使用根时：

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

示例

以下是一个典型的 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()

错误处理

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

连接错误
消息解析错误
协议错误
网络超时
资源清理

错误处理示例：

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 传输时：

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

安全注意事项

在实现传输时：

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

数据安全

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

网络安全

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

调试传输

调试传输问题的技巧：

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

核心架构#

概述#

核心组件#

协议层#

传输层#

消息类型#

连接生命周期#

1. 初始化#

2. 消息交换#

3. 终止#

错误处理#

实现示例#

最佳实践#

传输选择#

消息处理#

安全注意事项#

调试和监控#

资源#

概述#

资源 URI#

资源类型#

文本资源#

二进制资源#

资源发现#

直接资源#

资源模板#

读取资源#

资源更新#

列表变更#

内容变更#

实现示例#

最佳实践#

安全考虑#

提示 (Prompts)#

概述#

提示结构#

发现提示#

使用提示#

动态提示#

嵌入的资源上下文#

多步骤工作流#

实现示例#

最佳实践#

UI 集成#

更新与变更#

安全注意事项#

工具#

概述#

工具定义#

实现工具#

工具模式示例#

系统操作#

API 集成#

数据处理#

最佳实践#

安全注意事项#

输入验证#

访问控制#

错误处理#

错误处理#

采样（Sampling）#

采样工作原理#

消息格式#

请求参数#

消息 (Messages)#

模型偏好 (Model preferences)#

系统提示 (System prompt)#

上下文包含 (Context inclusion)#

采样参数 (Sampling parameters)#

响应格式#

请求示例#

最佳实践#

人在环路控制#

对于提示#

对于补全结果#

安全注意事项#

常见模式#

代理工作流 (Agentic workflows)#

上下文管理#

错误处理#

核心架构

概述

核心组件

协议层

传输层

消息类型

连接生命周期

1. 初始化

2. 消息交换

3. 终止

错误处理

实现示例

最佳实践

传输选择

消息处理

安全注意事项

调试和监控

资源

概述

资源 URI

资源类型

文本资源

二进制资源

资源发现

直接资源

资源模板

读取资源

资源更新

列表变更

内容变更

实现示例

最佳实践

安全考虑

提示 (Prompts)

概述

提示结构

发现提示

使用提示

动态提示

嵌入的资源上下文

多步骤工作流

实现示例

最佳实践

UI 集成

更新与变更

安全注意事项

工具

概述

工具定义

实现工具

工具模式示例

系统操作

API 集成

数据处理

最佳实践

安全注意事项

输入验证

访问控制

错误处理

错误处理

采样（Sampling）

采样工作原理

消息格式

请求参数

消息 (Messages)

模型偏好 (Model preferences)

系统提示 (System prompt)

上下文包含 (Context inclusion)

采样参数 (Sampling parameters)

响应格式

请求示例

最佳实践

人在环路控制

对于提示

对于补全结果

安全注意事项

常见模式

代理工作流 (Agentic workflows)

上下文管理

错误处理