ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 13:34:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

SEP-1577: Sampling with tools (#2551)

Browse source 
* MCP → SDK (vocab change only)

* WIP: Sampling API with SamplingResult[T] and result_type

* SEP-1577: Sampling with tools

- Add tools and result_type parameters to ctx.sample()
- Update OpenAI handler for tool content types
- Client advertises sampling.tools capability by default
- Collect tool results into single message with list content

* Fix tool result content handling in OpenAI handler

* Remove @sampling_tool decorator - pass functions directly to sample()

Functions passed to ctx.sample(tools=[...]) are now auto-converted
via SamplingTool.from_function(). Users can still use that method
directly for custom name/description overrides.

* Remove auto-conversion of MCP tools to sampling tools

Users want MCP tools passed to ctx.sample() to go through the full MCP
machinery (middleware, native responses) rather than being auto-converted
to direct function calls. Now only SamplingTool and plain callables are
accepted - passing a FastMCP Tool raises a clear TypeError.

Also bumps mcp dependency to >=1.24.0 for required sampling features.

* Refactor sampling API: replace sample_iter() with sample_step()

Replace the mutable SampleRun/sample_iter() pattern with a simpler stateless
sample_step() function. sample_step() makes a single LLM call and returns a
SampleStep with the response and history. sample() now loops sample_step()
internally.

Key changes:
- Add sample_step() for fine-grained control over the sampling loop
- Remove SampleRun class and sample_iter() method
- Structured output uses tool description only (no prompt modification)
- execute_tools parameter controls automatic vs manual tool execution

* Address CodeRabbit nitpicks

* Address CodeRabbit review feedback for sampling tools

- Fix temperature=0.0 being dropped due to falsy evaluation
- Add ToolChoice.name support for forcing specific tools
- Replace assert statements with explicit RuntimeError checks
- Add mask_error_details parameter to sample()/sample_step() with ToolError escape hatch
- Fix hasattr patterns with proper isinstance checks
- Document mask_error_details and add OpenAI prerequisites to docs

* Address additional CodeRabbit review feedback

- Catch ValidationError specifically instead of bare Exception
- Update result_type docs to mention dataclasses and basic types
- Raise ValueError for unknown tool_choice modes
- Validate sampling_handler_behavior to catch typos
- Remove ToolChoice.name handling (not part of MCP spec)
- Validate tool_choice string in sample_step()

* Review fixes for sampling tools PR

- Remove internal functions from sampling __init__.py exports
- Remove fragile is_text property, use not is_tool_use instead
- Inline call_client into context.py, remove from run.py
- Fix SamplingMessage docs to use TextContent
- Handle result.text being None in doc examples
- Simplify client sampling docs to recommend OpenAISamplingHandler
- Add sampling_capabilities override documentation
- Raise iteration limit from 50 to 100
- Remove _parse_model_preferences duplication
- Use AsyncOpenAI in OpenAISamplingHandler
- Fix tool_choice docstring

* Fix OpenAI handler tests to use AsyncOpenAI

* Address remaining CodeRabbit review comments

- Fix message ordering in OpenAI handler: tool results now correctly
  follow assistant message with tool_calls
- sample_step() now always includes assistant message in history
- Raise ValueError on JSON parse errors instead of silent {}
- Add has_sampling capability check when behavior is None
- Raise RuntimeError when structured output receives text response
- Wrap primitive result_type schemas in object wrapper
- Fix docs example using invalid SamplingMessage construction
- Add comprehensive client_sampling_test.py example

* Add return type annotation to OpenAISamplingHandler.__init__

* Use explicit 'is not None' check for sampling_capabilities defaulting

---------

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Bill Easton <strawgate@users.noreply.github.com>
This commit is contained in:
Jeremiah Lowin 2025-12-14 13:51:05 -05:00 • committed by GitHub
commit 41ec7ee06d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 2958 additions and 358 deletions
Download patch file Download diff file Expand all files Collapse all files

62
docs/clients/sampling.mdx
View file

@ -109,6 +109,14 @@ The sampling handler receives three parameters:
<ResponseField name="metadata" type="dict[str, Any] | None">
Optional metadata to pass through to the LLM provider.
</ResponseField>
<ResponseField name="tools" type="list[Tool] | None">
Optional list of tools the LLM can use during sampling. See [Using the OpenAI Handler](#using-the-openai-handler).
</ResponseField>
<ResponseField name="toolChoice" type="ToolChoice | None">
Optional control over tool usage behavior (`auto`, `required`, or `none`).
</ResponseField>
</Expandable>
</ResponseField>
@ -153,4 +161,56 @@ client = Client(
<Note>
If the client doesn't provide a sampling handler, servers can optionally configure a fallback handler. See [Server Sampling](/servers/sampling#sampling-fallback-handler) for details.
</Note>
</Note>
## Sampling Capabilities
When you provide a `sampling_handler`, FastMCP automatically advertises full sampling capabilities to the server, including tool support. To disable tool support (for simpler handlers that don't support tools), pass `sampling_capabilities` explicitly:
```python
from mcp.types import SamplingCapability
client = Client(
"my_mcp_server.py",
sampling_handler=basic_handler,
sampling_capabilities=SamplingCapability(), # No tool support
)
```
## Using the OpenAI Handler
<VersionBadge version="2.14.0" />
For full-featured sampling with tool support, use the built-in OpenAI handler. It handles message conversion, tool calls, and response formatting automatically:
```python
from fastmcp import Client
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
client = Client(
"my_mcp_server.py",
sampling_handler=OpenAISamplingHandler(default_model="gpt-4o"),
)
```
The handler works with any OpenAI-compatible API by passing a custom client:
```python
from openai import AsyncOpenAI
client = Client(
"my_mcp_server.py",
sampling_handler=OpenAISamplingHandler(
default_model="llama-3.1-70b",
client=AsyncOpenAI(base_url="http://localhost:8000/v1"),
),
)
```
<Note>
Tool execution happens on the server side. The client's role is to pass tools to the LLM and return the LLM's response (which may include tool use requests). The server then executes the tools and may send follow-up sampling requests with tool results.
</Note>
<Tip>
To implement a custom sampling handler, see the [OpenAISamplingHandler source code](https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/experimental/sampling/handlers/openai.py) as a reference.
</Tip>

618
docs/servers/sampling.mdx
View file

@ -3,268 +3,488 @@ 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'
import { VersionBadge } from "/snippets/version-badge.mdx";
<VersionBadge version="2.0.0" />
LLM sampling allows MCP tools to request LLM text generation 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. This is useful when tools need to leverage LLM capabilities to process data, generate responses, or perform text-based analysis.
LLM sampling allows your MCP tools to request text generation from an LLM during execution. This enables tools to leverage AI capabilities for analysis, generation, reasoning, and more—without the client needing to orchestrate multiple calls.
## Why Use LLM Sampling?
By default, sampling requests are routed to the client's LLM. You can also configure a fallback handler to use a specific provider (like OpenAI) when the client doesn't support sampling, or to always use your own LLM regardless of client capabilities.
LLM sampling enables tools to:
## Method Reference
- **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
- **Generate dynamic content**: Create responses, summaries, or transformations based on data
- **Maintain context**: Use the same LLM instance that the user is already interacting with
### Basic Usage
Use `ctx.sample()` to request text generation from the client's LLM:
```python {14}
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
response = await ctx.sample(prompt)
# Process the LLM's response
sentiment = response.text.strip().lower()
# Map to standard sentiment values
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">
<Card icon="code" title="ctx.sample()">
<ResponseField name="ctx.sample" type="async method">
Request text generation from the client's LLM
Request text generation from the LLM, running to completion automatically.
<Expandable title="Parameters">
<ResponseField name="messages" type="str | list[str | SamplingMessage]">
A string or list of strings/message objects to send to the LLM
The prompt to send. Can be a simple string or a list of messages for multi-turn conversations.
</ResponseField>
<ResponseField name="system_prompt" type="str | None" default="None">
Optional system prompt to guide the LLM's behavior
Instructions that establish the LLM's role and behavior.
</ResponseField>
<ResponseField name="temperature" type="float | None" default="None">
Optional sampling temperature (controls randomness, typically 0.0-1.0)
Controls randomness (0.0 = deterministic, 1.0 = creative).
</ResponseField>
<ResponseField name="max_tokens" type="int | None" default="512">
Optional maximum number of tokens to generate
Maximum 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 name="model_preferences" type="str | list[str] | None" default="None">
Hints for which model the client should use.
</ResponseField>
<ResponseField name="tools" type="list[Callable] | None" default="None">
Functions the LLM can call during sampling.
</ResponseField>
<ResponseField name="result_type" type="type[T] | None" default="None">
A type for validated structured output. Supports Pydantic models, dataclasses, and basic types like `int`, `list[str]`, or `dict[str, int]`.
</ResponseField>
<ResponseField name="mask_error_details" type="bool | None" default="None">
If True, mask detailed error messages from tool execution. When None (default), uses the global `settings.mask_error_details` value. Tools can raise `ToolError` to bypass masking and provide specific error messages to the LLM.
</ResponseField>
</Expandable>
<Expandable title="Response">
<ResponseField name="response" type="TextContent | ImageContent">
The LLM's response content (typically TextContent with a .text attribute)
<ResponseField name="SamplingResult[T]" type="dataclass">
- `.text`: The raw text response (or JSON for structured output)
- `.result`: The typed result—same as `.text` for plain text, or a validated Pydantic object for structured output
- `.history`: All messages exchanged during sampling
</ResponseField>
</Expandable>
</ResponseField>
</Card>
## Simple Text Generation
<Card icon="code" title="ctx.sample_step()">
<ResponseField name="ctx.sample_step" type="async method">
Make a single LLM sampling call. Use this for fine-grained control over the sampling loop.
### Basic Prompting
<Expandable title="Parameters">
Same as `sample()`, plus:
Generate text with simple string prompts:
<ResponseField name="tool_choice" type="str | None" default="None">
Controls tool usage: `"auto"`, `"required"`, or `"none"`.
</ResponseField>
<ResponseField name="execute_tools" type="bool" default="True">
If True, execute tool calls and append results to history. If False, return immediately with tool calls available for manual execution.
</ResponseField>
<ResponseField name="mask_error_details" type="bool | None" default="None">
If True, mask detailed error messages from tool execution. When None (default), uses the global `settings.mask_error_details` value. Tools can raise `ToolError` to bypass masking.
</ResponseField>
</Expandable>
<Expandable title="Response">
<ResponseField name="SampleStep" type="dataclass">
- `.response`: The raw LLM response
- `.history`: Messages including input, assistant response, and tool results
- `.is_tool_use`: True if the LLM requested tool execution
- `.tool_calls`: List of tool calls (if any)
- `.text`: The text content (if any)
</ResponseField>
</Expandable>
</ResponseField>
</Card>
## Basic Sampling
The simplest use of sampling is passing a prompt string to `ctx.sample()`. The method sends the prompt to the LLM, waits for the complete response, and returns a `SamplingResult`. You can access the generated text through the `.text` attribute.
```python
from fastmcp import FastMCP, Context
mcp = FastMCP()
```python {6}
@mcp.tool
async def generate_summary(content: str, ctx: Context) -> str:
async def summarize(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
result = await ctx.sample(f"Please summarize this:\n\n{content}")
return result.text or ""
```
### System Prompt
The `SamplingResult` also provides `.result` (identical to `.text` for plain text responses) and `.history` containing the full message exchange—useful if you need to continue the conversation or debug the interaction.
Use system prompts to guide the LLM's behavior:
### System Prompts
```python {4-9}
System prompts let you establish the LLM's role and behavioral guidelines before it processes your request. This is useful for controlling tone, enforcing constraints, or providing context that shouldn't clutter the user-facing prompt.
````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.",
async def generate_code(concept: str, ctx: Context) -> str:
"""Generate a Python code example for a concept."""
result = await ctx.sample(
messages=f"Write a Python example demonstrating '{concept}'.",
system_prompt=(
"You are an expert Python programmer. "
"Provide concise, working code without explanations."
),
temperature=0.7,
max_tokens=300
)
code_example = response.text
return f"```python\n{code_example}\n```"
```
return f"```python\n{result.text}\n```"
````
The `temperature` parameter controls randomness—higher values (up to 1.0) produce more varied outputs, while lower values make responses more deterministic. The `max_tokens` parameter limits response length.
### Model Preferences
Specify model preferences for different use cases:
```python {4-8, 17-22}
@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-3-sonnet", # Prefer a specific model
include_context="thisServer", # Use the server's context
temperature=0.9, # High creativity
max_tokens=1000
)
return response.text
Model preferences let you hint at which LLM the client should use for a request. You can pass a single model name or a list of preferences in priority order. These are hints rather than requirements—the actual model used depends on what the client has available.
```python
@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
"""Analyze data using a reasoning-focused model."""
result = await ctx.sample(
messages=f"Analyze this data:\n\n{data}",
model_preferences=["claude-opus-4-5", "gpt-5-2"],
temperature=0.2,
)
return response.text
return result.text or ""
```
### Complex Message Structures
Use model preferences when different tasks benefit from different model characteristics. Creative writing might prefer faster models with higher temperature, while complex analysis might benefit from larger reasoning-focused models.
Use structured messages for more complex interactions:
### Multi-Turn Conversations
```python {1, 6-10}
from fastmcp.client.sampling import SamplingMessage
For requests that need conversational context, construct a list of `SamplingMessage` objects representing the conversation history. Each message has a `role` ("user" or "assistant") and `content` (a `TextContent` object).
```python
from mcp.types import SamplingMessage, TextContent
@mcp.tool
async def multi_turn_analysis(user_query: str, context_data: str, ctx: Context) -> str:
"""Perform analysis using multi-turn conversation structure."""
async def contextual_analysis(query: str, data: str, ctx: Context) -> str:
"""Analyze data with conversational context."""
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
```
## 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.
Sampling handlers can be implemented using any LLM provider, but a sample implementation for OpenAI is provided as a Contrib module. Sampling lacks the full capabilities of typical LLM completions. For this reason, the OpenAI sampling handler, pointed at a third-party provider's OpenAI-compatible API, is often sufficient to implement a sampling handler.
### Fallback Mode (Default)
Uses the handler only when the client doesn't support sampling:
```python
import asyncio
import os
from mcp.types import ContentBlock
from openai import OpenAI
from fastmcp import FastMCP
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
from fastmcp.server.context import Context
async def async_main():
server = FastMCP(
name="OpenAI Sampling Fallback Example",
sampling_handler=OpenAISamplingHandler(
default_model="gpt-4o-mini",
client=OpenAI(
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL"),
),
SamplingMessage(
role="user",
content=TextContent(type="text", text=f"Here's my data: {data}"),
),
sampling_handler_behavior="fallback", # Default - only use when client doesn't support sampling
)
@server.tool
async def test_sample_fallback(ctx: Context) -> ContentBlock:
# Will use client's LLM if available, otherwise falls back to the handler
return await ctx.sample(
messages=["hello world!"],
)
await server.run_http_async()
if __name__ == "__main__":
asyncio.run(async_main())
SamplingMessage(
role="assistant",
content=TextContent(type="text", text="I see the data. What would you like to know?"),
),
SamplingMessage(
role="user",
content=TextContent(type="text", text=query),
),
]
result = await ctx.sample(messages=messages)
return result.text or ""
```
### Always Mode
The LLM receives the full conversation thread and responds with awareness of the preceding context.
Always uses the handler, bypassing the client:
## Structured Output
<VersionBadge version="2.14.1" />
When you need validated, typed data instead of free-form text, use the `result_type` parameter. FastMCP ensures the LLM returns data matching your type, handling validation and retries automatically. The `result_type` parameter accepts Pydantic models, dataclasses, and basic types like `int`, `list[str]`, or `dict[str, int]`.
```python
server = FastMCP(
name="Server-Controlled Sampling",
sampling_handler=OpenAISamplingHandler(
default_model="gpt-4o-mini",
client=OpenAI(api_key=os.getenv("API_KEY")),
),
sampling_handler_behavior="always", # Always use the handler, never the client
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 text sentiment with structured output."""
result = await ctx.sample(
messages=f"Analyze the sentiment of: {text}",
result_type=SentimentResult,
)
return result.result # A validated SentimentResult object
```
When you call this tool, the LLM returns a structured response that FastMCP validates against your Pydantic model. You access the validated object through `result.result`, while `result.text` contains the JSON representation.
<Note>
When you pass `result_type`, `sample()` automatically creates a
`final_response` tool that the LLM calls to provide its response. If
validation fails, the error is sent back to the LLM for retry. This automatic
handling only applies to `sample()`—with `sample_step()`, you must manage
structured output yourself.
</Note>
## Sampling with Tools
<VersionBadge version="2.14.1" />
Sampling with tools enables agentic workflows where the LLM can call functions to gather information before responding. This implements [SEP-1577](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1577), allowing the LLM to autonomously orchestrate multi-step operations.
Pass Python functions to the `tools` parameter, and FastMCP handles the execution loop automatically—calling tools, returning results to the LLM, and continuing until the LLM provides a final response.
### Defining Tools
Define regular Python functions with type hints and docstrings. FastMCP extracts the function's name, docstring, and parameter types to create tool schemas that the LLM can understand.
```python
from fastmcp import FastMCP, Context
def search(query: str) -> str:
"""Search the web for information."""
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(question: str, ctx: Context) -> str:
"""Answer questions using available tools."""
result = await ctx.sample(
messages=question,
tools=[search, get_time],
)
return result.text or ""
```
The LLM sees each function's signature and docstring, using this information to decide when and how to call them. Tool errors are caught and sent back to the LLM, allowing it to recover gracefully. An internal safety limit prevents infinite loops.
### Tool Error Handling
By default, when a sampling tool raises an exception, the error message (including details) is sent back to the LLM so it can attempt recovery. To prevent sensitive information from leaking to the LLM, use the `mask_error_details` parameter:
```python
result = await ctx.sample(
messages=question,
tools=[search],
mask_error_details=True, # Generic error messages only
)
```
When `mask_error_details=True`, tool errors become generic messages like `"Error executing tool 'search'"` instead of exposing stack traces or internal details.
To intentionally provide specific error messages to the LLM regardless of masking, raise `ToolError`:
```python
from fastmcp.exceptions import ToolError
def search(query: str) -> str:
"""Search for information."""
if not query.strip():
raise ToolError("Search query cannot be empty")
return f"Results for: {query}"
```
`ToolError` messages always pass through to the LLM, making it the escape hatch for errors you want the LLM to see and handle.
For custom names or descriptions, use `SamplingTool.from_function()`:
```python
from fastmcp.server.sampling import SamplingTool
tool = SamplingTool.from_function(
my_func,
name="custom_name",
description="Custom description"
)
@server.tool
async def analyze_data(data: str, ctx: Context) -> str:
# Will ALWAYS use the server's configured LLM, not the client's
result = await ctx.sample(
messages=f"Analyze this data: {data}",
system_prompt="You are a data analyst.",
)
return result.text
result = await ctx.sample(messages="...", tools=[tool])
```
## Client Requirements
### Combining with Structured Output
By default, LLM sampling requires client support:
Combine tools with `result_type` for agentic workflows that return validated, structured data. The LLM uses your tools to gather information, then returns a response matching your type.
- 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
```python
result = await ctx.sample(
messages="Research Python async patterns",
tools=[search, fetch_url],
result_type=ResearchResult,
)
```
## Loop Control
<VersionBadge version="2.14.1" />
While `sample()` handles the tool execution loop automatically, some scenarios require fine-grained control over each step. The `sample_step()` method makes a single LLM call and returns a `SampleStep` containing the response and updated history.
Unlike `sample()`, `sample_step()` is stateless—it doesn't remember previous calls. You control the conversation by passing the full message history each time. The returned `step.history` includes all messages up through the current response, making it easy to continue the loop.
Use `sample_step()` when you need to:
- Inspect tool calls before they execute
- Implement custom termination conditions
- Add logging, metrics, or checkpointing between steps
- Build custom agentic loops with domain-specific logic
### Using sample_step()
By default, `sample_step()` executes any tool calls and includes the results in the history. Call it in a loop, passing the updated history each time, until a stop condition is met.
```python
from mcp.types import SamplingMessage
@mcp.tool
async def controlled_agent(question: str, ctx: Context) -> str:
"""Agent with manual loop control."""
messages: list[str | SamplingMessage] = [question] # strings auto-convert
while True:
step = await ctx.sample_step(
messages=messages,
tools=[search, get_time],
)
if step.is_tool_use:
# Tools already executed (execute_tools=True by default)
# Log what was called before continuing
for call in step.tool_calls:
print(f"Called tool: {call.name}")
if not step.is_tool_use:
return step.text or ""
# Continue with updated history
messages = step.history
```
### SampleStep Properties
Each `SampleStep` provides information about what the LLM returned:
- `step.is_tool_use` — True if the LLM requested tool calls
- `step.tool_calls` — List of tool calls requested (if any)
- `step.text` — The text content (if any)
- `step.history` — All messages exchanged so far
The contents of `step.history` depend on `execute_tools`:
- **`execute_tools=True`** (default): Includes tool results, ready for the next iteration
- **`execute_tools=False`**: Includes the assistant's tool request, but you add results yourself
### Manual Tool Execution
Set `execute_tools=False` to handle tool execution yourself. When disabled, `step.history` contains the user message and the assistant's response with tool calls—but no tool results. You execute the tools and append the results as a user message.
```python
from mcp.types import SamplingMessage, ToolResultContent, TextContent
from fastmcp import FastMCP, Context
mcp = FastMCP()
@mcp.tool
async def research(question: str, ctx: Context) -> str:
"""Research with manual tool handling."""
def search(query: str) -> str:
return f"Results for: {query}"
def get_time() -> str:
return "12:00 PM"
# Map tool names to functions
tools = {"search": search, "get_time": get_time}
messages: list[SamplingMessage] = [question] # strings are converted automatically
while True:
step = await ctx.sample_step(
messages=messages,
tools=list(tools.values()),
execute_tools=False,
)
if not step.is_tool_use:
return step.text or ""
# Execute tools and collect results
tool_results = []
for call in step.tool_calls:
fn = tools[call.name]
result = fn(**call.input)
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=call.id,
content=[TextContent(type="text", text=result)],
)
)
messages = list(step.history)
messages.append(SamplingMessage(role="user", content=tool_results))
```
#### Error Handling
To report an error, set `isError=True`. The LLM will see the error and can decide how to proceed:
```python
tool_result = ToolResultContent(
type="tool_result",
toolUseId=call.id,
content=[TextContent(type="text", text="Permission denied")],
isError=True,
)
```
## Fallback Handlers
Client support for sampling is optional—some clients may not implement it. To ensure your tools work regardless of client capabilities, configure a `sampling_handler` that sends requests directly to an LLM provider.
### OpenAI Handler
FastMCP provides an OpenAI-compatible handler that works with OpenAI's API and compatible providers. It supports the full sampling API including tools, automatically converting your Python functions to OpenAI's function calling format.
<Note>
The OpenAI handler requires the `openai` package. Install it with:
```bash
pip install fastmcp[openai]
# or
pip install openai
```
You'll also need to set the `OPENAI_API_KEY` environment variable or pass it directly to the client.
</Note>
```python
import os
from openai import OpenAI
from fastmcp import FastMCP
from fastmcp.experimental.sampling.handlers.openai import (
OpenAISamplingHandler,
)
server = FastMCP(
name="My Server",
sampling_handler=OpenAISamplingHandler(
default_model="gpt-4o-mini",
client=OpenAI(api_key=os.getenv("OPENAI_API_KEY")),
),
sampling_handler_behavior="fallback",
)
```
### Behavior Modes
The `sampling_handler_behavior` parameter controls when the handler is used:
- **`"fallback"`** (default): Use the handler only when the client doesn't support sampling. This lets capable clients use their own LLM while ensuring your tools still work with clients that lack sampling support.
- **`"always"`**: Always use the handler, bypassing the client entirely. Use this when you need guaranteed control over which LLM processes requests—for cost control, compliance requirements, or when specific model characteristics are essential.
<Note>
Sampling with tools requires the client to advertise the `sampling.tools`
capability. FastMCP clients do this automatically. For external clients that
don't support tool-enabled sampling, configure a fallback handler with
`sampling_handler_behavior="always"`.
</Note>

40
examples/advanced_sampling/README.md Normal file
View file

@ -0,0 +1,40 @@
# Advanced Sampling Examples
These examples demonstrate FastMCP's sampling API with real LLM backends.
## Prerequisites
```bash
pip install fastmcp[openai]
export OPENAI_API_KEY=your-key
```
## Examples
### Structured Output (`structured_output.py`)
Uses `result_type` to get validated Pydantic models from the LLM:
```bash
python examples/advanced_sampling/structured_output.py
```
### Tool Use (`tool_use.py`)
Gives the LLM tools to use during sampling, with automatic tool execution:
```bash
python examples/advanced_sampling/tool_use.py
```
### Client Sampling Test (`client_sampling_test.py`)
Comprehensive test of advanced sampling features:
- Primitive `result_type` (`int`, `list[str]`) with automatic schema wrapping
- `sample_step()` for fine-grained loop control
- History tracking (verifies assistant messages are included)
- Multi-step reasoning with tools
```bash
python examples/advanced_sampling/client_sampling_test.py
```

159
examples/advanced_sampling/client_sampling_test.py Normal file
View file

@ -0,0 +1,159 @@
"""
Client Sampling Test
This example demonstrates advanced sampling features using a Client with an
OpenAI handler. It tests:
- Primitive result_type (int, list[str]) with automatic schema wrapping
- The sample_step() method for fine-grained loop control
- History tracking including assistant messages
- Tool execution with manual control
Prerequisites:
pip install fastmcp[openai]
export OPENAI_API_KEY=your-key
Run:
python examples/advanced_sampling/client_sampling_test.py
"""
import asyncio
from pydantic import BaseModel
from fastmcp import Client, Context, FastMCP
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
# Create the MCP server
mcp = FastMCP("Sampling Test Server")
# --- Test 1: Primitive result_type (int) ---
@mcp.tool
async def count_vowels(text: str, ctx: Context) -> int:
"""Count the number of vowels in the given text using LLM sampling."""
result = await ctx.sample(
messages=f"Count the number of vowels (a, e, i, o, u) in this text and return just the count as an integer:\n\n{text}",
system_prompt="You are a precise counting assistant. Return only the numeric count.",
result_type=int,
)
return result.result
# --- Test 2: Primitive result_type (list[str]) ---
@mcp.tool
async def extract_keywords(text: str, ctx: Context) -> list[str]:
"""Extract keywords from the given text."""
result = await ctx.sample(
messages=f"Extract the 3 most important keywords from this text:\n\n{text}",
system_prompt="You are a keyword extraction expert. Return a list of keywords.",
result_type=list[str],
)
return result.result
# --- Test 3: sample_step() with history tracking ---
@mcp.tool
async def multi_step_reasoning(question: str, ctx: Context) -> str:
"""Demonstrate multi-step reasoning with sample_step()."""
def think(thought: str) -> str:
"""Record a thought in the reasoning chain."""
return f"Thought recorded: {thought}"
# First step: initial reasoning
messages: list[str] = [question]
step_count = 0
max_steps = 5
while step_count < max_steps:
step = await ctx.sample_step(
messages=messages,
tools=[think],
system_prompt="Think step by step. Use the think() tool to record your reasoning, then provide your final answer.",
)
step_count += 1
# History should always include the assistant's response
assert len(step.history) > len(messages), (
"History should grow with assistant message"
)
if not step.is_tool_use:
return f"Answer (after {step_count} steps): {step.text or ''}"
# Continue with updated history
messages = step.history
return "Max steps reached without final answer"
# --- Test 4: Structured output with Pydantic model ---
class AnalysisResult(BaseModel):
main_topic: str
sentiment: str
word_count: int
@mcp.tool
async def analyze_text(text: str, ctx: Context) -> dict:
"""Analyze text and return structured results."""
result = await ctx.sample(
messages=f"Analyze this text:\n\n{text}",
system_prompt="Analyze the text and provide structured results.",
result_type=AnalysisResult,
)
return result.result.model_dump()
async def main():
sampling_handler = OpenAISamplingHandler(default_model="gpt-4o-mini")
async with Client(mcp, sampling_handler=sampling_handler) as client:
print("=" * 60)
print("Test 1: Primitive result_type (int)")
print("=" * 60)
result = await client.call_tool(
"count_vowels",
{"text": "Hello, world!"},
)
print(f" Vowel count: {result.data}")
print()
print("=" * 60)
print("Test 2: Primitive result_type (list[str])")
print("=" * 60)
result = await client.call_tool(
"extract_keywords",
{
"text": "FastMCP is a Python framework for building Model Context Protocol servers and clients."
},
)
print(f" Keywords: {result.data}")
print()
print("=" * 60)
print("Test 3: sample_step() with history tracking")
print("=" * 60)
result = await client.call_tool(
"multi_step_reasoning",
{"question": "What is 15 + 27? Think step by step."},
)
print(f" Result: {result.data}")
print()
print("=" * 60)
print("Test 4: Structured output (Pydantic model)")
print("=" * 60)
result = await client.call_tool(
"analyze_text",
{"text": "I really enjoyed learning about FastMCP today!"},
)
print(f" Analysis: {result.data}")
print()
print("All tests completed successfully!")
if __name__ == "__main__":
asyncio.run(main())

68
examples/advanced_sampling/structured_output.py Normal file
View file

@ -0,0 +1,68 @@
"""
Structured Output Example
This example demonstrates using `result_type` to get structured responses from an LLM.
The server exposes a tool that uses sampling to analyze text sentiment, returning
a structured Pydantic model instead of raw text.
Prerequisites:
pip install fastmcp[openai]
export OPENAI_API_KEY=your-key
Run:
python examples/advanced_sampling/structured_output.py
"""
import asyncio
from pydantic import BaseModel
from fastmcp import Client, Context, FastMCP
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
# Define a structured output model
class SentimentAnalysis(BaseModel):
sentiment: str # "positive", "negative", or "neutral"
confidence: float # 0.0 to 1.0
keywords: list[str] # Key words that influenced the analysis
# Create an MCP server with a sampling tool
mcp = FastMCP("Sentiment Analyzer")
@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> dict:
"""Analyze the sentiment of the given text."""
result = await ctx.sample(
messages=f"Analyze the sentiment of this text:\n\n{text}",
system_prompt="You are a sentiment analysis expert. Analyze text and return structured results.",
result_type=SentimentAnalysis,
)
# result.result is a validated SentimentAnalysis instance
return result.result.model_dump() # type: ignore[attr-defined]
async def main():
# Create an OpenAI-backed sampling handler
sampling_handler = OpenAISamplingHandler(default_model="gpt-4o-mini")
# Connect to the server with the sampling handler
async with Client(mcp, sampling_handler=sampling_handler) as client:
# Call the tool with some test text
result = await client.call_tool(
"analyze_sentiment",
{
"text": "I absolutely love this product! It exceeded all my expectations."
},
)
print("Analysis Result:")
print(f" Sentiment: {result.data['sentiment']}")
print(f" Confidence: {result.data['confidence']:.1%}")
print(f" Keywords: {', '.join(result.data['keywords'])}")
if __name__ == "__main__":
asyncio.run(main())

84
examples/advanced_sampling/tool_use.py Normal file
View file

@ -0,0 +1,84 @@
"""
Tool Use Example
This example demonstrates sampling with tools, where the LLM can use helper
functions to complete a task. The server exposes a "research assistant" tool
that uses sampling with search capabilities.
Prerequisites:
pip install fastmcp[openai]
export OPENAI_API_KEY=your-key
Run:
python examples/advanced_sampling/tool_use.py
"""
import asyncio
from pydantic import BaseModel, Field
from fastmcp import Client, Context, FastMCP
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
# Define tools (available to the LLM during sampling)
def search_web(query: str) -> str:
"""Search the web for information."""
# Simulated search results
results = {
"python async": "Python's asyncio provides async/await syntax for concurrent code.",
"fastmcp": "FastMCP is a framework for building MCP servers and clients in Python.",
"mcp protocol": "MCP (Model Context Protocol) enables AI models to use tools and resources.",
}
for key, value in results.items():
if key in query.lower():
return value
return f"No results found for: {query}"
def get_word_count(text: str) -> str:
"""Count words in text."""
return str(len(text.split()))
# Structured output for the final response
class ResearchReport(BaseModel):
summary: str
sources_used: list[str]
confidence: float = Field(ge=0.0, le=1.0)
# Create the MCP server
mcp = FastMCP("Research Assistant")
@mcp.tool
async def research(question: str, ctx: Context) -> dict:
"""Research a question using available tools and return a structured report."""
result = await ctx.sample(
messages=f"Research this question and provide a comprehensive answer:\n\n{question}",
system_prompt="You are a research assistant. Use the available tools to gather information, then call final_response with your structured report.",
tools=[search_web, get_word_count],
result_type=ResearchReport,
)
return result.result.model_dump() # type: ignore[attr-defined]
async def main():
sampling_handler = OpenAISamplingHandler(default_model="gpt-4o-mini")
async with Client(mcp, sampling_handler=sampling_handler) as client:
result = await client.call_tool(
"research",
{"question": "What is FastMCP and how does it relate to the MCP protocol?"},
)
print("Research Report:")
print(f" Summary: {result.data['summary']}")
print(f" Sources: {', '.join(result.data['sources_used'])}")
print(f" Confidence: {result.data['confidence'] * 100:.0f}%")
if __name__ == "__main__":
asyncio.run(main())

2
pyproject.toml
View file

@ -7,7 +7,7 @@ dependencies = [
"python-dotenv>=1.1.0",
"exceptiongroup>=1.2.2",
"httpx>=0.28.1",
"mcp>=1.23.1",
"mcp>=1.24.0",
"openapi-pydantic>=0.5.1",
"platformdirs>=4.0.0",
"pydocket>=0.15.5",

21
src/fastmcp/client/client.py
View file

@ -249,6 +249,7 @@ class Client(Generic[ClientTransportT]):
name: str | None = None,
roots: RootsList | RootsHandler | None = None,
sampling_handler: ClientSamplingHandler | None = None,
sampling_capabilities: mcp.types.SamplingCapability | None = None,
elicitation_handler: ElicitationHandler | None = None,
log_handler: LogHandler | None = None,
message_handler: MessageHandlerT | MessageHandler | None = None,
@ -306,6 +307,14 @@ class Client(Generic[ClientTransportT]):
self._session_kwargs["sampling_callback"] = create_sampling_callback(
sampling_handler
)
# Default to tools-enabled capabilities unless explicitly overridden
self._session_kwargs["sampling_capabilities"] = (
sampling_capabilities
if sampling_capabilities is not None
else mcp.types.SamplingCapability(
tools=mcp.types.SamplingToolsCapability()
)
)
if elicitation_handler is not None:
self._session_kwargs["elicitation_callback"] = create_elicitation_callback(
@ -357,11 +366,21 @@ class Client(Generic[ClientTransportT]):
"""Set the roots for the client. This does not automatically call `send_roots_list_changed`."""
self._session_kwargs["list_roots_callback"] = create_roots_callback(roots)
def set_sampling_callback(self, sampling_callback: ClientSamplingHandler) -> None:
def set_sampling_callback(
self,
sampling_callback: ClientSamplingHandler,
sampling_capabilities: mcp.types.SamplingCapability | None = None,
) -> None:
"""Set the sampling callback for the client."""
self._session_kwargs["sampling_callback"] = create_sampling_callback(
sampling_callback
)
# Default to tools-enabled capabilities unless explicitly overridden
self._session_kwargs["sampling_capabilities"] = (
sampling_capabilities
if sampling_capabilities is not None
else mcp.types.SamplingCapability(tools=mcp.types.SamplingToolsCapability())
)
def set_elicitation_callback(
self, elicitation_callback: ElicitationHandler

1
src/fastmcp/client/transports.py
View file

@ -66,6 +66,7 @@ class SessionKwargs(TypedDict, total=False):
read_timeout_seconds: datetime.timedelta | None
sampling_callback: SamplingFnT | None
sampling_capabilities: mcp.types.SamplingCapability | None
list_roots_callback: ListRootsFnT | None
logging_callback: LoggingFnT | None
elicitation_callback: ElicitationFnT | None

285
src/fastmcp/experimental/sampling/handlers/openai.py
View file

@ -1,26 +1,38 @@
import json
from collections.abc import Iterator, Sequence
from typing import get_args
from typing import Any, get_args
from mcp import ClientSession, ServerSession
from mcp.shared.context import LifespanContextT, RequestContext
from mcp.types import CreateMessageRequestParams as SamplingParams
from mcp.types import (
CreateMessageResult,
CreateMessageResultWithTools,
ModelPreferences,
SamplingMessage,
StopReason,
TextContent,
Tool,
ToolChoice,
ToolResultContent,
ToolUseContent,
)
try:
from openai import NOT_GIVEN, OpenAI
from openai import NOT_GIVEN, AsyncOpenAI, NotGiven
from openai.types.chat import (
ChatCompletion,
ChatCompletionAssistantMessageParam,
ChatCompletionMessageParam,
ChatCompletionMessageToolCallParam,
ChatCompletionSystemMessageParam,
ChatCompletionToolChoiceOptionParam,
ChatCompletionToolMessageParam,
ChatCompletionToolParam,
ChatCompletionUserMessageParam,
)
from openai.types.shared.chat_model import ChatModel
from openai.types.shared_params import FunctionDefinition
except ImportError as e:
raise ImportError(
"The `openai` package is not installed. Please install `fastmcp[openai]` or add `openai` to your dependencies manually."
@ -32,8 +44,12 @@ from fastmcp.experimental.sampling.handlers.base import BaseLLMSamplingHandler
class OpenAISamplingHandler(BaseLLMSamplingHandler):
def __init__(self, default_model: ChatModel, client: OpenAI | None = None):
self.client: OpenAI = client or OpenAI()
def __init__(
self,
default_model: ChatModel,
client: AsyncOpenAI | None = None,
) -> None:
self.client: AsyncOpenAI = client or AsyncOpenAI()
self.default_model: ChatModel = default_model
@override
@ -43,7 +59,7 @@ class OpenAISamplingHandler(BaseLLMSamplingHandler):
params: SamplingParams,
context: RequestContext[ServerSession, LifespanContextT]
| RequestContext[ClientSession, LifespanContextT],
) -> CreateMessageResult:
) -> CreateMessageResult | CreateMessageResultWithTools:
openai_messages: list[ChatCompletionMessageParam] = (
self._convert_to_openai_messages(
system_prompt=params.systemPrompt,
@ -53,14 +69,31 @@ class OpenAISamplingHandler(BaseLLMSamplingHandler):
model: ChatModel = self._select_model_from_preferences(params.modelPreferences)
response = self.client.chat.completions.create(
# Convert MCP tools to OpenAI format
openai_tools: list[ChatCompletionToolParam] | NotGiven = NOT_GIVEN
if params.tools:
openai_tools = self._convert_tools_to_openai(params.tools)
# Convert tool_choice to OpenAI format
openai_tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN
if params.toolChoice:
openai_tool_choice = self._convert_tool_choice_to_openai(params.toolChoice)
response = await self.client.chat.completions.create(
model=model,
messages=openai_messages,
temperature=params.temperature or NOT_GIVEN,
temperature=(
params.temperature if params.temperature is not None else NOT_GIVEN
),
max_tokens=params.maxTokens,
stop=params.stopSequences or NOT_GIVEN,
stop=params.stopSequences if params.stopSequences else NOT_GIVEN,
tools=openai_tools,
tool_choice=openai_tool_choice,
)
# Return appropriate result type based on whether tools were provided
if params.tools:
return self._chat_completion_to_result_with_tools(response)
return self._chat_completion_to_create_message_result(response)
@staticmethod
@ -121,23 +154,136 @@ class OpenAISamplingHandler(BaseLLMSamplingHandler):
)
continue
if not isinstance(message.content, TextContent):
raise ValueError("Only text content is supported")
content = message.content
if message.role == "user":
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=message.content.text,
)
)
else:
# Handle list content (from CreateMessageResultWithTools)
if isinstance(content, list):
# Collect tool calls and text from the list
tool_calls: list[ChatCompletionMessageToolCallParam] = []
text_parts: list[str] = []
# Collect tool results separately to maintain correct ordering
tool_messages: list[ChatCompletionToolMessageParam] = []
for item in content:
if isinstance(item, ToolUseContent):
tool_calls.append(
ChatCompletionMessageToolCallParam(
id=item.id,
type="function",
function={
"name": item.name,
"arguments": json.dumps(item.input),
},
)
)
elif isinstance(item, TextContent):
text_parts.append(item.text)
elif isinstance(item, ToolResultContent):
# Collect tool results (added after assistant message)
content_text = ""
if item.content:
result_texts = []
for sub_item in item.content:
if isinstance(sub_item, TextContent):
result_texts.append(sub_item.text)
content_text = "\n".join(result_texts)
tool_messages.append(
ChatCompletionToolMessageParam(
role="tool",
tool_call_id=item.toolUseId,
content=content_text,
)
)
# Add assistant message with tool calls if present
# OpenAI requires: assistant (with tool_calls) -> tool messages
if tool_calls or text_parts:
msg_content = "\n".join(text_parts) if text_parts else None
if tool_calls:
openai_messages.append(
ChatCompletionAssistantMessageParam(
role="assistant",
content=msg_content,
tool_calls=tool_calls,
)
)
# Add tool messages AFTER assistant message
openai_messages.extend(tool_messages)
elif msg_content:
if message.role == "user":
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=msg_content,
)
)
else:
openai_messages.append(
ChatCompletionAssistantMessageParam(
role="assistant",
content=msg_content,
)
)
elif tool_messages:
# Tool results only (assistant message was in previous message)
openai_messages.extend(tool_messages)
continue
# Handle ToolUseContent (assistant's tool calls)
if isinstance(content, ToolUseContent):
openai_messages.append(
ChatCompletionAssistantMessageParam(
role="assistant",
content=message.content.text,
tool_calls=[
ChatCompletionMessageToolCallParam(
id=content.id,
type="function",
function={
"name": content.name,
"arguments": json.dumps(content.input),
},
)
],
)
)
continue
# Handle ToolResultContent (user's tool results)
if isinstance(content, ToolResultContent):
# Extract text parts from the content list
result_texts: list[str] = []
if content.content:
for item in content.content:
if isinstance(item, TextContent):
result_texts.append(item.text)
openai_messages.append(
ChatCompletionToolMessageParam(
role="tool",
tool_call_id=content.toolUseId,
content="\n".join(result_texts),
)
)
continue
# Handle TextContent
if isinstance(content, TextContent):
if message.role == "user":
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=content.text,
)
)
else:
openai_messages.append(
ChatCompletionAssistantMessageParam(
role="assistant",
content=content.text,
)
)
continue
raise ValueError(f"Unsupported content type: {type(content)}")
return openai_messages
@ -168,3 +314,104 @@ class OpenAISamplingHandler(BaseLLMSamplingHandler):
return chosen_model
return self.default_model
@staticmethod
def _convert_tools_to_openai(tools: list[Tool]) -> list[ChatCompletionToolParam]:
"""Convert MCP tools to OpenAI tool format."""
openai_tools: list[ChatCompletionToolParam] = []
for tool in tools:
# Build parameters dict, ensuring required fields
parameters: dict[str, Any] = dict(tool.inputSchema)
if "type" not in parameters:
parameters["type"] = "object"
openai_tools.append(
ChatCompletionToolParam(
type="function",
function=FunctionDefinition(
name=tool.name,
description=tool.description or "",
parameters=parameters,
),
)
)
return openai_tools
@staticmethod
def _convert_tool_choice_to_openai(
tool_choice: ToolChoice,
) -> ChatCompletionToolChoiceOptionParam:
"""Convert MCP tool_choice to OpenAI format."""
if tool_choice.mode == "auto":
return "auto"
elif tool_choice.mode == "required":
return "required"
elif tool_choice.mode == "none":
return "none"
else:
raise ValueError(f"Unsupported tool_choice mode: {tool_choice.mode!r}")
@staticmethod
def _chat_completion_to_result_with_tools(
chat_completion: ChatCompletion,
) -> CreateMessageResultWithTools:
"""Convert OpenAI response to CreateMessageResultWithTools."""
if len(chat_completion.choices) == 0:
raise ValueError("No response for completion")
first_choice = chat_completion.choices[0]
message = first_choice.message
# Determine stop reason
stop_reason: StopReason
if first_choice.finish_reason == "tool_calls":
stop_reason = "toolUse"
elif first_choice.finish_reason == "stop":
stop_reason = "endTurn"
elif first_choice.finish_reason == "length":
stop_reason = "maxTokens"
else:
stop_reason = "endTurn"
# Build content list
content: list[TextContent | ToolUseContent] = []
# Add text content if present
if message.content:
content.append(TextContent(type="text", text=message.content))
# Add tool calls if present
if message.tool_calls:
for tool_call in message.tool_calls:
# Skip non-function tool calls
if not hasattr(tool_call, "function"):
continue
func = tool_call.function # type: ignore[union-attr]
# Parse the arguments JSON string
try:
arguments = json.loads(func.arguments) # type: ignore[union-attr]
except json.JSONDecodeError as e:
raise ValueError(
f"Invalid JSON in tool arguments for "
f"'{func.name}': {func.arguments}" # type: ignore[union-attr]
) from e
content.append(
ToolUseContent(
type="tool_use",
id=tool_call.id,
name=func.name, # type: ignore[union-attr]
input=arguments,
)
)
# Must have at least some content
if not content:
raise ValueError("No content in response from completion")
return CreateMessageResultWithTools(
content=content, # type: ignore[arg-type]
role="assistant",
model=chat_completion.model,
stopReason=stop_reason,
)

10
src/fastmcp/prompts/prompt.py
View file

@ -9,8 +9,8 @@ from typing import Annotated, Any
import pydantic_core
from mcp.types import ContentBlock, Icon, PromptMessage, Role, TextContent
from mcp.types import Prompt as MCPPrompt
from mcp.types import PromptArgument as MCPPromptArgument
from mcp.types import Prompt as SDKPrompt
from mcp.types import PromptArgument as SDKPromptArgument
from pydantic import Field, TypeAdapter
from fastmcp.exceptions import PromptError
@ -89,10 +89,10 @@ class Prompt(FastMCPComponent):
*,
include_fastmcp_meta: bool | None = None,
**overrides: Any,
) -> MCPPrompt:
) -> SDKPrompt:
"""Convert the prompt to an MCP prompt."""
arguments = [
MCPPromptArgument(
SDKPromptArgument(
name=arg.name,
description=arg.description,
required=arg.required,
@ -100,7 +100,7 @@ class Prompt(FastMCPComponent):
for arg in self.arguments or []
]
return MCPPrompt(
return SDKPrompt(
name=overrides.get("name", self.name),
description=overrides.get("description", self.description),
arguments=arguments,

8
src/fastmcp/resources/resource.py
View file

@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Annotated, Any
import pydantic_core
from mcp.types import Annotations, Icon
from mcp.types import Resource as MCPResource
from mcp.types import Resource as SDKResource
from pydantic import (
AnyUrl,
ConfigDict,
@ -126,10 +126,10 @@ class Resource(FastMCPComponent):
*,
include_fastmcp_meta: bool | None = None,
**overrides: Any,
) -> MCPResource:
"""Convert the resource to an MCPResource."""
) -> SDKResource:
"""Convert the resource to an SDKResource."""
return MCPResource(
return SDKResource(
name=overrides.get("name", self.name),
uri=overrides.get("uri", self.uri),
description=overrides.get("description", self.description),

10
src/fastmcp/resources/template.py
View file

@ -9,7 +9,7 @@ from typing import Annotated, Any
from urllib.parse import parse_qs, unquote
from mcp.types import Annotations, Icon
from mcp.types import ResourceTemplate as MCPResourceTemplate
from mcp.types import ResourceTemplate as SDKResourceTemplate
from pydantic import (
Field,
field_validator,
@ -188,10 +188,10 @@ class ResourceTemplate(FastMCPComponent):
*,
include_fastmcp_meta: bool | None = None,
**overrides: Any,
) -> MCPResourceTemplate:
"""Convert the resource template to an MCPResourceTemplate."""
) -> SDKResourceTemplate:
"""Convert the resource template to an SDKResourceTemplate."""
return MCPResourceTemplate(
return SDKResourceTemplate(
name=overrides.get("name", self.name),
uriTemplate=overrides.get("uriTemplate", self.uri_template),
description=overrides.get("description", self.description),
@ -205,7 +205,7 @@ class ResourceTemplate(FastMCPComponent):
)
@classmethod
def from_mcp_template(cls, mcp_template: MCPResourceTemplate) -> ResourceTemplate:
def from_mcp_template(cls, mcp_template: SDKResourceTemplate) -> ResourceTemplate:
"""Creates a FastMCP ResourceTemplate from a raw MCP ResourceTemplate object."""
# Note: This creates a simple ResourceTemplate instance. For function-based templates,
# the original function is lost, which is expected for remote templates.

562
src/fastmcp/server/context.py
View file

@ -1,15 +1,15 @@
from __future__ import annotations
import copy
import inspect
import json
import logging
import weakref
from collections.abc import Generator, Mapping, Sequence
from collections.abc import Callable, Generator, Mapping, Sequence
from contextlib import contextmanager
from contextvars import ContextVar, Token
from dataclasses import dataclass
from logging import Logger
from typing import Any, overload
from typing import Any, Literal, cast, overload
import anyio
from mcp import LoggingLevel, ServerSession
@ -17,25 +17,27 @@ from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import request_ctx
from mcp.shared.context import RequestContext
from mcp.types import (
ClientCapabilities,
CreateMessageResult,
CreateMessageResultWithTools,
GetPromptResult,
IncludeContext,
ModelHint,
ModelPreferences,
Root,
SamplingCapability,
SamplingMessage,
SamplingMessageContentBlock,
TextContent,
ToolChoice,
ToolResultContent,
ToolUseContent,
)
from mcp.types import CreateMessageRequestParams as SamplingParams
from mcp.types import Prompt as MCPPrompt
from mcp.types import Resource as MCPResource
from mcp.types import Prompt as SDKPrompt
from mcp.types import Resource as SDKResource
from mcp.types import Tool as SDKTool
from pydantic import ValidationError
from pydantic.networks import AnyUrl
from starlette.requests import Request
from typing_extensions import TypeVar
from fastmcp import settings
from fastmcp.server.elicitation import (
AcceptedElicitation,
CancelledElicitation,
@ -43,8 +45,19 @@ from fastmcp.server.elicitation import (
handle_elicit_accept,
parse_elicit_response_type,
)
from fastmcp.server.sampling import SampleStep, SamplingResult, SamplingTool
from fastmcp.server.sampling.run import (
_parse_model_preferences,
call_sampling_handler,
determine_handler_mode,
)
from fastmcp.server.sampling.run import (
execute_tools as run_sampling_tools,
)
from fastmcp.server.server import FastMCP
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import _clamp_logger, get_logger
from fastmcp.utilities.types import get_cached_typeadapter
logger: Logger = get_logger(name=__name__)
to_client_logger: Logger = logger.getChild(suffix="to_client")
@ -56,8 +69,14 @@ _clamp_logger(logger=to_client_logger, max_level="DEBUG")
T = TypeVar("T", default=Any)
ResultT = TypeVar("ResultT", default=str)
# Simplified tool choice type - just the mode string instead of the full MCP object
ToolChoiceOption = Literal["auto", "required", "none"]
_current_context: ContextVar[Context | None] = ContextVar("context", default=None) # type: ignore[assignment]
_flush_lock = anyio.Lock()
@ -245,7 +264,7 @@ class Context:
related_request_id=self.request_id,
)
async def list_resources(self) -> list[MCPResource]:
async def list_resources(self) -> list[SDKResource]:
"""List all available resources from the server.
Returns:
@ -253,7 +272,7 @@ class Context:
"""
return await self.fastmcp._list_resources_mcp()
async def list_prompts(self) -> list[MCPPrompt]:
async def list_prompts(self) -> list[SDKPrompt]:
"""List all available prompts from the server.
Returns:
@ -523,88 +542,344 @@ class Context:
return
await self.request_context.close_sse_stream()
async def sample(
async def sample_step(
self,
messages: str | Sequence[str | SamplingMessage],
*,
system_prompt: str | None = None,
include_context: IncludeContext | None = None,
temperature: float | None = None,
max_tokens: int | None = None,
model_preferences: ModelPreferences | str | list[str] | None = None,
) -> SamplingMessageContentBlock | list[SamplingMessageContentBlock]:
tools: Sequence[SamplingTool | Callable[..., Any]] | None = None,
tool_choice: ToolChoiceOption | str | None = None,
execute_tools: bool = True,
mask_error_details: bool | None = None,
) -> SampleStep:
"""
Make a single LLM sampling call.
This is a stateless function that makes exactly one LLM call and optionally
executes any requested tools. Use this for fine-grained control over the
sampling loop.
Args:
messages: The message(s) to send. Can be a string, list of strings,
or list of SamplingMessage objects.
system_prompt: Optional system prompt for the LLM.
temperature: Optional sampling temperature.
max_tokens: Maximum tokens to generate. Defaults to 512.
model_preferences: Optional model preferences.
tools: Optional list of tools the LLM can use.
tool_choice: Tool choice mode ("auto", "required", or "none").
execute_tools: If True (default), execute tool calls and append results
to history. If False, return immediately with tool_calls available
in the step for manual execution.
mask_error_details: If True, mask detailed error messages from tool
execution. When None (default), uses the global settings value.
Tools can raise ToolError to bypass masking.
Returns:
SampleStep containing:
- .response: The raw LLM response
- .history: Messages including input, assistant response, and tool results
- .is_tool_use: True if the LLM requested tool execution
- .tool_calls: List of tool calls (if any)
- .text: The text content (if any)
Example:
messages = "Research X"
while True:
step = await ctx.sample_step(messages, tools=[search])
if not step.is_tool_use:
print(step.text)
break
# Continue with tool results
messages = step.history
"""
# Convert messages to SamplingMessage objects
current_messages = _prepare_messages(messages)
# Convert tools to SamplingTools
sampling_tools = _prepare_tools(tools)
sdk_tools: list[SDKTool] | None = (
[t._to_sdk_tool() for t in sampling_tools] if sampling_tools else None
)
tool_map: dict[str, SamplingTool] = (
{t.name: t for t in sampling_tools} if sampling_tools else {}
)
# Determine whether to use fallback handler or client
use_fallback = determine_handler_mode(self, bool(sampling_tools))
# Build tool choice
effective_tool_choice: ToolChoice | None = None
if tool_choice is not None:
if tool_choice not in ("auto", "required", "none"):
raise ValueError(
f"Invalid tool_choice: {tool_choice!r}. "
"Must be 'auto', 'required', or 'none'."
)
effective_tool_choice = ToolChoice(
mode=cast(Literal["auto", "required", "none"], tool_choice)
)
# Effective max_tokens
effective_max_tokens = max_tokens if max_tokens is not None else 512
# Make the LLM call
if use_fallback:
response = await call_sampling_handler(
self,
current_messages,
system_prompt=system_prompt,
temperature=temperature,
max_tokens=effective_max_tokens,
model_preferences=model_preferences,
sdk_tools=sdk_tools,
tool_choice=effective_tool_choice,
)
else:
response = await self.session.create_message(
messages=current_messages,
system_prompt=system_prompt,
temperature=temperature,
max_tokens=effective_max_tokens,
model_preferences=_parse_model_preferences(model_preferences),
tools=sdk_tools,
tool_choice=effective_tool_choice,
related_request_id=self.request_id,
)
# Check if this is a tool use response
is_tool_use_response = (
isinstance(response, CreateMessageResultWithTools)
and response.stopReason == "toolUse"
)
# Always include the assistant response in history
current_messages.append(
SamplingMessage(role="assistant", content=response.content)
)
# If not a tool use, return immediately
if not is_tool_use_response:
return SampleStep(response=response, history=current_messages)
# If not executing tools, return with assistant message but no tool results
if not execute_tools:
return SampleStep(response=response, history=current_messages)
# Execute tools and add results to history
step_tool_calls = _extract_tool_calls(response)
if step_tool_calls:
effective_mask = (
mask_error_details
if mask_error_details is not None
else settings.mask_error_details
)
tool_results = await run_sampling_tools(
step_tool_calls, tool_map, mask_error_details=effective_mask
)
if tool_results:
current_messages.append(
SamplingMessage(
role="user",
content=tool_results, # type: ignore[arg-type]
)
)
return SampleStep(response=response, history=current_messages)
@overload
async def sample(
self,
messages: str | Sequence[str | SamplingMessage],
*,
system_prompt: str | None = None,
temperature: float | None = None,
max_tokens: int | None = None,
model_preferences: ModelPreferences | str | list[str] | None = None,
tools: Sequence[SamplingTool | Callable[..., Any]] | None = None,
result_type: type[ResultT],
mask_error_details: bool | None = None,
) -> SamplingResult[ResultT]:
"""Overload: With result_type, returns SamplingResult[ResultT]."""
@overload
async def sample(
self,
messages: str | Sequence[str | SamplingMessage],
*,
system_prompt: str | None = None,
temperature: float | None = None,
max_tokens: int | None = None,
model_preferences: ModelPreferences | str | list[str] | None = None,
tools: Sequence[SamplingTool | Callable[..., Any]] | None = None,
result_type: None = None,
mask_error_details: bool | None = None,
) -> SamplingResult[str]:
"""Overload: Without result_type, returns SamplingResult[str]."""
async def sample(
self,
messages: str | Sequence[str | SamplingMessage],
*,
system_prompt: str | None = None,
temperature: float | None = None,
max_tokens: int | None = None,
model_preferences: ModelPreferences | str | list[str] | None = None,
tools: Sequence[SamplingTool | Callable[..., Any]] | None = None,
result_type: type[ResultT] | None = None,
mask_error_details: bool | None = None,
) -> SamplingResult[ResultT] | SamplingResult[str]:
"""
Send a sampling request to the client and await the response.
Call this method at any time to have the server request an LLM
completion from the client. The client must be appropriately configured,
or the request will error.
This method runs to completion automatically. When tools are provided,
it executes a tool loop: if the LLM returns a tool use request, the tools
are executed and the results are sent back to the LLM. This continues
until the LLM provides a final text response.
When result_type is specified, a synthetic `final_response` tool is
created. The LLM calls this tool to provide the structured response,
which is validated against the result_type and returned as `.result`.
For fine-grained control over the sampling loop, use sample_step() instead.
Args:
messages: The message(s) to send. Can be a string, list of strings,
or list of SamplingMessage objects.
system_prompt: Optional system prompt for the LLM.
temperature: Optional sampling temperature.
max_tokens: Maximum tokens to generate. Defaults to 512.
model_preferences: Optional model preferences.
tools: Optional list of tools the LLM can use. Accepts plain
functions or SamplingTools.
result_type: Optional type for structured output. When specified,
a synthetic `final_response` tool is created and the LLM's
response is validated against this type.
mask_error_details: If True, mask detailed error messages from tool
execution. When None (default), uses the global settings value.
Tools can raise ToolError to bypass masking.
Returns:
SamplingResult[T] containing:
- .text: The text representation (raw text or JSON for structured)
- .result: The typed result (str for text, parsed object for structured)
- .history: All messages exchanged during sampling
"""
# Safety limit to prevent infinite loops
max_iterations = 100
if max_tokens is None:
max_tokens = 512
# Convert tools to SamplingTools
sampling_tools = _prepare_tools(tools)
has_user_tools = bool(sampling_tools)
if isinstance(messages, str):
sampling_messages = [
SamplingMessage(
content=TextContent(text=messages, type="text"), role="user"
)
]
elif isinstance(messages, Sequence):
sampling_messages = [
SamplingMessage(content=TextContent(text=m, type="text"), role="user")
if isinstance(m, str)
else m
for m in messages
]
# Handle structured output with result_type
tool_choice: str | None = None
if result_type is not None and result_type is not str:
final_response_tool = _create_final_response_tool(result_type)
sampling_tools = list(sampling_tools) if sampling_tools else []
sampling_tools.append(final_response_tool)
should_fallback = (
self.fastmcp.sampling_handler_behavior == "fallback"
and not self.session.check_client_capability(
capability=ClientCapabilities(sampling=SamplingCapability())
)
)
# If no user tools, force the LLM to call a tool (which is final_response)
if not has_user_tools:
tool_choice = "required"
if self.fastmcp.sampling_handler_behavior == "always" or should_fallback:
if self.fastmcp.sampling_handler is None:
raise ValueError("Client does not support sampling")
# Convert messages for the loop
current_messages: str | Sequence[str | SamplingMessage] = messages
create_message_result = self.fastmcp.sampling_handler(
sampling_messages,
SamplingParams(
systemPrompt=system_prompt,
messages=sampling_messages,
temperature=temperature,
maxTokens=max_tokens,
modelPreferences=_parse_model_preferences(model_preferences),
),
self.request_context, # type: ignore[arg-type]
for _iteration in range(max_iterations):
step = await self.sample_step(
messages=current_messages,
system_prompt=system_prompt,
temperature=temperature,
max_tokens=max_tokens,
model_preferences=model_preferences,
tools=sampling_tools,
tool_choice=tool_choice,
mask_error_details=mask_error_details,
)
if inspect.isawaitable(create_message_result):
create_message_result = await create_message_result
# Check for final_response tool call for structured output
if result_type is not None and result_type is not str and step.is_tool_use:
for tool_call in step.tool_calls:
if tool_call.name == "final_response":
# Validate and return the structured result
type_adapter = get_cached_typeadapter(result_type)
if isinstance(create_message_result, str):
return TextContent(text=create_message_result, type="text")
# Unwrap if we wrapped primitives (non-object schemas)
input_data = tool_call.input
original_schema = compress_schema(
type_adapter.json_schema(), prune_titles=True
)
if (
original_schema.get("type") != "object"
and isinstance(input_data, dict)
and "value" in input_data
):
input_data = input_data["value"]
if isinstance(create_message_result, CreateMessageResult):
return create_message_result.content
try:
validated_result = type_adapter.validate_python(input_data)
text = json.dumps(
type_adapter.dump_python(validated_result, mode="json")
)
return SamplingResult(
text=text,
result=validated_result,
history=step.history,
)
except ValidationError as e:
# Validation failed - add error as tool result
step.history.append(
SamplingMessage(
role="user",
content=[
ToolResultContent(
type="tool_result",
toolUseId=tool_call.id,
content=[
TextContent(
type="text",
text=(
f"Validation error: {e}. "
"Please try again with valid data."
),
)
],
isError=True,
)
], # type: ignore[arg-type]
)
)
else:
raise ValueError(
f"Unexpected sampling handler result: {create_message_result}"
# If not a tool use response, we're done
if not step.is_tool_use:
# For structured output, the LLM must use the final_response tool
if result_type is not None and result_type is not str:
raise RuntimeError(
f"Expected structured output of type {result_type.__name__}, "
"but the LLM returned a text response instead of calling "
"the final_response tool."
)
return SamplingResult(
text=step.text,
result=cast(ResultT, step.text if step.text else ""),
history=step.history,
)
result: CreateMessageResult = await self.session.create_message(
messages=sampling_messages,
system_prompt=system_prompt,
include_context=include_context,
temperature=temperature,
max_tokens=max_tokens,
model_preferences=_parse_model_preferences(model_preferences),
related_request_id=self.request_id,
)
# Continue with the updated history
current_messages = step.history
return result.content
# After first iteration, reset tool_choice to auto
tool_choice = None
raise RuntimeError(f"Sampling exceeded maximum iterations ({max_iterations})")
@overload
async def elicit(
@ -641,11 +916,12 @@ class Context:
async def elicit(
self,
message: str,
response_type: type[T] | list[str] | None = None,
response_type: type[T] | list[str] | dict[str, dict[str, str]] | None = None,
) -> (
AcceptedElicitation[T]
| AcceptedElicitation[dict[str, Any]]
| AcceptedElicitation[str]
| AcceptedElicitation[list[str]]
| DeclinedElicitation
| CancelledElicitation
):
@ -728,47 +1004,6 @@ class Context:
pass
def _parse_model_preferences(
model_preferences: ModelPreferences | str | list[str] | None,
) -> ModelPreferences | None:
"""
Validates and converts user input for model_preferences into a ModelPreferences object.
Args:
model_preferences (ModelPreferences | str | list[str] | None):
The model preferences to use. Accepts:
- ModelPreferences (returns as-is)
- str (single model hint)
- list[str] (multiple model hints)
- None (no preferences)
Returns:
ModelPreferences | None: The parsed ModelPreferences object, or None if not provided.
Raises:
ValueError: If the input is not a supported type or contains invalid values.
"""
if model_preferences is None:
return None
elif isinstance(model_preferences, ModelPreferences):
return model_preferences
elif isinstance(model_preferences, str):
# Single model hint
return ModelPreferences(hints=[ModelHint(name=model_preferences)])
elif isinstance(model_preferences, list):
# List of model hints (strings)
if not all(isinstance(h, str) for h in model_preferences):
raise ValueError(
"All elements of model_preferences list must be"
" strings (model name hints)."
)
return ModelPreferences(hints=[ModelHint(name=h) for h in model_preferences])
else:
raise ValueError(
"model_preferences must be one of: ModelPreferences, str, list[str], or None."
)
async def _log_to_server_and_client(
data: LogData,
session: ServerSession,
@ -795,3 +1030,104 @@ async def _log_to_server_and_client(
logger=logger_name,
related_request_id=related_request_id,
)
def _create_final_response_tool(result_type: type) -> SamplingTool:
"""Create a synthetic 'final_response' tool for structured output.
This tool is used to capture structured responses from the LLM.
The tool's schema is derived from the result_type.
"""
type_adapter = get_cached_typeadapter(result_type)
schema = type_adapter.json_schema()
schema = compress_schema(schema, prune_titles=True)
# Tool parameters must be object-shaped. Wrap primitives in {"value": <schema>}
if schema.get("type") != "object":
schema = {
"type": "object",
"properties": {"value": schema},
"required": ["value"],
}
# The fn just returns the input as-is (validation happens in the loop)
def final_response(**kwargs: Any) -> dict[str, Any]:
return kwargs
return SamplingTool(
name="final_response",
description=(
"Call this tool to provide your final response. "
"Use this when you have completed the task and are ready to return the result."
),
parameters=schema,
fn=final_response,
)
def _extract_text_from_content(
content: SamplingMessageContentBlock | list[SamplingMessageContentBlock],
) -> str | None:
"""Extract text from content block(s).
Returns the text if content is a TextContent or list containing TextContent,
otherwise returns None.
"""
if isinstance(content, list):
for block in content:
if isinstance(block, TextContent):
return block.text
return None
elif isinstance(content, TextContent):
return content.text
return None
def _prepare_messages(
messages: str | Sequence[str | SamplingMessage],
) -> list[SamplingMessage]:
"""Convert various message formats to a list of SamplingMessage objects."""
if isinstance(messages, str):
return [
SamplingMessage(
content=TextContent(text=messages, type="text"), role="user"
)
]
else:
return [
SamplingMessage(content=TextContent(text=m, type="text"), role="user")
if isinstance(m, str)
else m
for m in messages
]
def _prepare_tools(
tools: Sequence[SamplingTool | Callable[..., Any]] | None,
) -> list[SamplingTool] | None:
"""Convert tools to SamplingTool objects."""
if tools is None:
return None
sampling_tools: list[SamplingTool] = []
for t in tools:
if isinstance(t, SamplingTool):
sampling_tools.append(t)
elif callable(t):
sampling_tools.append(SamplingTool.from_function(t))
else:
raise TypeError(f"Expected SamplingTool or callable, got {type(t)}")
return sampling_tools if sampling_tools else None
def _extract_tool_calls(
response: CreateMessageResult | CreateMessageResultWithTools,
) -> list[ToolUseContent]:
"""Extract tool calls from a response."""
content = response.content
if isinstance(content, list):
return [c for c in content if isinstance(c, ToolUseContent)]
elif isinstance(content, ToolUseContent):
return [content]
return []

6
src/fastmcp/server/proxy.py
View file

@ -589,15 +589,15 @@ class ProxyClient(Client[ClientTransportT]):
A handler that forwards the sampling request from the remote server to the proxy's connected clients and relays the response back to the remote server.
"""
ctx = get_context()
content = await ctx.sample(
result = await ctx.sample(
list(messages),
system_prompt=params.systemPrompt,
temperature=params.temperature,
max_tokens=params.maxTokens,
model_preferences=params.modelPreferences,
)
if isinstance(content, mcp.types.ResourceLink | mcp.types.EmbeddedResource):
raise RuntimeError("Content is not supported")
# Create TextContent from the result text
content = mcp.types.TextContent(type="text", text=result.text or "")
return mcp.types.CreateMessageResult(
role="assistant",
model="fastmcp-client",

12
src/fastmcp/server/sampling/__init__.py Normal file
View file

@ -0,0 +1,12 @@
"""Sampling module for FastMCP servers."""
from fastmcp.server.sampling.handler import ServerSamplingHandler
from fastmcp.server.sampling.run import SampleStep, SamplingResult
from fastmcp.server.sampling.sampling_tool import SamplingTool
__all__ = [
"SampleStep",
"SamplingResult",
"SamplingTool",
"ServerSamplingHandler",
]

9
src/fastmcp/server/sampling/handler.py
View file

@ -5,8 +5,11 @@ from mcp import CreateMessageResult
from mcp.server.session import ServerSession
from mcp.shared.context import LifespanContextT, RequestContext
from mcp.types import CreateMessageRequestParams as SamplingParams
from mcp.types import (
SamplingMessage,
from mcp.types import CreateMessageResultWithTools, SamplingMessage
# Result type that handlers can return
SamplingHandlerResult: TypeAlias = (
str | CreateMessageResult | CreateMessageResultWithTools
)
ServerSamplingHandler: TypeAlias = Callable[
@ -15,5 +18,5 @@ ServerSamplingHandler: TypeAlias = Callable[
SamplingParams,
RequestContext[ServerSession, LifespanContextT],
],
str | CreateMessageResult | Awaitable[str | CreateMessageResult],
SamplingHandlerResult | Awaitable[SamplingHandlerResult],
]

301
src/fastmcp/server/sampling/run.py Normal file
View file

@ -0,0 +1,301 @@
"""Sampling types and helper functions for FastMCP servers."""
from __future__ import annotations
import inspect
from dataclasses import dataclass
from typing import TYPE_CHECKING, Generic
from mcp.types import (
ClientCapabilities,
CreateMessageResult,
CreateMessageResultWithTools,
ModelHint,
ModelPreferences,
SamplingCapability,
SamplingMessage,
SamplingToolsCapability,
TextContent,
ToolChoice,
ToolResultContent,
ToolUseContent,
)
from mcp.types import CreateMessageRequestParams as SamplingParams
from mcp.types import Tool as SDKTool
from typing_extensions import TypeVar
from fastmcp.exceptions import ToolError
from fastmcp.server.sampling.sampling_tool import SamplingTool
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
if TYPE_CHECKING:
from fastmcp.server.context import Context
ResultT = TypeVar("ResultT", default=str)
@dataclass
class SamplingResult(Generic[ResultT]):
"""Result of a sampling operation.
Attributes:
text: The text representation of the result (raw text or JSON for structured).
result: The typed result (str for text, parsed object for structured output).
history: All messages exchanged during sampling.
"""
text: str | None
result: ResultT
history: list[SamplingMessage]
@dataclass
class SampleStep:
"""Result of a single sampling call.
Represents what the LLM returned in this step plus the message history.
"""
response: CreateMessageResult | CreateMessageResultWithTools
history: list[SamplingMessage]
@property
def is_tool_use(self) -> bool:
"""True if the LLM is requesting tool execution."""
if isinstance(self.response, CreateMessageResultWithTools):
return self.response.stopReason == "toolUse"
return False
@property
def text(self) -> str | None:
"""Extract text from the response, if available."""
content = self.response.content
if isinstance(content, list):
for block in content:
if isinstance(block, TextContent):
return block.text
return None
elif isinstance(content, TextContent):
return content.text
return None
@property
def tool_calls(self) -> list[ToolUseContent]:
"""Get the list of tool calls from the response."""
content = self.response.content
if isinstance(content, list):
return [c for c in content if isinstance(c, ToolUseContent)]
elif isinstance(content, ToolUseContent):
return [content]
return []
def _parse_model_preferences(
model_preferences: ModelPreferences | str | list[str] | None,
) -> ModelPreferences | None:
"""Convert model preferences to ModelPreferences object."""
if model_preferences is None:
return None
elif isinstance(model_preferences, ModelPreferences):
return model_preferences
elif isinstance(model_preferences, str):
return ModelPreferences(hints=[ModelHint(name=model_preferences)])
elif isinstance(model_preferences, list):
if not all(isinstance(h, str) for h in model_preferences):
raise ValueError("All elements of model_preferences list must be strings.")
return ModelPreferences(hints=[ModelHint(name=h) for h in model_preferences])
else:
raise ValueError(
"model_preferences must be one of: ModelPreferences, str, list[str], or None."
)
# --- Standalone functions for sample_step() ---
def determine_handler_mode(context: Context, needs_tools: bool) -> bool:
"""Determine whether to use fallback handler or client for sampling.
Args:
context: The MCP context.
needs_tools: Whether the sampling request requires tool support.
Returns:
True if fallback handler should be used, False to use client.
Raises:
ValueError: If client lacks required capability and no fallback configured.
"""
fastmcp = context.fastmcp
session = context.session
# Check what capabilities the client has
has_sampling = session.check_client_capability(
capability=ClientCapabilities(sampling=SamplingCapability())
)
has_tools_capability = session.check_client_capability(
capability=ClientCapabilities(
sampling=SamplingCapability(tools=SamplingToolsCapability())
)
)
if fastmcp.sampling_handler_behavior == "always":
if fastmcp.sampling_handler is None:
raise ValueError(
"sampling_handler_behavior is 'always' but no handler configured"
)
return True
elif fastmcp.sampling_handler_behavior == "fallback":
client_sufficient = has_sampling and (not needs_tools or has_tools_capability)
if not client_sufficient:
if fastmcp.sampling_handler is None:
if needs_tools and has_sampling and not has_tools_capability:
raise ValueError(
"Client does not support sampling with tools. "
"The client must advertise the sampling.tools capability."
)
raise ValueError("Client does not support sampling")
return True
elif fastmcp.sampling_handler_behavior is not None:
raise ValueError(
f"Invalid sampling_handler_behavior: {fastmcp.sampling_handler_behavior!r}. "
"Must be 'always', 'fallback', or None."
)
elif not has_sampling:
raise ValueError("Client does not support sampling")
elif needs_tools and not has_tools_capability:
raise ValueError(
"Client does not support sampling with tools. "
"The client must advertise the sampling.tools capability."
)
return False
async def call_sampling_handler(
context: Context,
messages: list[SamplingMessage],
*,
system_prompt: str | None,
temperature: float | None,
max_tokens: int,
model_preferences: ModelPreferences | str | list[str] | None,
sdk_tools: list[SDKTool] | None,
tool_choice: ToolChoice | None,
) -> CreateMessageResult | CreateMessageResultWithTools:
"""Make LLM call using the fallback handler.
Note: This function expects the caller (sample_step) to have validated that
sampling_handler is set via determine_handler_mode(). The checks below are
safeguards against internal misuse.
"""
if context.fastmcp.sampling_handler is None:
raise RuntimeError("sampling_handler is None")
if context.request_context is None:
raise RuntimeError("request_context is None")
result = context.fastmcp.sampling_handler(
messages,
SamplingParams(
systemPrompt=system_prompt,
messages=messages,
temperature=temperature,
maxTokens=max_tokens,
modelPreferences=_parse_model_preferences(model_preferences),
tools=sdk_tools,
toolChoice=tool_choice,
),
context.request_context,
)
if inspect.isawaitable(result):
result = await result
# Convert string to CreateMessageResult
if isinstance(result, str):
return CreateMessageResult(
role="assistant",
content=TextContent(type="text", text=result),
model="unknown",
stopReason="endTurn",
)
return result
async def execute_tools(
tool_calls: list[ToolUseContent],
tool_map: dict[str, SamplingTool],
mask_error_details: bool = False,
) -> list[ToolResultContent]:
"""Execute tool calls and return results.
Args:
tool_calls: List of tool use requests from the LLM.
tool_map: Mapping from tool name to SamplingTool.
mask_error_details: If True, mask detailed error messages from tool execution.
When masked, only generic error messages are returned to the LLM.
Tools can explicitly raise ToolError to bypass masking when they want
to provide specific error messages to the LLM.
Returns:
List of tool result content blocks.
"""
tool_results: list[ToolResultContent] = []
for tool_use in tool_calls:
tool = tool_map.get(tool_use.name)
if tool is None:
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=tool_use.id,
content=[
TextContent(
type="text",
text=f"Error: Unknown tool '{tool_use.name}'",
)
],
isError=True,
)
)
else:
try:
result_value = await tool.run(tool_use.input)
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=tool_use.id,
content=[TextContent(type="text", text=str(result_value))],
)
)
except ToolError as e:
# ToolError is the escape hatch - always pass message through
logger.exception(f"Error calling sampling tool '{tool_use.name}'")
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=tool_use.id,
content=[TextContent(type="text", text=str(e))],
isError=True,
)
)
except Exception as e:
# Generic exceptions - mask based on setting
logger.exception(f"Error calling sampling tool '{tool_use.name}'")
if mask_error_details:
error_text = f"Error executing tool '{tool_use.name}'"
else:
error_text = f"Error executing tool '{tool_use.name}': {e}"
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=tool_use.id,
content=[TextContent(type="text", text=error_text)],
isError=True,
)
)
return tool_results

108
src/fastmcp/server/sampling/sampling_tool.py Normal file
View file

@ -0,0 +1,108 @@
"""SamplingTool for use during LLM sampling requests."""
from __future__ import annotations
import inspect
from collections.abc import Callable
from typing import Any
from mcp.types import Tool as SDKTool
from pydantic import BaseModel, ConfigDict
from fastmcp.tools.tool import ParsedFunction
class SamplingTool(BaseModel):
"""A tool that can be used during LLM sampling.
SamplingTools bundle a tool's schema (name, description, parameters) with
an executor function, enabling servers to execute agentic workflows where
the LLM can request tool calls during sampling.
In most cases, pass functions directly to ctx.sample():
def search(query: str) -> str:
'''Search the web.'''
return web_search(query)
result = await context.sample(
messages="Find info about Python",
tools=[search], # Plain functions work directly
)
Create a SamplingTool explicitly when you need custom name/description:
tool = SamplingTool.from_function(search, name="web_search")
"""
name: str
description: str | None = None
parameters: dict[str, Any]
fn: Callable[..., Any]
model_config = ConfigDict(arbitrary_types_allowed=True)
async def run(self, arguments: dict[str, Any] | None = None) -> Any:
"""Execute the tool with the given arguments.
Args:
arguments: Dictionary of arguments to pass to the tool function.
Returns:
The result of executing the tool function.
"""
if arguments is None:
arguments = {}
result = self.fn(**arguments)
if inspect.isawaitable(result):
result = await result
return result
def _to_sdk_tool(self) -> SDKTool:
"""Convert to an mcp.types.Tool for SDK compatibility.
This is used internally when passing tools to the MCP SDK's
create_message() method.
"""
return SDKTool(
name=self.name,
description=self.description,
inputSchema=self.parameters,
)
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
*,
name: str | None = None,
description: str | None = None,
) -> SamplingTool:
"""Create a SamplingTool from a function.
The function's signature is analyzed to generate a JSON schema for
the tool's parameters. Type hints are used to determine parameter types.
Args:
fn: The function to create a tool from.
name: Optional name override. Defaults to the function's name.
description: Optional description override. Defaults to the function's docstring.
Returns:
A SamplingTool wrapping the function.
Raises:
ValueError: If the function is a lambda without a name override.
"""
parsed = ParsedFunction.from_function(fn, validate=True)
if name is None and parsed.name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
return cls(
name=name or parsed.name,
description=description or parsed.description,
parameters=parsed.input_schema,
fn=parsed.fn,
)

16
src/fastmcp/server/server.py
View file

@ -46,10 +46,10 @@ from mcp.types import (
GetPromptResult,
ToolAnnotations,
)
from mcp.types import Prompt as MCPPrompt
from mcp.types import Resource as MCPResource
from mcp.types import ResourceTemplate as MCPResourceTemplate
from mcp.types import Tool as MCPTool
from mcp.types import Prompt as SDKPrompt
from mcp.types import Resource as SDKResource
from mcp.types import ResourceTemplate as SDKResourceTemplate
from mcp.types import Tool as SDKTool
from pydantic import AnyUrl
from starlette.middleware import Middleware as ASGIMiddleware
from starlette.requests import Request
@ -1153,7 +1153,7 @@ class FastMCP(Generic[LifespanResultT]):
return routes
async def _list_tools_mcp(self) -> list[MCPTool]:
async def _list_tools_mcp(self) -> list[SDKTool]:
"""
List all available tools, in the format expected by the low-level MCP
server.
@ -1237,7 +1237,7 @@ class FastMCP(Generic[LifespanResultT]):
return list(all_tools.values())
async def _list_resources_mcp(self) -> list[MCPResource]:
async def _list_resources_mcp(self) -> list[SDKResource]:
"""
List all available resources, in the format expected by the low-level MCP
server.
@ -1326,7 +1326,7 @@ class FastMCP(Generic[LifespanResultT]):
return list(all_resources.values())
async def _list_resource_templates_mcp(self) -> list[MCPResourceTemplate]:
async def _list_resource_templates_mcp(self) -> list[SDKResourceTemplate]:
"""
List all available resource templates, in the format expected by the low-level MCP
server.
@ -1420,7 +1420,7 @@ class FastMCP(Generic[LifespanResultT]):
return list(all_templates.values())
async def _list_prompts_mcp(self) -> list[MCPPrompt]:
async def _list_prompts_mcp(self) -> list[SDKPrompt]:
"""
List all available prompts, in the format expected by the low-level MCP
server.

823
tests/client/test_sampling.py
View file

@ -8,6 +8,7 @@ from pydantic_core import to_json
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
from fastmcp.server.sampling import SamplingTool
from fastmcp.utilities.types import Image
@ -163,3 +164,825 @@ async def test_sampling_with_image(fastmcp_server: FastMCP):
"_meta": None,
},
]
class TestSamplingWithTools:
"""Tests for sampling with tools functionality."""
async def test_sampling_with_tools_requires_capability(self):
"""Test that sampling with tools raises error when client lacks capability."""
import mcp.types as mcp_types
from fastmcp.exceptions import ToolError
server = FastMCP()
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
@server.tool
async def sample_with_tool(context: Context) -> str:
# This should fail because the client doesn't advertise tools capability
result = await context.sample(
messages="Search for Python tutorials",
tools=[search],
)
return str(result)
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> str:
return "Response"
# Explicitly disable tools capability by passing SamplingCapability without tools
async with Client(
server,
sampling_handler=sampling_handler,
sampling_capabilities=mcp_types.SamplingCapability(), # No tools
) as client:
with pytest.raises(ToolError, match="sampling.tools capability"):
await client.call_tool("sample_with_tool", {})
async def test_sampling_with_tools_fallback_handler_can_return_string(self):
"""Test that fallback handler can return a string even when tools are provided.
The LLM might choose not to use any tools and just return a text response.
"""
# This handler returns a string - valid even when tools are provided
simple_handler = AsyncMock(return_value="Direct response without tools")
mcp = FastMCP(sampling_handler=simple_handler)
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
@mcp.tool
async def sample_with_tool(context: Context) -> str:
result = await context.sample(
messages="Search for Python tutorials",
tools=[search],
)
return result.text or "no text"
# Client without sampling handler - will use server's fallback
async with Client(mcp) as client:
result = await client.call_tool("sample_with_tool", {})
# Handler returned string directly, which is treated as final text response
assert result.data == "Direct response without tools"
def test_sampling_tool_schema(self):
"""Test that SamplingTool generates correct schema."""
def search(query: str, limit: int = 10) -> str:
"""Search the web for results."""
return f"Results for: {query}"
tool = SamplingTool.from_function(search)
assert tool.name == "search"
assert tool.description == "Search the web for results."
assert "query" in tool.parameters.get("properties", {})
assert "limit" in tool.parameters.get("properties", {})
async def test_sampling_tool_run(self):
"""Test that SamplingTool.run() executes correctly."""
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
tool = SamplingTool.from_function(add)
result = await tool.run({"a": 5, "b": 3})
assert result == 8
async def test_sampling_tool_run_async(self):
"""Test that SamplingTool.run() works with async functions."""
async def async_multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
tool = SamplingTool.from_function(async_multiply)
result = await tool.run({"a": 4, "b": 7})
assert result == 28
def test_tool_choice_parameter(self):
"""Test that tool_choice parameter accepts string literals."""
from fastmcp.server.context import ToolChoiceOption
# Verify ToolChoiceOption type accepts the valid string values
choices: list[ToolChoiceOption] = ["auto", "required", "none"]
assert len(choices) == 3
assert "auto" in choices
assert "required" in choices
assert "none" in choices
class TestAutomaticToolLoop:
"""Tests for automatic tool execution loop in ctx.sample()."""
async def test_automatic_tool_loop_executes_tools(self):
"""Test that ctx.sample() automatically executes tool calls."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
call_count = 0
tool_was_called = False
def get_weather(city: str) -> str:
"""Get weather for a city."""
nonlocal tool_was_called
tool_was_called = True
return f"Weather in {city}: sunny, 72°F"
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
nonlocal call_count
call_count += 1
if call_count == 1:
# First call: return tool use
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="get_weather",
input={"city": "Seattle"},
)
],
model="test-model",
stopReason="toolUse",
)
else:
# Second call: return final response
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="The weather is sunny!")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def weather_assistant(question: str, context: Context) -> str:
result = await context.sample(
messages=question,
tools=[get_weather],
)
# Get text from SamplingResult
return result.text or ""
async with Client(mcp) as client:
result = await client.call_tool(
"weather_assistant", {"question": "What's the weather?"}
)
assert tool_was_called
assert call_count == 2
assert result.data == "The weather is sunny!"
async def test_automatic_tool_loop_multiple_tools(self):
"""Test that multiple tool calls in one response are all executed."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
executed_tools: list[str] = []
def tool_a(x: int) -> int:
"""Tool A."""
executed_tools.append(f"tool_a({x})")
return x * 2
def tool_b(y: int) -> int:
"""Tool B."""
executed_tools.append(f"tool_b({y})")
return y + 10
call_count = 0
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
nonlocal call_count
call_count += 1
if call_count == 1:
# Return multiple tool calls
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use", id="call_a", name="tool_a", input={"x": 5}
),
ToolUseContent(
type="tool_use", id="call_b", name="tool_b", input={"y": 3}
),
],
model="test-model",
stopReason="toolUse",
)
else:
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Done!")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def multi_tool(context: Context) -> str:
result = await context.sample(messages="Run tools", tools=[tool_a, tool_b])
return result.text or ""
async with Client(mcp) as client:
result = await client.call_tool("multi_tool", {})
assert executed_tools == ["tool_a(5)", "tool_b(3)"]
assert result.data == "Done!"
async def test_automatic_tool_loop_handles_unknown_tool(self):
"""Test that unknown tool names result in error being passed to LLM."""
from mcp.types import (
CreateMessageResultWithTools,
ToolResultContent,
ToolUseContent,
)
def known_tool() -> str:
"""A known tool."""
return "known result"
messages_received: list[list[SamplingMessage]] = []
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
messages_received.append(list(messages))
if len(messages_received) == 1:
# Request unknown tool
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="unknown_tool",
input={},
)
],
model="test-model",
stopReason="toolUse",
)
else:
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Handled error")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def test_unknown(context: Context) -> str:
result = await context.sample(messages="Test", tools=[known_tool])
return result.text or ""
async with Client(mcp) as client:
result = await client.call_tool("test_unknown", {})
# Check that error was passed back in messages
assert len(messages_received) == 2
last_messages = messages_received[1]
# Find the tool result in list content
tool_result = None
for msg in last_messages:
# Tool results are now in a list
if isinstance(msg.content, list):
for item in msg.content:
if isinstance(item, ToolResultContent):
tool_result = item
break
elif isinstance(msg.content, ToolResultContent):
tool_result = msg.content
break
assert tool_result is not None
assert tool_result.isError is True
# Content is list of TextContent objects
error_text = tool_result.content[0].text # type: ignore[union-attr]
assert "Unknown tool" in error_text
assert result.data == "Handled error"
async def test_automatic_tool_loop_handles_tool_exception(self):
"""Test that tool exceptions are caught and passed to LLM as errors."""
from mcp.types import (
CreateMessageResultWithTools,
ToolResultContent,
ToolUseContent,
)
def failing_tool() -> str:
"""A tool that raises an exception."""
raise ValueError("Tool failed intentionally")
messages_received: list[list[SamplingMessage]] = []
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
messages_received.append(list(messages))
if len(messages_received) == 1:
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="failing_tool",
input={},
)
],
model="test-model",
stopReason="toolUse",
)
else:
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Handled error")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def test_exception(context: Context) -> str:
result = await context.sample(messages="Test", tools=[failing_tool])
return result.text or ""
async with Client(mcp) as client:
result = await client.call_tool("test_exception", {})
# Check that error was passed back
assert len(messages_received) == 2
last_messages = messages_received[1]
# Find the tool result in list content
tool_result = None
for msg in last_messages:
# Tool results are now in a list
if isinstance(msg.content, list):
for item in msg.content:
if isinstance(item, ToolResultContent):
tool_result = item
break
elif isinstance(msg.content, ToolResultContent):
tool_result = msg.content
break
assert tool_result is not None
assert tool_result.isError is True
# Content is list of TextContent objects
error_text = tool_result.content[0].text # type: ignore[union-attr]
assert "Tool failed intentionally" in error_text
assert result.data == "Handled error"
class TestSamplingResultType:
"""Tests for result_type parameter (structured output)."""
async def test_result_type_creates_final_response_tool(self):
"""Test that result_type creates a synthetic final_response tool."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
from pydantic import BaseModel
class MathResult(BaseModel):
answer: int
explanation: str
received_tools: list = []
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
received_tools.extend(params.tools or [])
# Return the final_response tool call
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="final_response",
input={"answer": 42, "explanation": "The meaning of life"},
)
],
model="test-model",
stopReason="toolUse",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def math_tool(context: Context) -> str:
result = await context.sample(
messages="What is 6 * 7?",
result_type=MathResult,
)
# result.result should be a MathResult object
return f"{result.result.answer}: {result.result.explanation}" # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.call_tool("math_tool", {})
# Check that final_response tool was added
tool_names = [t.name for t in received_tools]
assert "final_response" in tool_names
# Check the result
assert result.data == "42: The meaning of life"
async def test_result_type_with_user_tools(self):
"""Test result_type works alongside user-provided tools."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
from pydantic import BaseModel
class SearchResult(BaseModel):
summary: str
sources: list[str]
def search(query: str) -> str:
"""Search for information."""
return f"Found info about: {query}"
call_count = 0
tool_was_called = False
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
nonlocal call_count, tool_was_called
call_count += 1
if call_count == 1:
# First call: use the search tool
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="search",
input={"query": "Python tutorials"},
)
],
model="test-model",
stopReason="toolUse",
)
else:
# Second call: call final_response
tool_was_called = True
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_2",
name="final_response",
input={
"summary": "Python is great",
"sources": ["python.org", "docs.python.org"],
},
)
],
model="test-model",
stopReason="toolUse",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def research(context: Context) -> str:
result = await context.sample(
messages="Research Python",
tools=[search],
result_type=SearchResult,
)
return f"{result.result.summary} - {len(result.result.sources)} sources" # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.call_tool("research", {})
assert tool_was_called
assert result.data == "Python is great - 2 sources"
async def test_result_type_validation_error_retries(self):
"""Test that validation errors are sent back to LLM for retry."""
from mcp.types import (
CreateMessageResultWithTools,
ToolResultContent,
ToolUseContent,
)
from pydantic import BaseModel
class StrictResult(BaseModel):
value: int # Must be an int
messages_received: list[list[SamplingMessage]] = []
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
messages_received.append(list(messages))
if len(messages_received) == 1:
# First call: invalid type
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="final_response",
input={"value": "not_an_int"}, # Wrong type
)
],
model="test-model",
stopReason="toolUse",
)
else:
# Second call: valid type after seeing error
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_2",
name="final_response",
input={"value": 42}, # Correct type
)
],
model="test-model",
stopReason="toolUse",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def validate_tool(context: Context) -> str:
result = await context.sample(
messages="Give me a number",
result_type=StrictResult,
)
return str(result.result.value) # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.call_tool("validate_tool", {})
# Should have retried after validation error
assert len(messages_received) == 2
# Check that error was passed back
last_messages = messages_received[1]
# Find the tool result in list content
tool_result = None
for msg in last_messages:
# Tool results are now in a list
if isinstance(msg.content, list):
for item in msg.content:
if isinstance(item, ToolResultContent):
tool_result = item
break
elif isinstance(msg.content, ToolResultContent):
tool_result = msg.content
break
assert tool_result is not None
assert tool_result.isError is True
error_text = tool_result.content[0].text # type: ignore[union-attr]
assert "Validation error" in error_text
# Final result should be correct
assert result.data == "42"
async def test_sampling_result_has_text_and_history(self):
"""Test that SamplingResult has text, result, and history attributes."""
from mcp.types import CreateMessageResultWithTools
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Hello world")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def check_result(context: Context) -> str:
result = await context.sample(messages="Say hello")
# Check all attributes exist
assert result.text == "Hello world"
assert result.result == "Hello world"
assert len(result.history) >= 1
return "ok"
async with Client(mcp) as client:
result = await client.call_tool("check_result", {})
assert result.data == "ok"
class TestSampleStep:
"""Tests for ctx.sample_step() - single LLM call with manual control."""
async def test_sample_step_basic(self):
"""Test basic sample_step returns text response."""
from mcp.types import CreateMessageResultWithTools
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Hello from step")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def test_step(context: Context) -> str:
step = await context.sample_step(messages="Hi")
assert not step.is_tool_use
assert step.text == "Hello from step"
return step.text or ""
async with Client(mcp) as client:
result = await client.call_tool("test_step", {})
assert result.data == "Hello from step"
async def test_sample_step_with_tool_execution(self):
"""Test sample_step executes tools by default."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
call_count = 0
def my_tool(x: int) -> str:
"""A test tool."""
return f"result:{x}"
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
nonlocal call_count
call_count += 1
if call_count == 1:
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="my_tool",
input={"x": 42},
)
],
model="test-model",
stopReason="toolUse",
)
else:
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Done")],
model="test-model",
stopReason="endTurn",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def test_step(context: Context) -> str:
messages: str | list[SamplingMessage] = "Run tool"
while True:
step = await context.sample_step(messages=messages, tools=[my_tool])
if not step.is_tool_use:
return step.text or ""
# History should include tool results when execute_tools=True
messages = step.history
async with Client(mcp) as client:
result = await client.call_tool("test_step", {})
assert result.data == "Done"
assert call_count == 2
async def test_sample_step_execute_tools_false(self):
"""Test sample_step with execute_tools=False doesn't execute tools."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
tool_executed = False
def my_tool() -> str:
"""A test tool."""
nonlocal tool_executed
tool_executed = True
return "executed"
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="my_tool",
input={},
)
],
model="test-model",
stopReason="toolUse",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def test_step(context: Context) -> str:
step = await context.sample_step(
messages="Run tool",
tools=[my_tool],
execute_tools=False,
)
assert step.is_tool_use
assert len(step.tool_calls) == 1
assert step.tool_calls[0].name == "my_tool"
# History should include assistant message but no tool results
assert len(step.history) == 2 # user + assistant
return "ok"
async with Client(mcp) as client:
result = await client.call_tool("test_step", {})
assert result.data == "ok"
assert not tool_executed # Tool should not have been executed
async def test_sample_step_history_includes_assistant_message(self):
"""Test that history includes assistant message when execute_tools=False."""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id="call_1",
name="my_tool",
input={"query": "test"},
)
],
model="test-model",
stopReason="toolUse",
)
mcp = FastMCP(sampling_handler=sampling_handler)
def my_tool(query: str) -> str:
return f"result for {query}"
@mcp.tool
async def test_step(context: Context) -> str:
step = await context.sample_step(
messages="Search",
tools=[my_tool],
execute_tools=False,
)
# History should have: user message + assistant message
assert len(step.history) == 2
assert step.history[0].role == "user"
assert step.history[1].role == "assistant"
return "ok"
async with Client(mcp) as client:
result = await client.call_tool("test_step", {})
assert result.data == "ok"

6
tests/experimental/sampling/test_openai_handler.py
View file

@ -8,7 +8,7 @@ from mcp.types import (
SamplingMessage,
TextContent,
)
from openai import OpenAI
from openai import AsyncOpenAI
from openai.types.chat import (
ChatCompletion,
ChatCompletionAssistantMessageParam,
@ -67,13 +67,13 @@ def test_convert_to_openai_messages_raises_on_non_text():
],
)
def test_select_model_from_preferences(prefs, expected):
mock_client = MagicMock(spec=OpenAI)
mock_client = MagicMock(spec=AsyncOpenAI)
handler = OpenAISamplingHandler(default_model="fallback-model", client=mock_client) # type: ignore[arg-type]
assert handler._select_model_from_preferences(prefs) == expected
async def test_chat_completion_to_create_message_result():
mock_client = MagicMock(spec=OpenAI)
mock_client = MagicMock(spec=AsyncOpenAI)
handler = OpenAISamplingHandler(default_model="fallback-model", client=mock_client) # type: ignore[arg-type]
mock_client.chat.completions.create.return_value = ChatCompletion(
id="123",

12
tests/server/middleware/test_tool_injection.py
View file

@ -5,7 +5,7 @@ import math
import pytest
from inline_snapshot import snapshot
from mcp.types import TextContent
from mcp.types import Tool as MCPTool
from mcp.types import Tool as SDKTool
from fastmcp import FastMCP
from fastmcp.client import Client
@ -68,7 +68,7 @@ class TestToolInjectionMiddleware:
base_server.add_middleware(middleware)
async with Client[FastMCPTransport](base_server) as client:
tools: list[MCPTool] = await client.list_tools()
tools: list[SDKTool] = await client.list_tools()
# Should have all tools: multiply, divide, add, subtract
assert len(tools) == 4
@ -249,7 +249,7 @@ class TestToolInjectionMiddleware:
base_server.exclude_tags = {"math"}
async with Client[FastMCPTransport](base_server) as client:
tools: list[MCPTool] = await client.list_tools()
tools: list[SDKTool] = await client.list_tools()
tool_names: list[str] = [tool.name for tool in tools]
assert "multiply" in tool_names
@ -259,7 +259,7 @@ class TestToolInjectionMiddleware:
base_server.add_middleware(middleware)
async with Client[FastMCPTransport](base_server) as client:
tools: list[MCPTool] = await client.list_tools()
tools: list[SDKTool] = await client.list_tools()
result: CallToolResult = await client.call_tool(
name="add", arguments={"a": 3, "b": 4}
)
@ -304,7 +304,7 @@ class TestPromptToolMiddleware:
server_with_prompts.add_middleware(middleware)
async with Client[FastMCPTransport](server_with_prompts) as client:
tools: list[MCPTool] = await client.list_tools()
tools: list[SDKTool] = await client.list_tools()
tool_names: list[str] = [tool.name for tool in tools]
# Should have: add, list_prompts, get_prompt
@ -428,7 +428,7 @@ class TestResourceToolMiddleware:
server_with_resources.add_middleware(middleware)
async with Client[FastMCPTransport](server_with_resources) as client:
tools: list[MCPTool] = await client.list_tools()
tools: list[SDKTool] = await client.list_tools()
tool_names: list[str] = [tool.name for tool in tools]
# Should have: add, list_resources, read_resource

4
tests/server/proxy/test_proxy_client.py
View file

@ -1,5 +1,4 @@
from dataclasses import dataclass
from typing import cast
import pytest
from anyio import create_task_group
@ -40,12 +39,11 @@ def fastmcp_server():
result = await context.sample(
"Hello, world!",
system_prompt="You love FastMCP",
include_context="thisServer",
temperature=0.5,
max_tokens=100,
model_preferences="gpt-4o",
)
return cast(TextContent, result).text
return result.text or ""
@dataclass
class Person:

0
tests/server/sampling/__init__.py Normal file
View file

121
tests/server/sampling/test_sampling_tool.py Normal file
View file

@ -0,0 +1,121 @@
"""Tests for SamplingTool."""
import pytest
from fastmcp.server.sampling import SamplingTool
class TestSamplingToolFromFunction:
"""Tests for SamplingTool.from_function()."""
def test_from_simple_function(self):
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
tool = SamplingTool.from_function(search)
assert tool.name == "search"
assert tool.description == "Search the web."
assert "query" in tool.parameters.get("properties", {})
assert tool.fn is search
def test_from_function_with_overrides(self):
def search(query: str) -> str:
return f"Results for: {query}"
tool = SamplingTool.from_function(
search,
name="web_search",
description="Search the internet",
)
assert tool.name == "web_search"
assert tool.description == "Search the internet"
def test_from_lambda_requires_name(self):
with pytest.raises(ValueError, match="must provide a name for lambda"):
SamplingTool.from_function(lambda x: x)
def test_from_lambda_with_name(self):
tool = SamplingTool.from_function(lambda x: x * 2, name="double")
assert tool.name == "double"
def test_from_async_function(self):
async def async_search(query: str) -> str:
"""Async search."""
return f"Async results for: {query}"
tool = SamplingTool.from_function(async_search)
assert tool.name == "async_search"
assert tool.description == "Async search."
def test_multiple_parameters(self):
def search(query: str, limit: int = 10, include_images: bool = False) -> str:
"""Search with options."""
return f"Results for: {query}"
tool = SamplingTool.from_function(search)
props = tool.parameters.get("properties", {})
assert "query" in props
assert "limit" in props
assert "include_images" in props
class TestSamplingToolRun:
"""Tests for SamplingTool.run()."""
async def test_run_sync_function(self):
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
tool = SamplingTool.from_function(add)
result = await tool.run({"a": 2, "b": 3})
assert result == 5
async def test_run_async_function(self):
async def async_add(a: int, b: int) -> int:
"""Add two numbers asynchronously."""
return a + b
tool = SamplingTool.from_function(async_add)
result = await tool.run({"a": 2, "b": 3})
assert result == 5
async def test_run_with_no_arguments(self):
def get_value() -> str:
"""Return a fixed value."""
return "hello"
tool = SamplingTool.from_function(get_value)
result = await tool.run()
assert result == "hello"
async def test_run_with_none_arguments(self):
def get_value() -> str:
"""Return a fixed value."""
return "hello"
tool = SamplingTool.from_function(get_value)
result = await tool.run(None)
assert result == "hello"
class TestSamplingToolSDKConversion:
"""Tests for SamplingTool._to_sdk_tool() internal method."""
def test_to_sdk_tool(self):
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
tool = SamplingTool.from_function(search)
sdk_tool = tool._to_sdk_tool()
assert sdk_tool.name == "search"
assert sdk_tool.description == "Search the web."
assert "query" in sdk_tool.inputSchema.get("properties", {})

8
tests/server/test_context.py
View file

@ -17,16 +17,16 @@ def context():
class TestParseModelPreferences:
def test_parse_model_preferences_string(self, context):
mp = _parse_model_preferences("claude-3-sonnet")
mp = _parse_model_preferences("claude-haiku-4-5")
assert isinstance(mp, ModelPreferences)
assert mp.hints is not None
assert mp.hints[0].name == "claude-3-sonnet"
assert mp.hints[0].name == "claude-haiku-4-5"
def test_parse_model_preferences_list(self, context):
mp = _parse_model_preferences(["claude-3-sonnet", "claude"])
mp = _parse_model_preferences(["claude-haiku-4-5", "claude"])
assert isinstance(mp, ModelPreferences)
assert mp.hints is not None
assert [h.name for h in mp.hints] == ["claude-3-sonnet", "claude"]
assert [h.name for h in mp.hints] == ["claude-haiku-4-5", "claude"]
def test_parse_model_preferences_object(self, context):
obj = ModelPreferences(hints=[])

8
uv.lock generated
View file

@ -723,7 +723,7 @@ requires-dist = [
{ name = "exceptiongroup", specifier = ">=1.2.2" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "jsonschema-path", specifier = ">=0.3.4" },
{ name = "mcp", specifier = ">=1.23.1" },
{ name = "mcp", specifier = ">=1.24.0" },
{ name = "openai", marker = "extra == 'openai'", specifier = ">=1.102.0" },
{ name = "openapi-pydantic", specifier = ">=0.5.1" },
{ name = "platformdirs", specifier = ">=4.0.0" },
@ -1237,7 +1237,7 @@ wheels = [
[[package]]
name = "mcp"
version = "1.23.3"
version = "1.24.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@ -1255,9 +1255,9 @@ dependencies = [
{ name = "typing-inspection" },
{ name = "uvicorn", marker = "sys_platform != 'emscripten'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a7/a4/d06a303f45997e266f2c228081abe299bbcba216cb806128e2e49095d25f/mcp-1.23.3.tar.gz", hash = "sha256:b3b0da2cc949950ce1259c7bfc1b081905a51916fcd7c8182125b85e70825201", size = 600697, upload-time = "2025-12-09T16:04:37.351Z" }
sdist = { url = "https://files.pythonhosted.org/packages/d6/2c/db9ae5ab1fcdd9cd2bcc7ca3b7361b712e30590b64d5151a31563af8f82d/mcp-1.24.0.tar.gz", hash = "sha256:aeaad134664ce56f2721d1abf300666a1e8348563f4d3baff361c3b652448efc", size = 604375, upload-time = "2025-12-12T14:19:38.205Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/32/c6/13c1a26b47b3f3a3b480783001ada4268917c9f42d78a079c336da2e75e5/mcp-1.23.3-py3-none-any.whl", hash = "sha256:32768af4b46a1b4f7df34e2bfdf5c6011e7b63d7f1b0e321d0fdef4cd6082031", size = 231570, upload-time = "2025-12-09T16:04:35.56Z" },
{ url = "https://files.pythonhosted.org/packages/61/0d/5cf14e177c8ae655a2fd9324a6ef657ca4cafd3fc2201c87716055e29641/mcp-1.24.0-py3-none-any.whl", hash = "sha256:db130e103cc50ddc3dffc928382f33ba3eaef0b711f7a87c05e7ded65b1ca062", size = 232896, upload-time = "2025-12-12T14:19:36.14Z" },
]
[[package]]