本文翻译自 MCP 官方文档。
核心架构
模型上下文协议 (MCP) 构建在一个灵活、可扩展的架构之上,旨在实现 LLM 应用和集成之间的无缝通信。本文档涵盖了其核心架构组件和概念。
概述
MCP 遵循客户端-服务器架构,其中:
- 主机 (Hosts) 是发起连接的 LLM 应用(例如 Claude Desktop 或 IDE)。
- 客户端 (Clients) 在主机应用内部,与服务器保持 1:1 连接。
- 服务器 (Servers) 向客户端提供上下文、工具和提示。
核心组件
协议层
协议层处理消息分帧、请求/响应关联以及高级通信模式。
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. 初始化
- 客户端发送包含协议版本和功能的
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 类型
- 为长时间运行的读取操作设置超时
- 妥善处理资源清理
好的,这是您提供的 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")
最佳实践
在实现提示时:
- 使用清晰、描述性的提示名称
- 为提示和参数提供详细描述
- 验证所有必需的参数
- 优雅地处理缺失的参数
- 考虑提示模板的版本控制
- 在适当时缓存动态内容
- 实现错误处理
- 文档化预期的参数格式
- 考虑提示的可组合性
- 使用各种输入测试提示
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) 场景
- 监控异常模式
- 实施适当的防火墙规则
调试传输
调试传输问题的技巧:
- 启用调试日志记录
- 监控消息流
- 检查连接状态
- 验证消息格式
- 测试错误场景
- 使用网络分析工具
- 实施健康检查
- 监控资源使用情况
- 测试边缘情况
- 使用适当的错误跟踪