Skip to main content
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)
ParameterTypeDescription
messageslist[SamplingMessage]Prompt or conversation messages
systemPromptstr | NoneInstructions for the LLM
temperaturefloat | NoneRandomness/creativity
maxTokensintMaximum output tokens (required)
modelstr | NoneOptional model hint
stopSequenceslist[str] | NoneStop strings
includeContext"none" | "thisServer" | "allServers" | NoneWhether the client should include additional context
modelPreferencesModelPreferences | NoneModel selection hints (client may ignore)
metadatadict[str, object] | NoneOpaque 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}"