MCP 采样机制

采样（Sampling）是 MCP 中服务器主动请求 LLM 进行推理的机制。

什么是采样？

采样允许 MCP 服务器发起 LLM 调用，实现：

• 递归 LLM 交互
• 代理行为
• 复杂工作流自动化

基本概念

采样请求结构

from mcp.types import SamplingMessage

SamplingMessage(
    role="user",           # 或 "assistant"
    content=ContentBlock(
        type="text",
        text="请分析这段代码..."
    )
)

服务器采样能力声明

# 服务器声明支持采样
server_capabilities = {
    "sampling": {}
}

使用采样

服务器端

from mcp.server import Server
from mcp.types import SamplingMessage, TextContent

app = Server("sampling-server")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="analyze_code",
            description="使用 LLM 分析代码质量",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "要分析的代码"
                    },
                    "focus": {
                        "type": "string",
                        "description": "分析重点"
                    }
                },
                "required": ["code"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict | None):
    if name == "analyze_code":
        # 请求 LLM 进行代码分析
        messages = [
            SamplingMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"""请分析以下代码，重点关注 {arguments.get('focus', 'best practices')}：

```{arguments.get('code')}```

请提供：
1. 代码优点
2. 潜在问题
3. 改进建议
"""
                )
            )
        ]
        
        # 发送采样请求
        result = await session.send_sampling_request(
            messages=messages,
            max_tokens=1000
        )
        
        return [TextContent(
            type="text",
            text=result.content[0].text
        )]

客户端响应

客户端收到采样请求后，会：

1. 将消息展示给用户或自动处理
2. 调用 LLM 获取响应
3. 返回结果给服务器

{
  "method": "sampling/createMessage",
  "params": {
    "messages": [...],
    "maxTokens": 1000
  }
}

采样参数

完整参数示例

result = await session.send_sampling_request(
    messages=messages,
    max_tokens=2000,                    // 最大 token 数
    stop_sequences=["END"],             // 停止序列
    temperature=0.7,                    // 温度参数
    top_p=0.9,                          // top-p 采样
    model="claude-3-5-sonnet-20241022"  // 指定模型
)

参数说明

参数	类型	说明
messages	list	消息列表
max_tokens	int	最大生成 token 数
stop_sequences	list	停止序列
temperature	float	采样温度 (0-2)
top_p	float	nucleus 采样
model	string	指定模型

应用场景

1. 智能代码审查

async def smart_review(code: str):
    messages = [
        SamplingMessage(role="user", content=TextContent(
            type="text",
            text=f"详细审查以下代码：\n{code}"
        ))
    ]
    
    # LLM 分析
    result = await session.send_sampling_request(messages)
    
    # 进一步询问
    follow_up = [
        SamplingMessage(role="user", content=TextContent(
            type="text",
            text="基于以上分析，提供具体的修复代码"
        ))
    ]
    
    fix = await session.send_sampling_request(follow_up)
    return result, fix

2. 自动文档生成

async def generate_docs(code: str):
    # 第一次调用：分析代码结构
    analysis = await session.send_sampling_request([
        SamplingMessage(role="user", content=TextContent(
            type="text",
            text=f"分析这段代码的结构和功能：\n{code}"
        ))
    ])
    
    # 第二次调用：生成文档
    docs = await session.send_sampling_request([
        SamplingMessage(role="user", content=TextContent(
            type="text",
            text=f"基于以下分析生成文档：\n{analysis.content[0].text}"
        ))
    ])
    
    return docs

3. 调试助手

async def debug(error: str, stack_trace: str):
    messages = [
        SamplingMessage(role="user", content=TextContent(
            type="text",
            text=f"""分析以下错误：
错误: {error}
堆栈: {stack_trace}

请：
1. 定位问题根源
2. 解释原因
3. 提供修复方案"""
        ))
    ]
    
    result = await session.send_sampling_request(messages)
    return result

安全考虑

采样风险

• 无限循环：服务器可能发起无限采样请求
• 资源消耗：大量采样请求消耗资源
• 数据泄露：敏感数据可能被发送给 LLM

客户端控制

客户端应实现：

1. 请求限制：限制采样频率和数量
2. 用户确认：重要操作需要用户确认
3. 超时控制：采样请求设置超时
4. 日志记录：记录所有采样请求

最佳实践

1. 谨慎使用：只在必要时使用采样
2. 设置限制：限制采样深度和频率
3. 提供上下文：确保 LLM 有足够信息
4. 处理超时：为采样请求设置超时
5. 用户控制：允许用户禁用采样