MCP 协议深度解析（八）：Prompts 提示模板与 Sampling 采样机制！

🎯 博主简介

CSDN 「新星创作者」 人工智能技术领域博主，码龄5年，累计发布 180+篇原创 文章，博客总访问量 24万+浏览!

🚀 持续更新AI前沿实战知识，专注于 AI 技术实战，核心方向包括：

• 🔥 生产级 RAG 系统 — 架构设计、多路召回、混合检索、生产优化
• 🤖 多 Agent 协作系统 — Swarm 架构、任务编排、状态管理
• 🔌 MCP 协议 — 协议规范与深度实现
• ⚡ OpenClaw — AI 助手框架进阶应用

同时也涉猎计算机视觉、Java 后端与 Spring 生态、Transformer 等深度学习技术。坚持从架构到代码、从原理到部署的实战风格。每篇文章配套代码与扩展资料，欢迎交流探讨。

📱 公众号： Anyi研习社— 每天一点AI：资料、笔记、工具、趋势，一起进步。

🤝 商务合作：请搜索关注微信公众号 「Anyi研习社」

系列导读：本文是《MCP 协议深度解析》系列的第 8 篇，深入探讨 MCP 协议中 Prompts 提示模板系统与 Sampling 采样机制的设计原理、实现细节与最佳实践。通过双语言代码示例（TypeScript + Python），帮助开发者掌握 MCP 的高级交互能力。

📑 目录

• 1. 引言：Prompt 提示模板系统设计理念
• 2. 参数化提示与动态生成
• 3. 提示模板注册与发现
• 4. 多轮对话上下文管理
• 5. Sampling 采样机制：Server 请求 LLM
• 6. 采样请求的生命周期管理
• 7. 嵌套调用与递归防护
• 8. 采样成本控制与优化
• 9. 提示模板与采样的安全考虑
• 10. 实际应用场景与案例
• 11. 常见问题 FAQ
• 12. 参考文献

1. 引言：Prompt 提示模板系统设计理念

1.1 什么是 MCP Prompts

Model Context Protocol（MCP）的 Prompts 特性允许服务器向客户端暴露结构化的提示模板。这些模板可以被用户直接调用，或者由 LLM 根据上下文自动选择使用。Prompts 是 MCP 协议中实现"可复用 AI 工作流"的核心机制。

┌─────────────────────────────────────────────────────────────────┐
│ MCP Prompts 架构概览 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────┐ │
│ │ Client │◄───────►│ Server │◄───────►│ LLM │ │
│ │ (Host) │ MCP │ (Prompts │ Sampling│ Provider │ │
│ │ │ Protocol│ Provider) │ Protocol│ │ │
│ └─────────────┘ └─────────────┘ └──────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ Prompt Templates │ │
│ │ • 代码审查模板 │ │
│ │ • 数据分析模板 │ │
│ │ • 文档生成模板 │ │
│ │ • 自定义工作流 │ │
│ └─────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘

1.2 Prompts 与 Tools 的区别

1.3 设计哲学

MCP Prompts 的设计遵循以下核心原则：

2. 参数化提示与动态生成

2.1 参数化提示的基本概念

参数化提示允许开发者定义带有变量的模板，在运行时根据用户输入动态生成完整的提示内容。这是实现可复用提示模板的基础。

2.1.1 参数类型系统

MCP 支持以下参数类型：

2.2 TypeScript 实现：参数化提示定义

// prompts/code-review.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import {
ListPromptsRequestSchema,
GetPromptRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// 定义提示模板接口
interface CodeReviewArgs {
code: string;
language: string;
focus?: "performance" | "security" | "readability" | "all";
includeExamples?: boolean;
}

// 注册提示模板
server.setRequestHandler(ListPromptsRequestSchema, async () => {
return {
prompts: [
{
name: "code-review",
description: "对代码进行专业审查，提供改进建议",
arguments: [
{
name: "code",
description: "需要审查的代码",
required: true,
},
{
name: "language",
description: "编程语言",
required: true,
},
{
name: "focus",
description: "审查重点",
required: false,
},
{
name: "includeExamples",
description: "是否包含改进示例",
required: false,
},
],
},
],
};
});

// 处理提示请求
server.setRequestHandler(GetPromptRequestSchema, async (request) => {
const { name, arguments: args } = request.params;

if (name !== "code-review") {
throw new Error(`Unknown prompt: ${name}`);
}

const { code, language, focus = "all", includeExamples = false } = args as CodeReviewArgs;

// 动态生成系统提示
const systemPrompt = generateReviewPrompt(focus, includeExamples);

// 构建消息列表
const messages = [
{
role: "system" as const,
content: {
type: "text" as const,
text: systemPrompt,
},
},
{
role: "user" as const,
content: {
type: "text" as const,
text: `请审查以下 ${language} 代码：\\n\\n\\`\\`\\`${language}\\n${code}\\n\\`\\`\\``,
},
},
];

return {
description: `代码审查 – ${language}`,
messages,
};
});

// 根据参数动态生成提示内容
function generateReviewPrompt(focus: string, includeExamples: boolean): string {
const focusAreas: Record<string, string> = {
performance: "性能优化、算法复杂度、资源使用",
security: "安全漏洞、输入验证、权限控制",
readability: "代码风格、命名规范、注释质量",
all: "性能、安全、可读性、可维护性",
};

let prompt = `你是一位资深 ${focusAreas[focus] || focusAreas.all} 专家。`;
prompt += `请对提供的代码进行专业审查，重点关注 ${focusAreas[focus] || focusAreas.all}。\\n\\n`;
prompt += `审查要求：\\n`;
prompt += `1. 识别潜在问题和风险\\n`;
prompt += `2. 提供具体的改进建议\\n`;
prompt += `3. 说明问题的严重程度（高/中/低）\\n`;

if (includeExamples) {
prompt += `4. 为每个问题提供改进后的代码示例\\n`;
}

prompt += `\\n请以结构化的方式输出审查结果。`;

return prompt;
}

2.3 Python 实现：参数化提示定义

# prompts/code_review.py
from mcp.server import Server
from mcp.types import (
ListPromptsRequest,
GetPromptRequest,
Prompt,
PromptArgument,
TextContent,
)
from typing import Optional
import json

# 创建服务器实例
server = Server("code-review-server")

# 定义提示模板配置
PROMPT_TEMPLATES = {
"code-review": {
"description": "对代码进行专业审查，提供改进建议",
"arguments": [
PromptArgument(
name="code",
description="需要审查的代码",
required=True,
),
PromptArgument(
name="language",
description="编程语言",
required=True,
),
PromptArgument(
name="focus",
description="审查重点 (performance/security/readability/all)",
required=False,
),
PromptArgument(
name="include_examples",
description="是否包含改进示例",
required=False,
),
],
},
"doc-generate": {
"description": "为代码生成文档",
"arguments": [
PromptArgument(
name="code",
description="需要生成文档的代码",
required=True,
),
PromptArgument(
name="style",
description="文档风格 (google/numpy/jsdoc)",
required=False,
),
],
},
}

# 提示内容生成器
class PromptGenerator:
"""动态提示内容生成器"""

FOCUS_AREAS = {
"performance": "性能优化、算法复杂度、资源使用",
"security": "安全漏洞、输入验证、权限控制",
"readability": "代码风格、命名规范、注释质量",
"all": "性能、安全、可读性、可维护性",
}

@classmethod
def generate_review_prompt(cls, focus: str, include_examples: bool) –> str:
focus_area = cls.FOCUS_AREAS.get(focus, cls.FOCUS_AREAS["all"])

prompt = f"你是一位资深 {focus_area} 专家。\\n"
prompt += f"请对提供的代码进行专业审查，重点关注 {focus_area}。\\n\\n"
prompt += "审查要求：\\n"
prompt += "1. 识别潜在问题和风险\\n"
prompt += "2. 提供具体的改进建议\\n"
prompt += "3. 说明问题的严重程度（高/中/低）\\n"

if include_examples:
prompt += "4. 为每个问题提供改进后的代码示例\\n"

prompt += "\\n请以结构化的方式输出审查结果。"

return prompt

# 注册提示列表处理器
@server.list_prompts()
async def list_prompts() –> list[Prompt]:
return [
Prompt(
name=name,
description=config["description"],
arguments=config["arguments"],
)
for name, config in PROMPT_TEMPLATES.items()
]

# 注册提示获取处理器
@server.get_prompt()
async def get_prompt(name: str, arguments: dict) –> dict:
if name not in PROMPT_TEMPLATES:
raise ValueError(f"Unknown prompt: {name}")

if name == "code-review":
code = arguments.get("code", "")
language = arguments.get("language", "")
focus = arguments.get("focus", "all")
include_examples = arguments.get("include_examples", False)

system_prompt = PromptGenerator.generate_review_prompt(focus, include_examples)

return {
"description": f"代码审查 – {language}",
"messages": [
{
"role": "system",
"content": TextContent(type="text", text=system_prompt),
},
{
"role": "user",
"content": TextContent(
type="text",
text=f"请审查以下 {language} 代码：\\n\\n\\`\\`\\`{language}\\n{code}\\n\\`\\`\\`"
),
},
],
}

# 其他提示模板处理…
return {}

2.4 动态提示生成策略

在实际应用中，提示模板可以根据多种因素动态生成：

3. 提示模板注册与发现

3.1 注册机制

MCP 服务器通过实现 prompts/list 方法来暴露可用的提示模板。

3.1.1 注册时序图

Prompt Registry

MCP Server

MCP Client

Prompt Registry

MCP Server

MCP Client

#mermaid-svg-RM7g1DP7WZ0Bv34g{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-RM7g1DP7WZ0Bv34g .error-icon{fill:#552222;}#mermaid-svg-RM7g1DP7WZ0Bv34g .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-RM7g1DP7WZ0Bv34g .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-RM7g1DP7WZ0Bv34g .marker{fill:#333333;stroke:#333333;}#mermaid-svg-RM7g1DP7WZ0Bv34g .marker.cross{stroke:#333333;}#mermaid-svg-RM7g1DP7WZ0Bv34g svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-RM7g1DP7WZ0Bv34g p{margin:0;}#mermaid-svg-RM7g1DP7WZ0Bv34g .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-RM7g1DP7WZ0Bv34g text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-RM7g1DP7WZ0Bv34g .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-RM7g1DP7WZ0Bv34g .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-RM7g1DP7WZ0Bv34g #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-RM7g1DP7WZ0Bv34g .sequenceNumber{fill:white;}#mermaid-svg-RM7g1DP7WZ0Bv34g #sequencenumber{fill:#333;}#mermaid-svg-RM7g1DP7WZ0Bv34g #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-RM7g1DP7WZ0Bv34g .messageText{fill:#333;stroke:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-RM7g1DP7WZ0Bv34g .labelText,#mermaid-svg-RM7g1DP7WZ0Bv34g .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .loopText,#mermaid-svg-RM7g1DP7WZ0Bv34g .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-RM7g1DP7WZ0Bv34g .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-RM7g1DP7WZ0Bv34g .noteText,#mermaid-svg-RM7g1DP7WZ0Bv34g .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-RM7g1DP7WZ0Bv34g .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-RM7g1DP7WZ0Bv34g .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-RM7g1DP7WZ0Bv34g .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-RM7g1DP7WZ0Bv34g .actorPopupMenu{position:absolute;}#mermaid-svg-RM7g1DP7WZ0Bv34g .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-RM7g1DP7WZ0Bv34g .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-RM7g1DP7WZ0Bv34g .actor-man circle,#mermaid-svg-RM7g1DP7WZ0Bv34g line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-RM7g1DP7WZ0Bv34g :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

prompts/list 请求

查询所有注册模板

返回模板列表

返回 prompts[]

prompts/get (name, args)

查找指定模板

返回模板处理器

执行模板生成

返回 messages[]

3.2 TypeScript 实现：完整注册流程

// server/prompt-registry.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import {
ListPromptsRequestSchema,
GetPromptRequestSchema,
Prompt,
PromptMessage,
} from "@modelcontextprotocol/sdk/types.js";

// 提示模板定义接口
interface PromptTemplate {
name: string;
description: string;
arguments: PromptArgument[];
handler: (args: Record<string, unknown>) => Promise<{
description: string;
messages: PromptMessage[];
}>;
}

// 提示注册表
class PromptRegistry {
private templates: Map<string, PromptTemplate> = new Map();

register(template: PromptTemplate): void {
this.templates.set(template.name, template);
console.log(`[PromptRegistry] Registered: ${template.name}`);
}

unregister(name: string): void {
this.templates.delete(name);
console.log(`[PromptRegistry] Unregistered: ${name}`);
}

list(): Prompt[] {
return Array.from(this.templates.values()).map((t) => ({
name: t.name,
description: t.description,
arguments: t.arguments,
}));
}

async execute(name: string, args: Record<string, unknown>): Promise<{
description: string;
messages: PromptMessage[];
}> {
const template = this.templates.get(name);
if (!template) {
throw new Error(`Prompt not found: ${name}`);
}
return await template.handler(args);
}
}

// 创建注册表实例
const registry = new PromptRegistry();

// 注册代码审查模板
registry.register({
name: "code-review",
description: "代码审查提示模板",
arguments: [
{ name: "code", description: "代码内容", required: true },
{ name: "language", description: "编程语言", required: true },
],
handler: async (args) => {
const { code, language } = args as { code: string; language: string };
return {
description: `代码审查 (${language})`,
messages: [
{
role: "system",
content: {
type: "text",
text: "你是一位资深代码审查专家…",
},
},
{
role: "user",
content: {
type: "text",
text: `请审查以下代码：\\n\\n\\`\\`\\`${language}\\n${code}\\n\\`\\`\\``,
},
},
],
};
},
});

// 注册数据分析模板
registry.register({
name: "data-analysis",
description: "数据分析提示模板",
arguments: [
{ name: "data", description: "数据内容", required: true },
{ name: "format", description: "数据格式", required: false },
],
handler: async (args) => {
// 实现…
return { description: "数据分析", messages: [] };
},
});

// 设置 MCP 请求处理器
export function setupPromptHandlers(server: Server): void {
// 处理列表请求
server.setRequestHandler(ListPromptsRequestSchema, async () => {
return { prompts: registry.list() };
});

// 处理获取请求
server.setRequestHandler(GetPromptRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
return await registry.execute(name, args || {});
});
}

3.3 Python 实现：完整注册流程

# server/prompt_registry.py
from typing import Callable, Dict, List, Any, Optional
from dataclasses import dataclass
from mcp.types import Prompt, PromptArgument, PromptMessage, TextContent
import asyncio

@dataclass
class PromptTemplate:
"""提示模板定义"""
name: str
description: str
arguments: List[PromptArgument]
handler: Callable[[Dict[str, Any]], asyncio.Future]

class PromptRegistry:
"""提示模板注册表"""

def __init__(self):
self._templates: Dict[str, PromptTemplate] = {}

def register(self, template: PromptTemplate) –> None:
"""注册提示模板"""
self._templates[template.name] = template
print(f"[PromptRegistry] Registered: {template.name}")

def unregister(self, name: str) –> None:
"""注销提示模板"""
if name in self._templates:
del self._templates[name]
print(f"[PromptRegistry] Unregistered: {name}")

def list(self) –> List[Prompt]:
"""列出所有提示模板"""
return [
Prompt(
name=t.name,
description=t.description,
arguments=t.arguments,
)
for t in self._templates.values()
]

async def execute(self, name: str, args: Dict[str, Any]) –> Dict[str, Any]:
"""执行提示模板"""
template = self._templates.get(name)
if not template:
raise ValueError(f"Prompt not found: {name}")
return await template.handler(args)

def get_template(self, name: str) –> Optional[PromptTemplate]:
"""获取指定模板"""
return self._templates.get(name)

# 创建全局注册表
registry = PromptRegistry()

# 装饰器方式注册
def prompt(
name: str,
description: str,
arguments: Optional[List[PromptArgument]] = None
):
"""提示模板装饰器"""
def decorator(func: Callable):
template = PromptTemplate(
name=name,
description=description,
arguments=arguments or [],
handler=func,
)
registry.register(template)
return func
return decorator

# 使用示例
@prompt(
name="code-review",
description="代码审查提示模板",
arguments=[
PromptArgument(name="code", description="代码内容", required=True),
PromptArgument(name="language", description="编程语言", required=True),
]
)
async def code_review_handler(args: Dict[str, Any]) –> Dict[str, Any]:
"""代码审查处理器"""
code = args.get("code", "")
language = args.get("language", "")

return {
"description": f"代码审查 ({language})",
"messages": [
{
"role": "system",
"content": TextContent(
type="text",
text="你是一位资深代码审查专家…"
),
},
{
"role": "user",
"content": TextContent(
type="text",
text=f"请审查以下代码：\\n\\n\\`\\`\\`{language}\\n{code}\\n\\`\\`\\`"
),
},
],
}

# 动态注册示例
def register_dynamic_prompt(name: str, config: Dict[str, Any]):
"""动态注册提示模板"""
async def handler(args: Dict[str, Any]) –> Dict[str, Any]:
# 根据配置动态生成提示
template_content = config.get("template", "")
return {
"description": config.get("description", name),
"messages": [
{
"role": "system",
"content": TextContent(type="text", text=template_content),
},
],
}

registry.register(PromptTemplate(
name=name,
description=config.get("description", ""),
arguments=[
PromptArgument(**arg)
for arg in config.get("arguments", [])
],
handler=handler,
))

4. 多轮对话上下文管理

4.1 上下文管理的重要性

在多轮对话场景中，Prompts 需要维护对话历史，确保 LLM 能够理解上下文并做出连贯的响应。

4.2 上下文传递机制

┌─────────────────────────────────────────────────────────────┐
│ 多轮对话上下文传递流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Round 1: │
│ ┌─────────┐ prompts/get ┌─────────┐ │
│ │ Client │ ─────────────────► │ Server │ │
│ │ │ ◄───────────────── │ │ │
│ │ │ messages[role] │ │ │
│ └────┬────┘ └─────────┘ │
│ │ │
│ ▼ 存储到会话上下文 │
│ ┌─────────────┐ │
│ │ SessionCtx │ {role: "assistant", content: "…"} │
│ └─────────────┘ │
│ │
│ Round 2: │
│ ┌─────────┐ prompts/get ┌─────────┐ │
│ │ Client │ ─────────────────► │ Server │ │
│ │ (含历史) │ ◄───────────────── │ │ │
│ └────┬────┘ messages[] └─────────┘ │
│ │ │
│ ▼ 追加到历史 │
│ ┌─────────────┐ │
│ │ SessionCtx │ [历史消息1, 历史消息2, …] │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘

4.3 TypeScript 实现：上下文管理

// context/session-manager.ts
import { PromptMessage } from "@modelcontextprotocol/sdk/types.js";

interface SessionContext {
sessionId: string;
messages: PromptMessage[];
metadata: Record<string, unknown>;
createdAt: Date;
lastAccessedAt: Date;
}

class SessionManager {
private sessions: Map<string, SessionContext> = new Map();
private readonly TTL_MS = 30 * 60 * 1000; // 30分钟过期

createSession(): string {
const sessionId = crypto.randomUUID();
this.sessions.set(sessionId, {
sessionId,
messages: [],
metadata: {},
createdAt: new Date(),
lastAccessedAt: new Date(),
});
return sessionId;
}

getSession(sessionId: string): SessionContext | undefined {
const session = this.sessions.get(sessionId);
if (session) {
session.lastAccessedAt = new Date();
}
return session;
}

appendMessage(sessionId: string, message: PromptMessage): void {
const session = this.sessions.get(sessionId);
if (session) {
session.messages.push(message);
session.lastAccessedAt = new Date();
}
}

getContextMessages(sessionId: string, maxHistory: number = 10): PromptMessage[] {
const session = this.sessions.get(sessionId);
if (!session) return [];

// 返回最近的消息
return session.messages.slice(–maxHistory);
}

// 清理过期会话
cleanup(): void {
const now = Date.now();
for (const [id, session] of this.sessions.entries()) {
if (now – session.lastAccessedAt.getTime() > this.TTL_MS) {
this.sessions.delete(id);
console.log(`[SessionManager] Cleaned up session: ${id}`);
}
}
}
}

// 使用示例：构建带上下文的提示
async function buildContextualPrompt(
sessionManager: SessionManager,
sessionId: string,
newPrompt: PromptMessage[]
): Promise<PromptMessage[]> {
const history = sessionManager.getContextMessages(sessionId, 5);

// 组合历史消息和新提示
return […history, …newPrompt];
}

4.4 Python 实现：上下文管理

# context/session_manager.py
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import asyncio
import uuid

@dataclass
class SessionContext:
"""会话上下文"""
session_id: str
messages: List[Dict[str, Any]] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.now)
last_accessed_at: datetime = field(default_factory=datetime.now)

class SessionManager:
"""会话管理器"""

TTL_SECONDS = 30 * 60 # 30分钟过期

def __init__(self):
self._sessions: Dict[str, SessionContext] = {}
self._cleanup_task: Optional[asyncio.Task] = None

def create_session(self) –> str:
"""创建新会话"""
session_id = str(uuid.uuid4())
self._sessions[session_id] = SessionContext(session_id=session_id)
print(f"[SessionManager] Created session: {session_id}")
return session_id

def get_session(self, session_id: str) –> Optional[SessionContext]:
"""获取会话"""
session = self._sessions.get(session_id)
if session:
session.last_accessed_at = datetime.now()
return session

def append_message(self, session_id: str, message: Dict[str, Any]) –> None:
"""追加消息到会话"""
session = self._sessions.get(session_id)
if session:
session.messages.append(message)
session.last_accessed_at = datetime.now()

def get_context_messages(
self,
session_id: str,
max_history: int = 10
) –> List[Dict[str, Any]]:
"""获取上下文消息"""
session = self._sessions.get(session_id)
if not session:
return []
return session.messages[–max_history:]

def cleanup(self) –> None:
"""清理过期会话"""
now = datetime.now()
expired = [
sid for sid, session in self._sessions.items()
if (now – session.last_accessed_at).total_seconds() > self.TTL_SECONDS
]
for sid in expired:
del self._sessions[sid]
print(f"[SessionManager] Cleaned up session: {sid}")

async def start_cleanup_scheduler(self):
"""启动定期清理任务"""
while True:
await asyncio.sleep(60) # 每分钟检查一次
self.cleanup()

# 使用示例
async def build_contextual_prompt(
session_manager: SessionManager,
session_id: str,
new_messages: List[Dict[str, Any]]
) –> List[Dict[str, Any]]:
"""构建带上下文的提示"""
history = session_manager.get_context_messages(session_id, max_history=5)
return history + new_messages

5. Sampling 采样机制：Server 请求 LLM

5.1 Sampling 概述

Sampling 是 MCP 协议中允许服务器向客户端（Host）请求 LLM 采样的机制。这使得服务器能够在需要时获取 LLM 的生成能力，而不需要直接访问 LLM API。

┌─────────────────────────────────────────────────────────────────┐
│ Sampling 架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ sampling/createMessage ┌─────────┐│
│ │ Server │ ─────────────────────────────────► │ Client ││
│ │ │ │ (Host) ││
│ │ │ ◄───────────────────────────────── │ ││
│ │ │ sampling/createMessage/result ││
│ └─────────────┘ └────┬────┘│
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ LLM │ │
│ │ Provider │ │
│ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘

5.2 Sampling 请求流程

LLM Provider

MCP Client (Host)

MCP Server

LLM Provider

MCP Client (Host)

MCP Server

#mermaid-svg-SI9vleRtm8wJj4Ur{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-SI9vleRtm8wJj4Ur .error-icon{fill:#552222;}#mermaid-svg-SI9vleRtm8wJj4Ur .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-SI9vleRtm8wJj4Ur .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-SI9vleRtm8wJj4Ur .marker{fill:#333333;stroke:#333333;}#mermaid-svg-SI9vleRtm8wJj4Ur .marker.cross{stroke:#333333;}#mermaid-svg-SI9vleRtm8wJj4Ur svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-SI9vleRtm8wJj4Ur p{margin:0;}#mermaid-svg-SI9vleRtm8wJj4Ur .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-SI9vleRtm8wJj4Ur text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-SI9vleRtm8wJj4Ur .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-SI9vleRtm8wJj4Ur .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-SI9vleRtm8wJj4Ur #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-SI9vleRtm8wJj4Ur .sequenceNumber{fill:white;}#mermaid-svg-SI9vleRtm8wJj4Ur #sequencenumber{fill:#333;}#mermaid-svg-SI9vleRtm8wJj4Ur #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-SI9vleRtm8wJj4Ur .messageText{fill:#333;stroke:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-SI9vleRtm8wJj4Ur .labelText,#mermaid-svg-SI9vleRtm8wJj4Ur .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .loopText,#mermaid-svg-SI9vleRtm8wJj4Ur .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-SI9vleRtm8wJj4Ur .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-SI9vleRtm8wJj4Ur .noteText,#mermaid-svg-SI9vleRtm8wJj4Ur .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-SI9vleRtm8wJj4Ur .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-SI9vleRtm8wJj4Ur .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-SI9vleRtm8wJj4Ur .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-SI9vleRtm8wJj4Ur .actorPopupMenu{position:absolute;}#mermaid-svg-SI9vleRtm8wJj4Ur .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-SI9vleRtm8wJj4Ur .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-SI9vleRtm8wJj4Ur .actor-man circle,#mermaid-svg-SI9vleRtm8wJj4Ur line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-SI9vleRtm8wJj4Ur :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

包含 messages, modelPreferences, systemPrompt

alt

[请求有效]

[请求无效]

sampling/createMessage

验证请求

检查权限/配额

转发采样请求

返回生成结果

sampling/createMessage/result

返回错误

5.3 TypeScript 实现：Sampling 请求

// sampling/sampling-client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import {
CreateMessageRequest,
CreateMessageResult,
SamplingMessage,
ModelPreferences,
} from "@modelcontextprotocol/sdk/types.js";

interface SamplingOptions {
messages: SamplingMessage[];
modelPreferences?: ModelPreferences;
systemPrompt?: string;
maxTokens?: number;
temperature?: number;
stopSequences?: string[];
}

class SamplingClient {
private client: Client;

constructor(client: Client) {
this.client = client;
}

/**
* 发送采样请求
*/
async createMessage(options: SamplingOptions): Promise<CreateMessageResult> {
const request: CreateMessageRequest = {
method: "sampling/createMessage",
params: {
messages: options.messages,
modelPreferences: options.modelPreferences,
systemPrompt: options.systemPrompt,
maxTokens: options.maxTokens,
temperature: options.temperature,
stopSequences: options.stopSequences,
},
};

console.log("[SamplingClient] Sending request:", JSON.stringify(request, null, 2));

const result = await this.client.request(request, CreateMessageResultSchema);

console.log("[SamplingClient] Received result:", JSON.stringify(result, null, 2));

return result;
}

/**
* 简单文本生成
*/
async generateText(
prompt: string,
options: Omit<SamplingOptions, "messages"> = {}
): Promise<string> {
const result = await this.createMessage({
messages: [
{
role: "user",
content: {
type: "text",
text: prompt,
},
},
],
…options,
});

if (result.content.type === "text") {
return result.content.text;
}

throw new Error("Expected text response, got: " + result.content.type);
}

/**
* 带上下文的对话
*/
async chat(
messages: SamplingMessage[],
options: Omit<SamplingOptions, "messages"> = {}
): Promise<CreateMessageResult> {
return await this.createMessage({
messages,
…options,
});
}
}

// 使用示例
async function demonstrateSampling() {
const client = new Client({ name: "example-client", version: "1.0.0" });
const sampling = new SamplingClient(client);

// 简单生成
const response = await sampling.generateText(
"解释什么是 Model Context Protocol",
{
maxTokens: 500,
temperature: 0.7,
modelPreferences: {
hints: [{ name: "claude" }],
intelligencePriority: 0.8,
speedPriority: 0.3,
},
}
);

console.log("Generated:", response);
}

5.4 Python 实现：Sampling 请求

# sampling/sampling_client.py
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
from mcp.client import Client
from mcp.types import (
SamplingMessage,
ModelPreferences,
CreateMessageRequest,
CreateMessageResult,
TextContent,
)

@dataclass
class SamplingOptions:
"""采样选项"""
messages: List[SamplingMessage]
model_preferences: Optional[ModelPreferences] = None
system_prompt: Optional[str] = None
max_tokens: Optional[int] = None
temperature: Optional[float] = None
stop_sequences: Optional[List[str]] = None

class SamplingClient:
"""采样客户端"""

def __init__(self, client: Client):
self._client = client

async def create_message(self, options: SamplingOptions) –> CreateMessageResult:
"""发送采样请求"""
request = CreateMessageRequest(
method="sampling/createMessage",
params={
"messages": options.messages,
"modelPreferences": options.model_preferences,
"systemPrompt": options.system_prompt,
"maxTokens": options.max_tokens,
"temperature": options.temperature,
"stopSequences": options.stop_sequences,
}
)

print(f"[SamplingClient] Sending request: {request}")

result = await self._client.request(request)

print(f"[SamplingClient] Received result: {result}")

return result

async def generate_text(
self,
prompt: str,
**kwargs
) –> str:
"""简单文本生成"""
result = await self.create_message(SamplingOptions(
messages=[
SamplingMessage(
role="user",
content=TextContent(type="text", text=prompt)
)
],
**kwargs
))

if result.content.type == "text":
return result.content.text

raise ValueError(f"Expected text response, got: {result.content.type}")

async def chat(
self,
messages: List[SamplingMessage],
**kwargs
) –> CreateMessageResult:
"""带上下文的对话"""
return await self.create_message(SamplingOptions(
messages=messages,
**kwargs
))

# 使用示例
async def demonstrate_sampling():
client = Client(name="example-client", version="1.0.0")
sampling = SamplingClient(client)

# 简单生成
response = await sampling.generate_text(
"解释什么是 Model Context Protocol",
max_tokens=500,
temperature=0.7,
model_preferences=ModelPreferences(
hints=[{"name": "claude"}],
intelligencePriority=0.8,
speedPriority=0.3,
)
)

print(f"Generated: {response}")

6. 采样请求的生命周期管理

6.1 生命周期阶段

┌─────────────────────────────────────────────────────────────────┐
│ Sampling 请求生命周期 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Created │──►│Pending │──►│Active │──►│Completed│ │
│ │ │ │ │ │ │ │ │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ 请求创建 等待处理 正在采样 完成/失败 │
│ │
│ 超时处理: Pending ──(timeout)──► Failed │
│ 取消处理: Active ──(cancel)───► Cancelled │
│ │
└─────────────────────────────────────────────────────────────────┘

6.2 TypeScript 实现：生命周期管理

// sampling/lifecycle-manager.ts
import { EventEmitter } from "events";

type SamplingStatus =
| "created"
| "pending"
| "active"
| "completed"
| "failed"
| "cancelled";

interface SamplingRequest {
id: string;
status: SamplingStatus;
createdAt: Date;
startedAt?: Date;
completedAt?: Date;
error?: string;
result?: any;
}

class SamplingLifecycleManager extends EventEmitter {
private requests: Map<string, SamplingRequest> = new Map();
private readonly DEFAULT_TIMEOUT_MS = 60000; // 60秒默认超时

createRequest(id: string): SamplingRequest {
const request: SamplingRequest = {
id,
status: "created",
createdAt: new Date(),
};

this.requests.set(id, request);
this.emit("request:created", request);

return request;
}

transitionTo(id: string, status: SamplingStatus, data?: any): void {
const request = this.requests.get(id);
if (!request) {
throw new Error(`Request not found: ${id}`);
}

const oldStatus = request.status;
request.status = status;

switch (status) {
case "pending":
// 启动超时计时器
this.startTimeout(id);
break;
case "active":
request.startedAt = new Date();
break;
case "completed":
request.completedAt = new Date();
request.result = data;
break;
case "failed":
case "cancelled":
request.completedAt = new Date();
request.error = data;
break;
}

this.emit(`request:${status}`, request);
this.emit("request:transition", { request, oldStatus, newStatus: status });
}

private startTimeout(id: string): void {
setTimeout(() => {
const request = this.requests.get(id);
if (request && request.status === "pending") {
this.transitionTo(id, "failed", "Request timeout");
}
}, this.DEFAULT_TIMEOUT_MS);
}

cancelRequest(id: string): boolean {
const request = this.requests.get(id);
if (!request) return false;

if (request.status === "pending" || request.status === "active") {
this.transitionTo(id, "cancelled", "Cancelled by user");
return true;
}

return false;
}

getRequest(id: string): SamplingRequest | undefined {
return this.requests.get(id);
}

getActiveRequests(): SamplingRequest[] {
return Array.from(this.requests.values()).filter(
(r) => r.status === "pending" || r.status === "active"
);
}

// 清理完成的请求
cleanup(maxAgeMs: number = 3600000): void {
const now = Date.now();
for (const [id, request] of this.requests.entries()) {
if (request.completedAt) {
const age = now – request.completedAt.getTime();
if (age > maxAgeMs) {
this.requests.delete(id);
this.emit("request:cleaned", { id, age });
}
}
}
}
}

// 使用示例
const lifecycle = new SamplingLifecycleManager();

// 监听状态变化
lifecycle.on("request:created", (req) => {
console.log(`[Lifecycle] Request ${req.id} created`);
});

lifecycle.on("request:completed", (req) => {
console.log(`[Lifecycle] Request ${req.id} completed`);
});

lifecycle.on("request:failed", (req) => {
console.log(`[Lifecycle] Request ${req.id} failed: ${req.error}`);
});

6.3 Python 实现：生命周期管理

# sampling/lifecycle_manager.py
from typing import Dict, Optional, List, Any, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum, auto
import asyncio
import uuid

class SamplingStatus(Enum):
"""采样状态"""
CREATED = auto()
PENDING = auto()
ACTIVE = auto()
COMPLETED = auto()
FAILED = auto()
CANCELLED = auto()

@dataclass
class SamplingRequest:
"""采样请求"""
id: str
status: SamplingStatus = SamplingStatus.CREATED
created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
error: Optional[str] = None
result: Optional[Any] = None

class SamplingLifecycleManager:
"""采样生命周期管理器"""

DEFAULT_TIMEOUT_SECONDS = 60

def __init__(self):
self._requests: Dict[str, SamplingRequest] = {}
self._handlers: Dict[str, List[Callable]] = {}
self._timeout_tasks: Dict[str, asyncio.Task] = {}

def create_request(self, request_id: Optional[str] = None) –> SamplingRequest:
"""创建请求"""
req_id = request_id or str(uuid.uuid4())
request = SamplingRequest(id=req_id)
self._requests[req_id] = request
self._emit("request:created", request)
return request

def transition_to(
self,
request_id: str,
status: SamplingStatus,
data: Any = None
) –> None:
"""状态转换"""
request = self._requests.get(request_id)
if not request:
raise ValueError(f"Request not found: {request_id}")

old_status = request.status
request.status = status

if status == SamplingStatus.PENDING:
self._start_timeout(request_id)
elif status == SamplingStatus.ACTIVE:
request.started_at = datetime.now()
self._cancel_timeout(request_id)
elif status in (SamplingStatus.COMPLETED, SamplingStatus.FAILED, SamplingStatus.CANCELLED):
request.completed_at = datetime.now()
if status != SamplingStatus.COMPLETED:
request.error = str(data) if data else None
else:
request.result = data
self._cancel_timeout(request_id)

self._emit(f"request:{status.name.lower()}", request)
self._emit("request:transition", {
"request": request,
"old_status": old_status,
"new_status": status
})

def _start_timeout(self, request_id: str) –> None:
"""启动超时计时"""
async def timeout_handler():
await asyncio.sleep(self.DEFAULT_TIMEOUT_SECONDS)
request = self._requests.get(request_id)
if request and request.status == SamplingStatus.PENDING:
self.transition_to(request_id, SamplingStatus.FAILED, "Request timeout")

self._timeout_tasks[request_id] = asyncio.create_task(timeout_handler())

def _cancel_timeout(self, request_id: str) –> None:
"""取消超时计时"""
task = self._timeout_tasks.pop(request_id, None)
if task:
task.cancel()

def cancel_request(self, request_id: str) –> bool:
"""取消请求"""
request = self._requests.get(request_id)
if not request:
return False

if request.status in (SamplingStatus.PENDING, SamplingStatus.ACTIVE):
self.transition_to(request_id, SamplingStatus.CANCELLED, "Cancelled by user")
return True

return False

def get_request(self, request_id: str) –> Optional[SamplingRequest]:
"""获取请求"""
return self._requests.get(request_id)

def get_active_requests(self) –> List[SamplingRequest]:
"""获取活跃请求"""
return [
r for r in self._requests.values()
if r.status in (SamplingStatus.PENDING, SamplingStatus.ACTIVE)
]

def on(self, event: str, handler: Callable):
"""注册事件处理器"""
if event not in self._handlers:
self._handlers[event] = []
self._handlers[event].append(handler)

def _emit(self, event: str, data: Any):
"""触发事件"""
handlers = self._handlers.get(event, [])
for handler in handlers:
try:
handler(data)
except Exception as e:
print(f"Error in event handler: {e}")

def cleanup(self, max_age_seconds: int = 3600) –> None:
"""清理完成的请求"""
now = datetime.now()
to_remove = []

for req_id, request in self._requests.items():
if request.completed_at:
age = (now – request.completed_at).total_seconds()
if age > max_age_seconds:
to_remove.append(req_id)

for req_id in to_remove:
del self._requests[req_id]
self._emit("request:cleaned", {"id": req_id})

7. 嵌套调用与递归防护

7.1 嵌套调用风险

当服务器使用 Sampling 请求 LLM，而 LLM 又调用同一服务器的 Tools 时，可能形成嵌套调用链，导致无限递归。

危险场景：
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ User │────►│ Server │────►│ Host │────►│ LLM │
└─────────┘ └────┬────┘ └─────────┘ └────┬────┘
│ │
│◄───────────────────────────────┘
│ LLM 调用 tool
│
▼
┌─────────────┐
│ tool 执行中 │
│ 需要采样 │────────────────────────►
└─────────────┘ 再次请求采样（递归！）

7.2 递归防护策略

7.3 TypeScript 实现：递归防护

// guards/recursion-guard.ts
interface CallContext {
requestId: string;
parentId?: string;
depth: number;
timestamp: number;
toolName?: string;
}

class RecursionGuard {
private readonly MAX_DEPTH = 3;
private readonly MAX_CALLS_PER_MINUTE = 10;

private callStack: Map<string, CallContext> = new Map();
private callHistory: Array<{ timestamp: number; requestId: string }> = [];

/**
* 开始新的调用上下文
*/
beginContext(requestId: string, parentId?: string, toolName?: string): CallContext {
const parent = parentId ? this.callStack.get(parentId) : undefined;
const depth = parent ? parent.depth + 1 : 0;

// 检查深度限制
if (depth > this.MAX_DEPTH) {
throw new Error(
`Recursion depth exceeded: ${depth} > ${this.MAX_DEPTH}. ` +
`Possible infinite recursion detected.`
);
}

// 检查调用频率
this.checkRateLimit();

const context: CallContext = {
requestId,
parentId,
depth,
timestamp: Date.now(),
toolName,
};

this.callStack.set(requestId, context);
this.callHistory.push({ timestamp: Date.now(), requestId });

console.log(`[RecursionGuard] Context started: ${requestId} (depth: ${depth})`);

return context;
}

/**
* 结束调用上下文
*/
endContext(requestId: string): void {
const context = this.callStack.get(requestId);
if (context) {
console.log(`[RecursionGuard] Context ended: ${requestId} (duration: ${Date.now() – context.timestamp}ms)`);
this.callStack.delete(requestId);
}
}

/**
* 检查调用频率限制
*/
private checkRateLimit(): void {
const now = Date.now();
const oneMinuteAgo = now – 60000;

// 清理旧记录
this.callHistory = this.callHistory.filter(c => c.timestamp > oneMinuteAgo);

// 检查限制
if (this.callHistory.length >= this.MAX_CALLS_PER_MINUTE) {
throw new Error(
`Rate limit exceeded: ${this.callHistory.length} calls in the last minute. ` +
`Maximum allowed: ${this.MAX_CALLS_PER_MINUTE}`
);
}
}

/**
* 检测循环调用模式
*/
detectCycle(requestId: string, toolName: string): boolean {
const context = this.callStack.get(requestId);
if (!context) return false;

// 检查父链中是否有相同 toolName
let current: CallContext | undefined = context;
while (current?.parentId) {
const parent = this.callStack.get(current.parentId);
if (parent?.toolName === toolName) {
console.warn(`[RecursionGuard] Cycle detected: ${toolName} called recursively`);
return true;
}
current = parent;
}

return false;
}

/**
* 获取当前调用链
*/
getCallChain(requestId: string): string[] {
const chain: string[] = [];
let current: CallContext | undefined = this.callStack.get(requestId);

while (current) {
chain.unshift(current.toolName || current.requestId);
current = current.parentId ? this.callStack.get(current.parentId) : undefined;
}

return chain;
}

/**
* 获取当前活跃上下文数量
*/
getActiveContextCount(): number {
return this.callStack.size;
}
}

// 使用示例
const guard = new RecursionGuard();

async function handleToolCall(toolName: string, parentId?: string) {
const requestId = crypto.randomUUID();

try {
// 开始上下文
const context = guard.beginContext(requestId, parentId, toolName);

// 检测循环
if (guard.detectCycle(requestId, toolName)) {
throw new Error(`Circular dependency detected for tool: ${toolName}`);
}

// 执行工具逻辑…
console.log(`Call chain: ${guard.getCallChain(requestId).join(" -> ")}`);

// 模拟嵌套调用
if (toolName === "analyze" && context.depth < 2) {
await handleToolCall("summarize", requestId);
}

} finally {
// 结束上下文
guard.endContext(requestId);
}
}

7.4 Python 实现：递归防护

# guards/recursion_guard.py
from typing import Dict, Optional, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import threading
import uuid

@dataclass
class CallContext:
"""调用上下文"""
request_id: str
parent_id: Optional[str] = None
depth: int = 0
timestamp: datetime = field(default_factory=datetime.now)
tool_name: Optional[str] = None

class RecursionGuard:
"""递归防护器"""

MAX_DEPTH = 3
MAX_CALLS_PER_MINUTE = 10

def __init__(self):
self._call_stack: Dict[str, CallContext] = {}
self._call_history: List[Dict] = []
self._lock = threading.RLock()

def begin_context(
self,
request_id: Optional[str] = None,
parent_id: Optional[str] = None,
tool_name: Optional[str] = None
) –> CallContext:
"""开始新的调用上下文"""
req_id = request_id or str(uuid.uuid4())

with self._lock:
parent = self._call_stack.get(parent_id) if parent_id else None
depth = parent.depth + 1 if parent else 0

# 检查深度限制
if depth > self.MAX_DEPTH:
raise RecursionError(
f"Recursion depth exceeded: {depth} > {self.MAX_DEPTH}. "
f"Possible infinite recursion detected."
)

# 检查调用频率
self._check_rate_limit()

context = CallContext(
request_id=req_id,
parent_id=parent_id,
depth=depth,
tool_name=tool_name
)

self._call_stack[req_id] = context
self._call_history.append({
"timestamp": datetime.now(),
"request_id": req_id
})

print(f"[RecursionGuard] Context started: {req_id} (depth: {depth})")

return context

def end_context(self, request_id: str) –> None:
"""结束调用上下文"""
with self._lock:
context = self._call_stack.get(request_id)
if context:
duration = (datetime.now() – context.timestamp).total_seconds() * 1000
print(f"[RecursionGuard] Context ended: {request_id} (duration: {duration:.0f}ms)")
del self._call_stack[request_id]

def _check_rate_limit(self) –> None:
"""检查调用频率限制"""
now = datetime.now()
one_minute_ago = now – timedelta(minutes=1)

# 清理旧记录
self._call_history = [
c for c in self._call_history
if c["timestamp"] > one_minute_ago
]

# 检查限制
if len(self._call_history) >= self.MAX_CALLS_PER_MINUTE:
raise RuntimeError(
f"Rate limit exceeded: {len(self._call_history)} calls in the last minute. "
f"Maximum allowed: {self.MAX_CALLS_PER_MINUTE}"
)

def detect_cycle(self, request_id: str, tool_name: str) –> bool:
"""检测循环调用模式"""
with self._lock:
context = self._call_stack.get(request_id)
if not context:
return False

# 检查父链中是否有相同 tool_name
current: Optional[CallContext] = context
while current and current.parent_id:
parent = self._call_stack.get(current.parent_id)
if parent and parent.tool_name == tool_name:
print(f"[RecursionGuard] Cycle detected: {tool_name} called recursively")
return True
current = parent

return False

def get_call_chain(self, request_id: str) –> List[str]:
"""获取当前调用链"""
with self._lock:
chain = []
current: Optional[CallContext] = self._call_stack.get(request_id)

while current:
chain.insert(0, current.tool_name or current.request_id)
current = self._call_stack.get(current.parent_id) if current.parent_id else None

return chain

def get_active_context_count(self) –> int:
"""获取当前活跃上下文数量"""
with self._lock:
return len(self._call_stack)

# 上下文管理器
class GuardedContext:
"""受保护的上下文管理器"""

def __init__(self, guard: RecursionGuard, tool_name: str, parent_id: Optional[str] = None):
self.guard = guard
self.tool_name = tool_name
self.parent_id = parent_id
self.request_id: Optional[str] = None
self.context: Optional[CallContext] = None

def __enter__(self) –> CallContext:
self.request_id = str(uuid.uuid4())
self.context = self.guard.begin_context(
self.request_id,
self.parent_id,
self.tool_name
)

if self.guard.detect_cycle(self.request_id, self.tool_name):
raise RecursionError(f"Circular dependency detected for tool: {self.tool_name}")

return self.context

def __exit__(self, exc_type, exc_val, exc_tb):
if self.request_id:
self.guard.end_context(self.request_id)

# 使用示例
guard = RecursionGuard()

def handle_tool_call(tool_name: str, parent_id: Optional[str] = None):
with GuardedContext(guard, tool_name, parent_id) as context:
print(f"Call chain: {' -> '.join(guard.get_call_chain(context.request_id))}")

# 模拟嵌套调用
if tool_name == "analyze" and context.depth < 2:
handle_tool_call("summarize", context.request_id)

8. 采样成本控制与优化

8.1 成本因素分析

Sampling 请求的成本主要由以下因素决定：

8.2 成本优化策略

┌─────────────────────────────────────────────────────────────────┐
│ 采样成本优化策略 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 提示优化 │───►│ 模型选择 │───►│ 缓存策略 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ • 去除冗余内容 • 智能路由 • 结果缓存 │
│ • 使用模板变量 • 分层模型 • 相似查询合并 │
│ • 上下文压缩 • 成本预估 • 预热机制 │
│ │
└─────────────────────────────────────────────────────────────────┘

8.3 TypeScript 实现：成本控制器

// cost/cost-controller.ts
interface CostConfig {
maxInputTokens: number;
maxOutputTokens: number;
maxTotalCost: number; // 美元
preferredModels: string[];
budgetPeriod: "daily" | "weekly" | "monthly";
}

interface SamplingCost {
inputTokens: number;
outputTokens: number;
estimatedCost: number;
model: string;
timestamp: Date;
}

class CostController {
private config: CostConfig;
private costs: SamplingCost[] = [];
private cache: Map<string, { result: any; cost: SamplingCost }> = new Map();

constructor(config: CostConfig) {
this.config = config;
}

/**
* 估算采样成本
*/
estimateCost(inputText: string, outputTokens: number, model: string): SamplingCost {
// 简化的 Token 估算（实际应使用 tokenizer）
const inputTokens = Math.ceil(inputText.length / 4);

// 模型价格表（每 1K tokens）
const prices: Record<string, { input: number; output: number }> = {
"gpt-4": { input: 0.03, output: 0.06 },
"gpt-3.5-turbo": { input: 0.0015, output: 0.002 },
"claude-3-opus": { input: 0.015, output: 0.075 },
"claude-3-sonnet": { input: 0.003, output: 0.015 },
};

const price = prices[model] || prices["gpt-3.5-turbo"];
const estimatedCost =
(inputTokens / 1000) * price.input +
(outputTokens / 1000) * price.output;

return {
inputTokens,
outputTokens,
estimatedCost,
model,
timestamp: new Date(),
};
}

/**
* 检查是否超出预算
*/
checkBudget(): { allowed: boolean; remaining: number } {
const periodCosts = this.getPeriodCosts();
const totalCost = periodCosts.reduce((sum, c) => sum + c.estimatedCost, 0);
const remaining = this.config.maxTotalCost – totalCost;

return {
allowed: remaining > 0,
remaining: Math.max(0, remaining),
};
}

/**
* 选择最优模型
*/
selectModel(complexity: "low" | "medium" | "high"): string {
const modelTiers: Record<string, string[]> = {
low: ["gpt-3.5-turbo", "claude-3-sonnet"],
medium: ["claude-3-sonnet", "gpt-4"],
high: ["claude-3-opus", "gpt-4"],
};

const candidates = modelTiers[complexity] || modelTiers.medium;

// 优先选择配置中指定的模型
for (const preferred of this.config.preferredModels) {
if (candidates.includes(preferred)) {
return preferred;
}
}

return candidates[0];
}

/**
* 检查缓存
*/
checkCache(prompt: string): { result: any; cost: SamplingCost } | undefined {
const hash = this.hashPrompt(prompt);
return this.cache.get(hash);
}

/**
* 存入缓存
*/
setCache(prompt: string, result: any, cost: SamplingCost): void {
const hash = this.hashPrompt(prompt);
this.cache.set(hash, { result, cost });

// 限制缓存大小
if (this.cache.size > 1000) {
const firstKey = this.cache.keys().next().value;
this.cache.delete(firstKey);
}
}

/**
* 记录实际成本
*/
recordCost(cost: SamplingCost): void {
this.costs.push(cost);
this.cleanupOldCosts();
}

/**
* 获取周期内的成本
*/
private getPeriodCosts(): SamplingCost[] {
const now = new Date();
const periodStart = new Date();

switch (this.config.budgetPeriod) {
case "daily":
periodStart.setHours(0, 0, 0, 0);
break;
case "weekly":
periodStart.setDate(now.getDate() – now.getDay());
periodStart.setHours(0, 0, 0, 0);
break;
case "monthly":
periodStart.setDate(1);
periodStart.setHours(0, 0, 0, 0);
break;
}

return this.costs.filter(c => c.timestamp >= periodStart);
}

/**
* 清理旧成本记录
*/
private cleanupOldCosts(): void {
const oneMonthAgo = new Date();
oneMonthAgo.setMonth(oneMonthAgo.getMonth() – 1);
this.costs = this.costs.filter(c => c.timestamp >= oneMonthAgo);
}

/**
* 生成提示哈希
*/
private hashPrompt(prompt: string): string {
// 简化实现，实际应使用 crypto 模块
let hash = 0;
for (let i = 0; i < prompt.length; i++) {
const char = prompt.charCodeAt(i);
hash = ((hash << 5) – hash) + char;
hash = hash & hash;
}
return hash.toString(16);
}

/**
* 获取成本报告
*/
getCostReport(): {
totalCost: number;
totalCalls: number;
averageCost: number;
byModel: Record<string, { calls: number; cost: number }>;
} {
const periodCosts = this.getPeriodCosts();
const totalCost = periodCosts.reduce((sum, c) => sum + c.estimatedCost, 0);

const byModel: Record<string, { calls: number; cost: number }> = {};
for (const cost of periodCosts) {
if (!byModel[cost.model]) {
byModel[cost.model] = { calls: 0, cost: 0 };
}
byModel[cost.model].calls++;
byModel[cost.model].cost += cost.estimatedCost;
}

return {
totalCost,
totalCalls: periodCosts.length,
averageCost: totalCost / periodCosts.length || 0,
byModel,
};
}
}

// 使用示例
const costController = new CostController({
maxInputTokens: 4000,
maxOutputTokens: 1000,
maxTotalCost: 10.0, // $10 per period
preferredModels: ["claude-3-sonnet", "gpt-3.5-turbo"],
budgetPeriod: "daily",
});

async function optimizedSampling(prompt: string) {
// 检查缓存
const cached = costController.checkCache(prompt);
if (cached) {
console.log("[CostController] Cache hit!");
return cached.result;
}