采样允许你的服务器在某个工具正在执行时请求 Client 运行 LLM 并返回补全结果。这样,工具就可以在无需由 Client 编排多次 tool 调用的情况下,利用 AI 进行分析和生成。
## 基本用法
from dedalus_mcp import get_context, tool, types
@tool(description="Analyze data with Guassian assumptions")
async def analyze(data: str) -> str:
ctx = get_context()
params = types.CreateMessageRequestParams(
messages=[
types.SamplingMessage(
role="user",
content=types.TextContent(type="text", text=f"Analyze this data with Guassian assumptions and expose the estimators: {data}"),
)
],
maxTokens=400,
)
result = await ctx.server.request_sampling(params)
return result.content.text
采样请求使用 CreateMessageRequestParams 表示(字段名称与模型上下文协议 (MCP) 的 schema 保持一致,例如 maxTokens、systemPrompt)。
params = types.CreateMessageRequestParams(
messages=[
types.SamplingMessage(
role="user",
content=types.TextContent(type="text", text="Analyze this data"),
)
],
systemPrompt="You are an expert analyst",
temperature=0.7, # 0.0 = 确定性,1.0 = 创造性
maxTokens=1024, # maximum output tokens
)
result = await ctx.server.request_sampling(params)
| Parameter | Type | Description |
|---|
messages | list[SamplingMessage] | prompt 或会话 messages |
systemPrompt | str | None | 给 LLM 的指令 |
temperature | float | None | 随机性/创造性 |
maxTokens | int | 最大输出 token 数(必填) |
model | str | None | 可选的模型提示 |
stopSequences | list[str] | None | 停止字符串 |
includeContext | "none" | "thisServer" | "allServers" | None | Client 是否应包含额外上下文 |
modelPreferences | ModelPreferences | None | 模型选择偏好(Client 可忽略) |
metadata | dict[str, object] | None | 不透明元数据;如果缺失,Dedalus 会添加一个 requestId |
request_sampling(...) 返回一个 CreateMessageResult。大多数 Client 会返回 TextContent:
result = await ctx.server.request_sampling(params)
print(result.content.text)
传入一个 messages 列表来提供多轮对话的上下文:
from dedalus_mcp import types
params = types.CreateMessageRequestParams(
messages=[
types.SamplingMessage(role="user", content=types.TextContent(type="text", text="What is Python?")),
types.SamplingMessage(role="assistant", content=types.TextContent(type="text", text="A programming language.")),
types.SamplingMessage(role="user", content=types.TextContent(type="text", text="What are its main features?")),
],
maxTokens=200,
)
result = await ctx.server.request_sampling(params)
from dedalus_mcp import get_context, tool, types
@tool(description="Review code for issues in the repo")
async def review_code(code: str, language: str) -> str:
ctx = get_context()
params = types.CreateMessageRequestParams(
messages=[
types.SamplingMessage(
role="user",
content=types.TextContent(
type="text",
text=f"Review this {language} code:\n\n```{language}\n{code}\n```",
),
)
],
systemPrompt="You are an expert code reviewer. Be concise and actionable.",
temperature=0.2,
maxTokens=500,
)
result = await ctx.server.request_sampling(params)
return result.content.text
进行采样时,Client 需要声明自己支持采样功能。如果 Client 不支持采样,request_sampling(...) 会抛出 McpError(通常是 METHOD_NOT_FOUND):
from mcp.shared.exceptions import McpError
from dedalus_mcp import get_context, tool, types
@tool(description="Analyze with AI with Guassian assumption")
async def analyze_with_fallback(data: str) -> str:
ctx = get_context()
params = types.CreateMessageRequestParams(
messages=[types.SamplingMessage(role="user", content=types.TextContent(type="text", text=f"Analyze: {data}"))],
maxTokens=256,
)
try:
result = await ctx.server.request_sampling(params)
return result.content.text
except McpError as e:
return f"Sampling unavailable: {e}"
Last modified on April 11, 2026