fastmcp/docs/servers/sampling.mdx
2025-12-09 19:29:02 -05:00

501 lines
17 KiB
Text

---
title: LLM Sampling
sidebarTitle: Sampling
description: Request LLM text generation from the client or a configured provider through the MCP context.
icon: robot
tag: NEW
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="2.0.0" />
LLM sampling allows MCP tools to request LLM completions based on provided messages. By default, sampling requests are sent to the client's LLM, but you can also configure a fallback handler or always use a specific LLM provider.
## Why Use LLM Sampling?
LLM sampling enables tools to:
- **Leverage AI capabilities**: Use the client's LLM for text generation and analysis
- **Offload complex reasoning**: Let the LLM handle tasks requiring natural language understanding
- **Execute agentic workflows**: Let the LLM call tools during sampling to accomplish complex tasks
- **Get structured output**: Request validated, typed responses using `result_type`
- **Generate dynamic content**: Create responses, summaries, or transformations based on data
## Basic Usage
Use `ctx.sample()` to request text generation from the client's LLM. The method returns a `SamplingResult` with `.text`, `.result`, and `.history` attributes:
```python
from fastmcp import FastMCP, Context
mcp = FastMCP("SamplingDemo")
@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> dict:
"""Analyze the sentiment of text using the client's LLM."""
prompt = f"""Analyze the sentiment of the following text as positive, negative, or neutral.
Just output a single word - 'positive', 'negative', or 'neutral'.
Text to analyze: {text}"""
# Request LLM analysis - returns SamplingResult
response = await ctx.sample(prompt)
# Access the text via .text or .result (same for text responses)
sentiment = response.text.strip().lower()
if "positive" in sentiment:
sentiment = "positive"
elif "negative" in sentiment:
sentiment = "negative"
else:
sentiment = "neutral"
return {"text": text, "sentiment": sentiment}
```
## Method Signature
<Card icon="code" title="Context Sampling Method">
<ResponseField name="ctx.sample" type="async method">
Request text generation from the client's LLM
<Expandable title="Parameters">
<ResponseField name="messages" type="str | list[str | SamplingMessage]">
A string or list of strings/message objects to send to the LLM
</ResponseField>
<ResponseField name="system_prompt" type="str | None" default="None">
Optional system prompt to guide the LLM's behavior
</ResponseField>
<ResponseField name="temperature" type="float | None" default="None">
Optional sampling temperature (controls randomness, typically 0.0-1.0)
</ResponseField>
<ResponseField name="max_tokens" type="int | None" default="512">
Optional maximum number of tokens to generate
</ResponseField>
<ResponseField name="model_preferences" type="ModelPreferences | str | list[str] | None" default="None">
Optional model selection preferences (e.g., model hint string, list of hints, or ModelPreferences object)
</ResponseField>
<ResponseField name="tools" type="list[SamplingTool | Tool] | None" default="None">
Optional list of tools the LLM can use during sampling. See [Sampling with Tools](#sampling-with-tools).
</ResponseField>
<ResponseField name="tool_choice" type="str | None" default="None">
Optional control over tool usage behavior: `"auto"`, `"required"`, or `"none"`
</ResponseField>
<ResponseField name="max_iterations" type="int" default="10">
Maximum number of LLM calls before returning. When tools are provided, FastMCP automatically executes tool calls and continues the conversation. On the last iteration, forces a text response (or `final_response` if using `result_type`).
</ResponseField>
<ResponseField name="result_type" type="type[T] | None" default="None">
Optional type for structured output. When specified, creates a synthetic `final_response` tool and validates the LLM's response against this type. See [Structured Output](#structured-output).
</ResponseField>
</Expandable>
<Expandable title="Response">
<ResponseField name="SamplingResult[T]" type="dataclass">
Always returns a `SamplingResult` with:
- `.text`: The text representation (raw text, or JSON for structured output). `None` if the LLM requested a tool on the last iteration.
- `.result`: The typed result. For text responses, equals `.text`. For structured output, the validated object.
- `.history`: List of all messages exchanged during sampling, useful for continuing conversations.
</ResponseField>
</Expandable>
</ResponseField>
</Card>
## Simple Text Generation
### Basic Prompting
Generate text with simple string prompts:
```python
@mcp.tool
async def generate_summary(content: str, ctx: Context) -> str:
"""Generate a summary of the provided content."""
prompt = f"Please provide a concise summary of the following content:\n\n{content}"
response = await ctx.sample(prompt)
return response.text
```
### System Prompt
Use system prompts to guide the LLM's behavior:
```python
@mcp.tool
async def generate_code_example(concept: str, ctx: Context) -> str:
"""Generate a Python code example for a given concept."""
response = await ctx.sample(
messages=f"Write a simple Python code example demonstrating '{concept}'.",
system_prompt="You are an expert Python programmer. Provide concise, working code examples without explanations.",
temperature=0.7,
max_tokens=300
)
return f"```python\n{response.text}\n```"
```
### Model Preferences
Specify model preferences for different use cases:
```python
@mcp.tool
async def creative_writing(topic: str, ctx: Context) -> str:
"""Generate creative content using a specific model."""
response = await ctx.sample(
messages=f"Write a creative short story about {topic}",
model_preferences="claude-haiku-4-5", # Prefer a specific model
temperature=0.9, # High creativity
max_tokens=1000
)
return response.text
@mcp.tool
async def technical_analysis(data: str, ctx: Context) -> str:
"""Perform technical analysis with a reasoning-focused model."""
response = await ctx.sample(
messages=f"Analyze this technical data and provide insights: {data}",
model_preferences=["claude-3-opus", "gpt-4"], # Prefer reasoning models
temperature=0.2, # Low randomness for consistency
max_tokens=800
)
return response.text
```
### Complex Message Structures
Use structured messages for more complex interactions:
```python
from fastmcp.client.sampling import SamplingMessage
@mcp.tool
async def multi_turn_analysis(user_query: str, context_data: str, ctx: Context) -> str:
"""Perform analysis using multi-turn conversation structure."""
messages = [
SamplingMessage(role="user", content=f"I have this data: {context_data}"),
SamplingMessage(role="assistant", content="I can see your data. What would you like me to analyze?"),
SamplingMessage(role="user", content=user_query)
]
response = await ctx.sample(
messages=messages,
system_prompt="You are a data analyst. Provide detailed insights based on the conversation context.",
temperature=0.3
)
return response.text
```
## Structured Output
<VersionBadge version="2.14.0" />
Use `result_type` to get validated, typed responses from the LLM. When you specify a `result_type`, FastMCP creates a synthetic `final_response` tool that the LLM calls to provide its structured response.
```python
from pydantic import BaseModel
from fastmcp import FastMCP, Context
mcp = FastMCP()
class SentimentResult(BaseModel):
sentiment: str
confidence: float
reasoning: str
@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> SentimentResult:
"""Analyze sentiment with structured output."""
result = await ctx.sample(
messages=f"Analyze the sentiment of: {text}",
result_type=SentimentResult,
)
# result.result is a validated SentimentResult object
return result.result
```
### How Structured Output Works
When you pass `result_type`:
1. FastMCP creates a synthetic `final_response` tool with a schema derived from your type
2. A hint is added to the system prompt: "Call final_response when you have completed the task."
3. The LLM calls `final_response` with its structured response
4. FastMCP validates the input against your type
5. If validation fails, the error is sent back to the LLM for retry
6. On success, `result.result` contains the validated object and `result.text` contains the JSON
## Sampling with Tools
<VersionBadge version="2.14.0" />
Sampling with tools enables agentic workflows where the LLM can request tool calls during sampling. FastMCP automatically executes the tools and continues the conversation until the LLM provides a final response.
### Creating Sampling Tools
Define regular Python functions and pass them directly to `ctx.sample()`:
```python
from fastmcp import FastMCP, Context
def search(query: str) -> str:
"""Search the web for information."""
return f"Search results for: {query}"
def get_current_time() -> str:
"""Get the current time."""
from datetime import datetime
return datetime.now().strftime("%H:%M:%S")
```
<Note>
To create a tool with a custom name or description, use `SamplingTool.from_function()`:
```python
from fastmcp.server.sampling import SamplingTool
tool = SamplingTool.from_function(
my_search_function,
name="web_search",
description="Search the web for information"
)
result = await ctx.sample(messages="...", tools=[tool])
```
</Note>
You can also pass existing FastMCP tools directly to `ctx.sample()`:
```python
mcp = FastMCP()
@mcp.tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"Weather in {city}: 72°F, sunny"
@mcp.tool
async def weather_agent(question: str, ctx: Context) -> str:
"""Answer weather questions using the weather tool."""
result = await ctx.sample(
messages=question,
tools=[get_weather], # FastMCP tools work directly
)
return result.text
```
### Using Tools in Sampling
Pass tools to `ctx.sample()` to enable agentic tool use. FastMCP automatically handles the tool execution loop:
```python
from fastmcp import FastMCP, Context
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
def get_time() -> str:
"""Get the current time."""
from datetime import datetime
return datetime.now().strftime("%H:%M:%S")
mcp = FastMCP()
@mcp.tool
async def research_assistant(question: str, ctx: Context) -> str:
"""Answer questions using search and calculation tools."""
result = await ctx.sample(
messages=question,
system_prompt="You are a research assistant. Use the available tools to answer questions.",
tools=[search, get_time],
)
return result.text
```
### Tool Loop
When you call `ctx.sample()` with tools, FastMCP automatically handles the execution loop:
1. The LLM receives your prompt and the available tools
2. If the LLM calls a tool, FastMCP executes it and sends the result back
3. This continues until the LLM provides a final text response
Tool errors are automatically caught and sent back to the LLM, allowing it to handle failures gracefully.
Use `max_iterations` to limit the number of LLM calls (default is 10). On the last iteration, FastMCP forces the LLM to return a final response rather than continuing to call tools:
```python
# Allow up to 5 tool-use cycles
result = await ctx.sample(
messages=prompt,
tools=[search, get_time],
max_iterations=5,
)
# Single iteration mode - forces text on first call
# Useful for building your own loop with result.history
result = await ctx.sample(
messages=prompt,
tools=[search, get_time],
max_iterations=1,
)
```
### Tool Choice Options
Control how the LLM uses tools with `tool_choice`:
```python
# Let the LLM decide (default)
result = await ctx.sample(messages=prompt, tools=tools, tool_choice="auto")
# Force the LLM to use a tool
result = await ctx.sample(messages=prompt, tools=tools, tool_choice="required")
# Prevent tool use (useful for final response)
result = await ctx.sample(messages=prompt, tools=tools, tool_choice="none")
```
### Structured Output
Combine `result_type` with tools for agentic workflows that return structured data:
```python
from pydantic import BaseModel
from fastmcp import FastMCP, Context
class ResearchResult(BaseModel):
summary: str
sources: list[str]
confidence: float
def search(query: str) -> str:
"""Search for information."""
return f"Found information about: {query}"
def fetch_url(url: str) -> str:
"""Fetch content from a URL."""
return f"Content from {url}"
mcp = FastMCP()
@mcp.tool
async def research(topic: str, ctx: Context) -> ResearchResult:
"""Research a topic and return structured results."""
result = await ctx.sample(
messages=f"Research this topic thoroughly: {topic}",
tools=[search, fetch_url],
result_type=ResearchResult,
)
# The LLM uses search/fetch_url as needed, then calls final_response
# result.result is the validated ResearchResult
return result.result
```
The LLM can call your tools as many times as needed, and when it's ready to provide the final answer, it calls `final_response` with the structured data.
### Building Custom Loops
The automatic tool loop handles most use cases, but sometimes you need finer control. Setting `max_iterations=1` runs a single sampling iteration and returns immediately, giving you access to `result.history` so you can continue the conversation yourself.
This is useful when you need to:
- Dynamically change which tools are available based on intermediate results
- Inject additional context or instructions mid-conversation
- Implement custom termination conditions beyond "LLM stopped calling tools"
- Add logging, metrics, or checkpointing between iterations
```python
@mcp.tool
async def custom_agent(question: str, ctx: Context) -> str:
"""Agent with custom loop logic."""
history = []
tools = [search, get_time]
while True:
result = await ctx.sample(
messages=history or question,
tools=tools,
max_iterations=1,
)
history = result.history
# Custom logic: check conditions, change tools, etc.
if some_condition:
tools = [different_tool]
# When we get a text response, we're done
if result.text:
return result.text
```
## Sampling Fallback Handler
Client support for sampling is optional. If the client does not support sampling, the server will report an error indicating that the client does not support sampling.
However, you can provide a `sampling_handler` to the FastMCP server, which sends sampling requests directly to an LLM provider instead of routing through the client. The `sampling_handler_behavior` parameter controls when this handler is used:
- **`"fallback"`** (default): Uses the handler only when the client doesn't support sampling. Requests go to the client first, falling back to the handler if needed.
- **`"always"`**: Always uses the handler, bypassing the client entirely. Useful when you want full control over the LLM used for sampling.
### Using Pre-built Handlers
The server fallback handler uses the same signature as client sampling handlers. FastMCP provides pre-built handlers for [Anthropic and OpenAI](/clients/sampling#pre-built-handlers) that work in both contexts:
```python
from fastmcp import FastMCP
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
server = FastMCP(
name="Sampling Server",
sampling_handler=AnthropicSamplingHandler(
default_model="claude-sonnet-4-5-20250929",
),
sampling_handler_behavior="fallback",
)
```
```python
from openai import OpenAI
from fastmcp import FastMCP
from fastmcp.server.sampling.openai import OpenAISamplingHandler
server = FastMCP(
name="Sampling Server",
sampling_handler=OpenAISamplingHandler(
default_model="gpt-4o-mini",
client=OpenAI(),
),
sampling_handler_behavior="always", # Always use server's LLM
)
```
## Client Requirements
By default, LLM sampling requires client support:
- Clients must implement sampling handlers to process requests (see [Client Sampling](/clients/sampling))
- If the client doesn't support sampling and no fallback handler is configured, `ctx.sample()` will raise an error
- Configure a `sampling_handler` with `sampling_handler_behavior="fallback"` to automatically handle clients that don't support sampling
- Use `sampling_handler_behavior="always"` to completely bypass the client and control which LLM is used
<Note>
Sampling with tools requires a client that advertises the `sampling.tools` capability. FastMCP clients automatically advertise this capability when a `sampling_handler` is provided. For external clients that don't support tool-enabled sampling, configure a fallback handler with `sampling_handler_behavior="always"`.
</Note>