Sampling lets your server ask the client to run an LLM and return a completion while a tool is executing. This enables tools to leverage AI for analysis and generation without the client having to orchestrate multiple tool calls.
Basic usage
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
Parameters
Sampling requests are expressed as CreateMessageRequestParams (field names match the MCP schema, e.g. 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 = deterministic, 1.0 = creative
maxTokens=1024, # maximum output tokens
)
result = await ctx.server.request_sampling(params)
| Parameter | Type | Description |
|---|
messages | list[SamplingMessage] | Prompt or conversation messages |
systemPrompt | str | None | Instructions for the LLM |
temperature | float | None | Randomness/creativity |
maxTokens | int | Maximum output tokens (required) |
model | str | None | Optional model hint |
stopSequences | list[str] | None | Stop strings |
includeContext | "none" | "thisServer" | "allServers" | None | Whether the client should include additional context |
modelPreferences | ModelPreferences | None | Model selection hints (client may ignore) |
metadata | dict[str, object] | None | Opaque metadata; Dedalus will add a requestId if missing |
Response
request_sampling(...) returns a CreateMessageResult. Most clients return TextContent:
result = await ctx.server.request_sampling(params)
print(result.content.text)
Multi-turn conversations
Pass a list of messages for multi-turn context:
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)
Example: Code review
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
Error handling
Sampling requires the client to advertise the sampling capability. If the client doesn’t support sampling, request_sampling(...) raises McpError (typically 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 January 27, 2026
