# MCP高级原语全实战：Sampling、进度通知与取消机制
## 概述
本教程深入MCP协议的三大高级原语——Sampling（服务端反向调用LLM）、Progress Notifications（进度通知）和Cancellation（请求取消），构建一个完整的「智能代码审查助手」MCP服务器。适合已掌握MCP三基础原语（Tools/Resources/Prompts）、希望实现服务端主动行为和精细控制流的开发者。
**学习目标**：掌握Sampling请求构建、进度Token流转、取消信号传播三大机制，构建具备LLM自主决策能力的生产级MCP服务器。
## 前置条件
- Node.js 18+ 或 Python 3.10+（本教程以TypeScript为主，Python附赠关键示例）
- 已理解MCP基础三原语（Tools、Resources、Prompts）
- Claude Code CLI 已安装并配置
- 了解 JSON-RPC 2.0 基本概念
---
## 第一章：MCP高级原语全景图
MCP协议在基础三原语之上，定义了四组高级能力，它们共同构成了服务端从"被动响应"到"主动协作"的跃迁：
```mermaid
graph TB
subgraph "基础原语"
T[Tools 工具调用]
R[Resources 资源读取]
P[Prompts 提示模板]
end
subgraph "高级原语"
S[Sampling
服务端调用LLM]
PN[Progress
进度通知]
CN[Cancellation
请求取消]
LN[Logging
日志通知]
end
T --> S
T --> PN
S --> CN
T --> LN
S --> LN
style S fill:#e1f5fe,stroke:#0288d1,stroke-width:2px
style PN fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
style CN fill:#fff3e0,stroke:#f57c00,stroke-width:2px
style LN fill:#fce4ec,stroke:#c62828,stroke-width:2px
```
| 高级原语 | 方向 | 核心价值 |
|---------|------|---------|
| Sampling | Server → Client | 让服务端借用宿主LLM做决策，无需自建模型 |
| Progress | 双向 | 长任务实时反馈，避免超时 |
| Cancellation | 双向 | 优雅终止进行中的操作 |
| Logging | Server → Client | 结构化日志输出，辅助调试 |
**关键洞察**：Sampling是MCP协议最独特的设计——服务端不再是纯被动的工具提供者，而是可以主动发起对LLM的调用，实现"服务端驱动的智能循环"。
---
## 第二章：Sampling——让服务端调用LLM
### 2.1 Sampling协议流程
Sampling打破了传统的"客户端问、服务端答"模式。服务端可以在处理Tool调用时，反过来向客户端（宿主LLM）发起`sampling/createMessage`请求，实现多轮自主决策。
```mermaid
sequenceDiagram
participant C as Claude (Client)
participant S as MCP Server
C->>S: tools/call {name: "review_code"}
Note over S: 开始处理...
S->>C: sampling/createMessage {
messages: [{role:"user",content:"分析这段代码..."}],
maxTokens: 1024
}
C-->>S: result {content: "这段代码有3个问题..."}
Note over S: 根据LLM回复决定下一步
S->>C: sampling/createMessage {
messages: [{role:"user",content:"生成修复建议"}]
}
C-->>S: result {content: "建议修改为..."}
S->>C: 返回 tools/call 结果
```
**核心机制**：
1. 服务端发送`sampling/createMessage`请求
2. 客户端（宿主LLM）处理并返回结果
3. 服务端根据结果决定后续行为——可以继续发起更多Sampling请求
4. 客户端可以选择拒绝Sampling请求（出于安全或成本考虑）
### 2.2 Sampling请求结构
```typescript
// sampling/createMessage 请求参数
interface CreateMessageRequest {
method: "sampling/createMessage";
params: {
messages: SamplingMessage[]; // 对话消息列表
modelPreferences?: ModelPreferences; // 模型偏好
systemPrompt?: string; // 系统提示
maxTokens: number; // 最大生成token数（必填）
temperature?: number; // 温度参数
stopSequences?: string[]; // 停止序列
metadata?: Record; // 扩展元数据
};
}
interface SamplingMessage {
role: "user" | "assistant";
content: TextContent | ImageContent;
}
interface ModelPreferences {
hints?: ModelHint[]; // 模型提示（如 "claude-sonnet"）
costPriority?: number; // 0-1，成本优先级
speedPriority?: number; // 0-1，速度优先级
intelligencePriority?: number; // 0-1，智能优先级
}
```
### 2.3 实战：构建智能代码审查Tool
创建项目结构：
```bash
mkdir mcp-advanced-server && cd mcp-advanced-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
```
`tsconfig.json`：
```json
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"outDir": "./dist",
"strict": true,
"esModuleInterop": true
},
"include": ["src/**/*"]
}
```
核心服务器代码 `src/index.ts`：
```typescript
#!/usr/bin/env node
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
CallToolRequestSchema,
ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";
// 定义服务端能力——声明支持Sampling
const server = new Server(
{
name: "smart-code-reviewer",
version: "1.0.0",
},
{
capabilities: {
tools: {},
},
}
);
// 辅助函数：发起Sampling请求
async function sampleFromLLM(
server: Server,
messages: Array<{ role: "user" | "assistant"; content: string }>,
options?: {
systemPrompt?: string;
maxTokens?: number;
temperature?: number;
}
): Promise {
const result = await server.request(
{
method: "sampling/createMessage",
params: {
messages: messages.map((m) => ({
role: m.role,
content: { type: "text", text: m.content },
})),
systemPrompt: options?.systemPrompt,
maxTokens: options?.maxTokens ?? 2048,
temperature: options?.temperature ?? 0.3,
},
},
z.object({
content: z.object({ type: z.literal("text"), text: z.string() }),
model: z.string(),
stopReason: z.string().optional(),
})
);
return result.content.text;
}
// 注册Tools列表
server.setRequestHandler(ListToolsRequestSchema, async () => ({
tools: [
{
name: "review_code",
description:
"智能代码审查：分析代码质量、安全性、性能，并给出修复建议",
inputSchema: {
type: "object" as const,
properties: {
code: {
type: "string",
description: "需要审查的代码",
},
language: {
type: "string",
description: "编程语言（如typescript, python）",
},
focus: {
type: "string",
enum: ["security", "performance", "readability", "all"],
description: "审查重点",
},
},
required: ["code", "language"],
},
},
{
name: "generate_tests",
description: "根据代码自动生成单元测试",
inputSchema: {
type: "object" as const,
properties: {
code: { type: "string", description: "源代码" },
framework: {
type: "string",
description: "测试框架（jest, pytest等）",
},
},
required: ["code"],
},
},
],
}));
// 处理Tool调用——使用Sampling实现多轮智能审查
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
if (name === "review_code") {
const { code, language, focus = "all" } = args as {
code: string;
language: string;
focus: string;
};
// 第一轮Sampling：发现代码问题
const analysisPrompt = `你是一位资深${language}代码审查专家。请仔细分析以下代码，从${
focus === "all" ? "安全性、性能、可读性" : focus
}角度找出所有问题。
代码：
\`\`\`${language}
${code}
\`\`\`
请用JSON格式输出：
{"issues": [{"severity": "critical|warning|info", "line": 行号, "description": "问题描述"}]}`;
const analysisResult = await sampleFromLLM(server, [
{ role: "user", content: analysisPrompt },
], {
systemPrompt: "你是代码审查专家。输出严格的JSON格式。",
temperature: 0.2,
});
// 第二轮Sampling：生成修复建议
const fixPrompt = `基于以下代码审查结果，为每个问题提供具体的修复建议和代码示例。
原始代码：
\`\`\`${language}
${code}
\`\`\`
审查发现的问题：
${analysisResult}
请为每个问题提供：
1. 修复原理
2. 修复后的代码片段
3. 修复验证方法`;
const fixResult = await sampleFromLLM(server, [
{ role: "user", content: analysisPrompt },
{ role: "assistant", content: analysisResult },
{ role: "user", content: fixPrompt },
], {
systemPrompt: "你是代码修复专家。提供可执行的修复方案。",
maxTokens: 4096,
});
return {
content: [
{
type: "text",
text: `## 代码审查结果\n\n### 发现的问题\n${analysisResult}\n\n### 修复建议\n${fixResult}`,
},
],
};
}
if (name === "generate_tests") {
const { code, framework = "jest" } = args as {
code: string;
framework: string;
};
const testResult = await sampleFromLLM(
server,
[
{
role: "user",
content: `为以下代码生成完整的${framework}单元测试。要求覆盖所有分支和边界情况。
\`\`\`
${code}
\`\`\``,
},
],
{
systemPrompt: `你是测试工程师。生成${framework}测试代码，确保：
1. 覆盖所有公共方法
2. 包含正常和异常路径
3. 使用描述性测试名称`,
maxTokens: 4096,
temperature: 0.4,
}
);
return {
content: [{ type: "text", text: testResult }],
};
}
throw new Error(`Unknown tool: ${name}`);
});
// 启动服务器
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
}
main().catch(console.error);
```
### 2.4 常见问题与解决方案
**Q: Sampling请求被拒绝怎么办？**
部分MCP客户端（如Claude Code）可能不默认支持Sampling。处理方式：
```typescript
async function safeSample(server: Server, messages: Array<{role: "user"|"assistant"; content: string}>) {
try {
return await sampleFromLLM(server, messages);
} catch (error: any) {
if (error.code === -32601 || error.message?.includes("sampling")) {
// Sampling不被支持，降级到纯工具返回
return "Sampling不可用，请手动审查代码";
}
throw error;
}
}
```
**Q: 如何避免无限Sampling循环？**
```typescript
const MAX_SAMPLING_ROUNDS = 5;
async function boundedSampling(
server: Server,
initialPrompt: string,
evaluator: (response: string) => { done: boolean; nextPrompt?: string }
) {
let messages: Array<{ role: "user" | "assistant"; content: string }> = [
{ role: "user", content: initialPrompt },
];
for (let round = 0; round < MAX_SAMPLING_ROUNDS; round++) {
const response = await sampleFromLLM(server, messages);
messages.push({ role: "assistant", content: response });
const evaluation = evaluator(response);
if (evaluation.done) {
return { response, rounds: round + 1 };
}
if (evaluation.nextPrompt) {
messages.push({ role: "user", content: evaluation.nextPrompt });
}
}
return { response: messages[messages.length - 1].content, rounds: MAX_SAMPLING_ROUNDS };
}
```
---
## 第三章：Progress Notifications——进度通知
### 3.1 进度通知协议
Progress用于长耗时操作的实时反馈。核心机制是**进度Token**——客户端在请求中携带Token，服务端在处理过程中用该Token发送进度更新。
```mermaid
sequenceDiagram
participant C as Client
participant S as Server
C->>S: tools/call {name: "analyze_repo",
_meta: {progressToken: "abc-123"}}
S-->>C: notifications/progress {
progressToken: "abc-123",
progress: 25, total: 100
}
S-->>C: notifications/progress {
progressToken: "abc-123",
progress: 50, total: 100
}
S-->>C: notifications/progress {
progressToken: "abc-123",
progress: 75, total: 100
}
S->>C: tools/call result
```
**协议要点**：
- `progressToken`：由客户端提供，可以是字符串或整数
- `progress`：当前进度值（单调递增）
- `total`：可选的总进度值
- 服务端必须检查请求中是否包含`_meta.progressToken`
### 3.2 实战：带进度的仓库分析工具
在上一章的`index.ts`中添加进度通知支持：
```typescript
import { ProgressNotification } from "@modelcontextprotocol/sdk/types.js";
// 辅助函数：发送进度通知
async function reportProgress(
server: Server,
progressToken: string | number,
progress: number,
total?: number
) {
await server.notification({
method: "notifications/progress",
params: { progressToken, progress, total },
});
}
// 在ListTools中添加新工具
// tools数组中追加：
{
name: "analyze_repo",
description: "分析Git仓库的代码质量指标",
inputSchema: {
type: "object" as const,
properties: {
repoPath: { type: "string", description: "仓库路径" },
metrics: {
type: "array",
items: {
type: "string",
enum: ["complexity", "duplication", "coverage", "dependencies"],
},
description: "分析指标",
},
},
required: ["repoPath"],
},
}
// 在CallTool处理中追加：
if (name === "analyze_repo") {
const { repoPath, metrics = ["complexity", "duplication"] } = args as {
repoPath: string;
metrics: string[];
};
// 获取进度Token
const progressToken = request.params._meta?.progressToken;
const totalSteps = metrics.length;
const results: string[] = [];
for (let i = 0; i < metrics.length; i++) {
// 模拟分析过程
const metric = metrics[i];
// 发送进度通知
if (progressToken !== undefined) {
await reportProgress(server, progressToken, i, totalSteps);
}
// 使用Sampling对每个指标进行智能分析
const analysis = await sampleFromLLM(server, [
{
role: "user",
content: `分析仓库 ${repoPath} 的${metric}指标。
请给出1-10的质量评分和改进建议。`,
},
], {
maxTokens: 1024,
temperature: 0.3,
});
results.push(`### ${metric}\n${analysis}`);
// 每步完成后更新进度
if (progressToken !== undefined) {
await reportProgress(server, progressToken, i + 1, totalSteps);
}
}
return {
content: [
{
type: "text",
text: `# 仓库分析报告：${repoPath}\n\n${results.join("\n\n")}`,
},
],
};
}
```
### 3.3 客户端如何传递进度Token
Claude Code在调用MCP工具时会自动处理进度Token。如果你需要从客户端代码传递：
```json
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "analyze_repo",
"arguments": {
"repoPath": "/path/to/repo",
"metrics": ["complexity", "dependencies"]
},
"_meta": {
"progressToken": "my-progress-001"
}
}
}
```
---
## 第四章：Cancellation——请求取消
### 4.1 取消机制
MCP的取消机制基于JSON-RPC的`notifications/cancelled`通知。任何一方都可以取消正在进行的请求：
```typescript
// 取消通知结构
interface CancelledNotification {
method: "notifications/cancelled";
params: {
requestId: string | number; // 被取消的请求ID
reason?: string; // 取消原因
};
}
```
### 4.2 实战：可取消的长时间分析
```typescript
import { CancelledNotificationSchema } from "@modelcontextprotocol/sdk/types.js";
// 维护进行中的任务
const activeTasks = new Map();
// 注册取消通知处理器
server.setNotificationHandler(CancelledNotificationSchema, async (notification) => {
const { requestId, reason } = notification.params;
const controller = activeTasks.get(String(requestId));
if (controller) {
controller.abort(reason);
activeTasks.delete(String(requestId));
}
});
// 可取消的工具调用实现
async function cancellableAnalyze(
server: Server,
requestId: string | number,
repoPath: string,
progressToken?: string | number
) {
const controller = new AbortController();
activeTasks.set(String(requestId), controller);
try {
const files = ["src/main.ts", "src/utils.ts", "src/config.ts"];
const results: string[] = [];
for (let i = 0; i < files.length; i++) {
// 检查取消信号
if (controller.signal.aborted) {
return {
content: [
{
type: "text",
text: `分析被取消（${controller.signal.reason || "用户取消"}）。已完成 ${i}/${files.length} 个文件。`,
},
],
isError: false,
};
}
// 报告进度
if (progressToken !== undefined) {
await reportProgress(server, progressToken, i, files.length);
}
// 执行分析
const analysis = await sampleFromLLM(server, [
{
role: "user",
content: `快速分析文件 ${files[i]} 的代码质量。`,
},
], { maxTokens: 512 });
results.push(`**${files[i]}**: ${analysis}`);
}
if (progressToken !== undefined) {
await reportProgress(server, progressToken, files.length, files.length);
}
return {
content: [
{
type: "text",
text: results.join("\n\n"),
},
],
};
} finally {
activeTasks.delete(String(requestId));
}
}
```
---
## 第五章：Logging——结构化日志通知
### 5.1 日志通知协议
服务端可以向客户端发送结构化日志，用于调试和监控：
```typescript
// 日志级别
type LogLevel = "debug" | "info" | "notice" | "warning" | "error" | "critical" | "alert" | "emergency";
// 发送日志通知
await server.notification({
method: "notifications/message",
params: {
level: "info" as LogLevel,
logger: "code-reviewer",
data: "开始分析代码复杂度...",
},
});
```
### 5.2 实战：带日志的审查流程
```typescript
async function logAndSample(
server: Server,
messages: Array<{ role: "user" | "assistant"; content: string }>,
context: string,
options?: { maxTokens?: number; temperature?: number }
) {
// 发送info级别日志
await server.notification({
method: "notifications/message",
params: {
level: "info",
logger: "smart-reviewer",
data: `[Sampling] ${context} | tokens: ${options?.maxTokens ?? 2048}`,
},
});
const startTime = Date.now();
try {
const result = await sampleFromLLM(server, messages, options);
await server.notification({
method: "notifications/message",
params: {
level: "debug",
logger: "smart-reviewer",
data: `[Sampling完成] ${context} | 耗时: ${Date.now() - startTime}ms`,
},
});
return result;
} catch (error: any) {
await server.notification({
method: "notifications/message",
params: {
level: "error",
logger: "smart-reviewer",
data: `[Sampling失败] ${context} | 错误: ${error.message}`,
},
});
throw error;
}
}
```
---
## 第六章：完整实战——智能代码审查助手
将以上所有高级原语整合为一个完整项目。
### 6.1 完整服务器代码
`src/smart-reviewer.ts`：
```typescript
#!/usr/bin/env node
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
CallToolRequestSchema,
ListToolsRequestSchema,
CancelledNotificationSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";
const activeTasks = new Map();
const MAX_SAMPLING_ROUNDS = 5;
const server = new Server(
{ name: "smart-code-reviewer", version: "2.0.0" },
{ capabilities: { tools: {} } }
);
// ========== 工具函数 ==========
async function sampleLLM(
messages: Array<{ role: "user" | "assistant"; content: string }>,
opts?: { systemPrompt?: string; maxTokens?: number; temperature?: number }
): Promise {
const result = await server.request(
{
method: "sampling/createMessage",
params: {
messages: messages.map((m) => ({
role: m.role,
content: { type: "text", text: m.content },
})),
systemPrompt: opts?.systemPrompt,
maxTokens: opts?.maxTokens ?? 2048,
temperature: opts?.temperature ?? 0.3,
},
},
z.object({
content: z.object({ type: z.literal("text"), text: z.string() }),
model: z.string(),
})
);
return result.content.text;
}
async function log(level: string, message: string) {
await server.notification({
method: "notifications/message",
params: { level: level as any, logger: "smart-reviewer", data: message },
});
}
async function progress(token: string | number | undefined, current: number, total: number) {
if (token === undefined) return;
await server.notification({
method: "notifications/progress",
params: { progressToken: token, progress: current, total },
});
}
// ========== 取消处理器 ==========
server.setNotificationHandler(CancelledNotificationSchema, async (notification) => {
const controller = activeTasks.get(String(notification.params.requestId));
if (controller) {
controller.abort(notification.params.reason);
activeTasks.delete(String(notification.params.requestId));
}
await log("info", `请求已取消: ${notification.params.reason || "用户操作"}`);
});
// ========== 工具注册 ==========
server.setRequestHandler(ListToolsRequestSchema, async () => ({
tools: [
{
name: "smart_review",
description: "智能代码审查：发现安全漏洞、性能问题和代码异味，并自动生成修复",
inputSchema: {
type: "object" as const,
properties: {
code: { type: "string", description: "待审查代码" },
language: { type: "string", description: "编程语言" },
depth: {
type: "string",
enum: ["quick", "standard", "deep"],
description: "审查深度",
},
},
required: ["code", "language"],
},
},
],
}));
// ========== 工具调用处理 ==========
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args, _meta } = request.params;
const requestId = request.id!;
const progressToken = _meta?.progressToken as string | number | undefined;
if (name === "smart_review") {
const { code, language, depth = "standard" } = args as {
code: string;
language: string;
depth: string;
};
const controller = new AbortController();
activeTasks.set(String(requestId), controller);
try {
await log("info", `开始${depth}级代码审查 | 语言: ${language}`);
await progress(progressToken, 0, 3);
// Step 1: 发现问题
await log("debug", "Phase 1: 问题发现");
const issues = await sampleLLM(
[{ role: "user", content: `以${depth}级别审查这段${language}代码的安全性和性能。\n\n\`\`\`${language}\n${code}\n\`\`\`` }],
{ systemPrompt: "你是安全审计专家。列出所有发现的问题，按严重程度排序。", temperature: 0.2 }
);
if (controller.signal.aborted) {
return { content: [{ type: "text", text: "审查已取消" }] };
}
await progress(progressToken, 1, 3);
// Step 2: 生成修复
await log("debug", "Phase 2: 生成修复方案");
const fixes = await sampleLLM(
[
{ role: "user", content: `审查${language}代码` },
{ role: "assistant", content: issues },
{ role: "user", content: `为每个问题提供具体修复代码。\n\n原始代码:\n\`\`\`${language}\n${code}\n\`\`\`` },
],
{ systemPrompt: "你是代码修复专家。提供可直接应用的修复代码。", maxTokens: 4096 }
);
if (controller.signal.aborted) {
return { content: [{ type: "text", text: "审查已取消" }] };
}
await progress(progressToken, 2, 3);
// Step 3: 总结评分
await log("debug", "Phase 3: 生成总结");
const summary = await sampleLLM(
[{ role: "user", content: `基于以下审查结果，给出总体评分(1-10)和关键行动项。\n\n问题:\n${issues}\n\n修复:\n${fixes}` }],
{ systemPrompt: "给出简洁的质量评分和优先行动清单。", maxTokens: 512, temperature: 0.1 }
);
await progress(progressToken, 3, 3);
await log("info", "代码审查完成");
return {
content: [
{
type: "text",
text: `# 智能代码审查报告\n\n## 总体评估\n${summary}\n\n## 详细问题\n${issues}\n\n## 修复方案\n${fixes}`,
},
],
};
} catch (error: any) {
await log("error", `审查失败: ${error.message}`);
return {
content: [{ type: "text", text: `审查失败: ${error.message}` }],
isError: true,
};
} finally {
activeTasks.delete(String(requestId));
}
}
throw new Error(`Unknown tool: ${name}`);
});
// ========== 启动 ==========
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
}
main().catch(console.error);
```
### 6.2 构建与配置
```bash
# 构建
npx tsc
# 在Claude Code中配置
# 编辑 ~/.claude/settings.json 或项目 .claude/settings.json
```
`.claude/settings.json` 配置：
```json
{
"mcpServers": {
"smart-reviewer": {
"command": "node",
"args": ["/absolute/path/to/mcp-advanced-server/dist/smart-reviewer.js"]
}
}
}
```
### 6.3 验证安装
```bash
# 重启Claude Code后，在对话中测试
claude "使用 smart_review 工具审查以下代码：
function getUser(id) {
const query = 'SELECT * FROM users WHERE id = ' + id;
return db.execute(query);
}
"
```
预期输出：
```
# 智能代码审查报告
## 总体评估
评分: 3/10
关键行动项:
1. 立即修复SQL注入漏洞
2. 添加参数化查询
3. 增加输入验证
## 详细问题
1. **[Critical] SQL注入**: 直接拼接用户输入到SQL查询...
2. **[Warning] 缺少错误处理**: 没有try-catch...
## 修复方案
```typescript
function getUser(id: string): Promise {
if (!/^\d+$/.test(id)) throw new Error('Invalid ID');
const query = 'SELECT * FROM users WHERE id = ?';
return db.execute(query, [id]);
}
```
```
---
## 第七章：Python版FastMCP实现
对于Python开发者，FastMCP同样支持这些高级原语：
```python
# pip install mcp
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("smart-reviewer-python")
@mcp.tool()
async def smart_review(code: str, language: str, ctx: Context) -> str:
"""智能代码审查"""
# 发送日志通知
await ctx.info(f"开始审查 {language} 代码...")
# Sampling - 调用宿主LLM
result = await ctx.session.create_message(
messages=[
{
"role": "user",
"content": {
"type": "text",
"text": f"审查以下{language}代码的安全性和质量:\n\n```\n{code}\n```"
}
}
],
max_tokens=2048,
temperature=0.3,
)
analysis = result.content.text if hasattr(result.content, 'text') else str(result.content)
# 发送进度通知
await ctx.report_progress(1, 2)
# 第二轮Sampling生成修复
fix_result = await ctx.session.create_message(
messages=[
{"role": "user", "content": {"type": "text", "text": f"分析代码"}},
{"role": "assistant", "content": {"type": "text", "text": analysis}},
{"role": "user", "content": {"type": "text", "text": f"为所有问题提供修复代码:\n```\n{code}\n```"}},
],
max_tokens=4096,
)
await ctx.report_progress(2, 2)
return f"## 审查结果\n{analysis}\n\n## 修复方案\n{fix_result.content.text}"
if __name__ == "__main__":
mcp.run()
```
配置到Claude Code：
```json
{
"mcpServers": {
"smart-reviewer-py": {
"command": "python",
"args": ["/path/to/smart_reviewer.py"]
}
}
}
```
---
## 第八章：高级模式与最佳实践
### 8.1 完整工作流架构
```mermaid
flowchart TB
subgraph "Claude Code (MCP Client)"
A[用户发起请求] --> B{请求包含
progressToken?}
B -->|Yes| C[发送带Token的
tools/call请求]
B -->|No| D[发送普通
tools/call请求]
end
subgraph "MCP Server - smart-reviewer"
E[接收tools/call] --> F[注册AbortController]
F --> G[发送日志通知
notifications/message]
G --> H[Phase 1: Sampling
问题发现]
H --> I{检查
取消信号?}
I -->|已取消| J[返回取消结果]
I -->|继续| K[发送进度通知
notifications/progress]
K --> L[Phase 2: Sampling
生成修复]
L --> M{检查取消}
M -->|已取消| J
M -->|继续| N[发送进度通知]
N --> O[Phase 3: Sampling
总结评分]
O --> P[返回完整结果]
end
C --> E
D --> E
style H fill:#e1f5fe,stroke:#0288d1
style L fill:#e1f5fe,stroke:#0288d1
style O fill:#e1f5fe,stroke:#0288d1
style K fill:#e8f5e9,stroke:#388e3c
style N fill:#e8f5e9,stroke:#388e3c
style G fill:#fce4ec,stroke:#c62828
```
### 8.2 Sampling深度控制模式
```typescript
// 渐进式Sampling：从快速摘要到详细分析
async function progressiveAnalysis(
code: string,
budget: "low" | "medium" | "high"
) {
const configs = {
low: { rounds: 1, maxTokens: 512, temperature: 0.1 },
medium: { rounds: 2, maxTokens: 2048, temperature: 0.3 },
high: { rounds: 3, maxTokens: 4096, temperature: 0.4 },
};
const config = configs[budget];
const messages: Array<{ role: "user" | "assistant"; content: string }> = [];
// Round 1: 快速扫描
messages.push({
role: "user",
content: `快速扫描代码，列出最关键的3个问题:\n\`\`\`\n${code}\n\`\`\``,
});
const scan = await sampleLLM(messages, {
maxTokens: config.maxTokens,
temperature: config.temperature,
});
messages.push({ role: "assistant", content: scan });
if (config.rounds >= 2) {
// Round 2: 深入分析
messages.push({
role: "user",
content: "对每个问题进行深入分析，给出根因和影响范围。",
});
const deep = await sampleLLM(messages, { maxTokens: config.maxTokens });
messages.push({ role: "assistant", content: deep });
}
if (config.rounds >= 3) {
// Round 3: 修复验证
messages.push({
role: "user",
content: "生成修复代码并验证修复后是否引入新问题。",
});
const verify = await sampleLLM(messages, { maxTokens: config.maxTokens });
messages.push({ role: "assistant", content: verify });
}
return messages[messages.length - 1].content;
}
```
### 8.3 生产部署检查清单
| 检查项 | 要求 | 验证方法 |
|--------|------|---------|
| Sampling降级 | 不支持Sampling时优雅降级 | 断开Sampling后测试 |
| 取消清理 | 取消后释放所有资源 | 监控AbortController |
| Token限制 | 设置maxTokens防止成本失控 | 日志审计 |
| 超时保护 | 单次Sampling设30s超时 | 长时间请求测试 |
| 日志完整 | 关键节点发送日志通知 | 检查客户端日志 |
| 进度准确 | progress值单调递增 | 多次运行验证 |
### 8.4 安全注意事项
```typescript
// Sampling安全策略
const SAMPLING_POLICY = {
maxTokensPerRequest: 4096,
maxTokensPerSession: 32768,
maxRoundsPerTool: 5,
allowedModels: ["claude-sonnet", "claude-haiku"],
blockedPatterns: [/\bpassword\b/i, /\bsecret\b/i, /\bapi[_-]?key\b/i],
};
function validateSamplingRequest(messages: Array<{ content: string }>): boolean {
// 防止敏感信息泄露到Sampling请求
const allContent = messages.map((m) => m.content).join(" ");
for (const pattern of SAMPLING_POLICY.blockedPatterns) {
if (pattern.test(allContent)) {
return false;
}
}
return true;
}
```
---
## 第九章：调试与验证
### 9.1 使用MCP Inspector调试
```bash
# 安装Inspector
npx @modelcontextprotocol/inspector
# 连接到你的服务器
npx @modelcontextprotocol/inspector node dist/smart-reviewer.js
```
Inspector允许你：
- 可视化查看Tool列表和Schema
- 手动触发Tool调用并查看Sampling请求
- 观察进度通知和日志通知
- 模拟取消操作
### 9.2 Claude Code中的调试技巧
```bash
# 启用MCP调试日志
DEBUG=mcp:* claude
# 查看MCP服务器状态
claude mcp list
# 查看特定服务器配置
claude mcp get smart-reviewer
```
### 9.3 常见错误排查
| 错误 | 原因 | 解决方案 |
|------|------|---------|
| `Method not found: sampling/createMessage` | 客户端不支持Sampling | 添加降级逻辑 |
| `Progress token not acknowledged` | 服务端未读取`_meta.progressToken` | 检查参数解析 |
| `Request timeout` | Sampling链路过长 | 减少rounds或增大超时 |
| `Connection closed` | 服务端进程崩溃 | 检查TypeScript编译错误 |
| `Empty content` | LLM返回空内容 | 增加重试逻辑 |
---
## 延伸阅读
- [MCP Specification - Progress](https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/progress) — 进度通知协议规范
- [MCP Specification - Sampling](https://modelcontextprotocol.io/specification/2025-06-18/client/sampling) — Sampling完整协议定义
- [Microsoft MCP for Beginners - Advanced Topics](https://github.com/microsoft/mcp-for-beginners/blob/main/05-AdvancedTopics/README.md) — 微软官方MCP进阶教程
- [Advanced MCP: Transports, Sampling, and Production Patterns](https://learn-prompting.fr/blog/claude-mcp-advanced-patterns) — 生产级MCP模式指南
- [MCP Server Everything](https://github.com/modelcontextprotocol/servers/tree/main/src/everything) — 官方参考实现，展示所有高级特性
## 参考来源
1. [Diving into MCP Advanced Server Capabilities](https://blog.fka.dev/blog/2025-06-11-diving-into-mcp-advanced-server-capabilities/) — MCP高级服务端能力深度解析，涵盖进度、取消、Sampling
2. [MCP Progress Specification](https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/progress) — MCP官方进度通知协议规范
3. [Microsoft mcp-for-beginners: Sampling](https://github.com/microsoft/mcp-for-beginners/blob/main/05-AdvancedTopics/mcp-sampling/README.md) — 微软官方MCP入门教程的Sampling章节
4. [Advanced MCP: Transports, Sampling, and Production Patterns](https://learn-prompting.fr/blog/claude-mcp-advanced-patterns) — 传输机制、Sampling和生产部署模式
5. [Add Progress Bars and Logging Notifications to Your MCP Server](https://zazencodes.substack.com/p/add-progress-bars-and-logging-notifications) — FastMCP进度条和日志通知实践
## 训练营适用性评估
| 维度 | 评估 |
|------|------|
| 适用训练营 | MCP高级开发训练营（Day 3-4） |
| 前置知识 | 已完成MCP基础三原语（Tools/Resources/Prompts） |
| 教学时长 | 3-4小时（含动手实操） |
| 难度等级 | 中高级 |
| 实验环境 | Node.js 18+ / Python 3.10+，Claude Code CLI |
| 产出物 | 可运行的智能代码审查MCP服务器，支持Sampling、进度通知和取消 |
> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。
