Compare commits

...

10 commits

Author SHA1 Message Date
Jeremiah Lowin
5d57f5bd0f Use claude-sonnet-4-5-20250929 model name 2025-12-09 19:29:02 -05:00
Jeremiah Lowin
179874628e Document sampling handlers as client feature, reference from server docs
- Add Pre-built Handlers section to clients/sampling.mdx with Anthropic and OpenAI
- Simplify servers/sampling.mdx to reference client docs for handler details
- Add examples for Azure OpenAI and local model providers
2025-12-09 19:26:46 -05:00
Jeremiah Lowin
d34f065417 Move OpenAI sampling handler out of experimental
- Move from fastmcp.experimental.sampling.handlers.openai to fastmcp.server.sampling.openai
- Update docs and tests to use new import path
- Remove experimental handlers directory
2025-12-09 15:07:13 -05:00
Jeremiah Lowin
7e02fa29b2 Add Anthropic sampling handler and consolidate sampling examples
- Add AnthropicSamplingHandler in server/sampling/anthropic.py
- Consolidate all sampling examples into examples/sampling/ with rich output
- Examples: text.py, structured_output.py, tool_use.py, server_fallback.py
- Add anthropic optional dependency
2025-12-09 15:04:35 -05:00
Jeremiah Lowin
b544cddca7 Merge main into sampling-tools-sep-1577 2025-12-04 22:23:54 -05:00
Jeremiah Lowin
5c0f05511b 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.
2025-12-04 22:19:27 -05:00
Jeremiah Lowin
ead4730423 Fix tool result content handling in OpenAI handler 2025-12-04 17:20:05 -05:00
Jeremiah Lowin
0d8679c0ad 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
2025-12-04 12:41:19 -05:00
Jeremiah Lowin
5fb9103e4c WIP: Sampling API with SamplingResult[T] and result_type 2025-12-04 11:39:30 -05:00
Jeremiah Lowin
84f17361dd MCP → SDK (vocab change only) 2025-12-04 10:52:35 -05:00
32 changed files with 4882 additions and 1207 deletions

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 [Handling Tool Requests](#handling-tool-requests).
</ResponseField>
<ResponseField name="toolChoice" type="ToolChoice | None">
Optional control over tool usage behavior (`auto`, `required`, or `none`).
</ResponseField>
</Expandable>
</ResponseField>
@ -153,4 +161,201 @@ 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>
<Tip>
When you provide a `sampling_handler`, FastMCP automatically advertises full sampling capabilities including tool support to the server.
</Tip>
## Handling Tool Requests
<VersionBadge version="2.14.0" />
Servers may request sampling with tools, allowing the LLM to make tool calls during generation. When tools are provided in `params.tools`, your handler should return a `CreateMessageResultWithTools` object instead of a simple string.
### Checking for Tools
```python
from fastmcp import Client
from fastmcp.client.sampling import SamplingMessage, SamplingParams, RequestContext
from mcp.types import (
CreateMessageResultWithTools,
TextContent,
ToolUseContent,
ToolChoice,
)
async def sampling_handler_with_tools(
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext
) -> str | CreateMessageResultWithTools:
# Check if tools were provided
if params.tools:
# Call your LLM with tool support
# Return CreateMessageResultWithTools with appropriate content
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="I'll help you with that.")],
model="gpt-4",
stopReason="endTurn",
)
# Standard text response when no tools
return "Generated response"
client = Client(
"my_mcp_server.py",
sampling_handler=sampling_handler_with_tools,
)
```
### Returning Tool Use
When the LLM wants to call a tool, return a `CreateMessageResultWithTools` with `stopReason="toolUse"` and `ToolUseContent` in the content:
```python
from mcp.types import CreateMessageResultWithTools, ToolUseContent
async def sampling_handler_with_tools(
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext
) -> str | CreateMessageResultWithTools:
if params.tools:
# Your LLM decided to call a tool
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="toolUse",
id="call_123",
name="search",
input={"query": "Python tutorials"},
)
],
model="gpt-4",
stopReason="toolUse", # Indicates tool use
)
return "Response without tools"
```
### Tool Choice
The `params.toolChoice` field indicates how the server wants tools to be used:
- **`auto`**: The LLM decides whether to use tools
- **`required`**: The LLM must use at least one tool
- **`none`**: The LLM should not use any tools
```python
async def sampling_handler_with_tools(
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext
) -> str | CreateMessageResultWithTools:
if params.tools and params.toolChoice:
if params.toolChoice.mode == "required":
# Must return a tool call
pass
elif params.toolChoice.mode == "none":
# Should not return tool calls even though tools are available
pass
# "auto" - let the LLM decide
return "Response"
```
<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>
## Pre-built Handlers
<VersionBadge version="2.14.0" />
FastMCP provides ready-to-use sampling handlers for Anthropic and OpenAI that handle all the complexity of message conversion, tool formatting, and response parsing.
### Anthropic Handler
The Anthropic handler uses the Claude API:
```python
from anthropic import Anthropic
from fastmcp import Client
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
handler = AnthropicSamplingHandler(
default_model="claude-sonnet-4-5-20250929",
client=Anthropic(), # Uses ANTHROPIC_API_KEY env var
)
async with Client("server.py", sampling_handler=handler) as client:
result = await client.call_tool("summarize", {"text": "..."})
```
The handler automatically:
- Converts MCP messages to Anthropic's format
- Translates tool definitions to Anthropic's tool format
- Maps stop reasons (`tool_use` → `toolUse`, `end_turn` → `endTurn`)
- Selects models based on server preferences (any model starting with `claude` is accepted)
<Note>
Requires the `anthropic` package. Install with `pip install fastmcp[anthropic]` or `pip install anthropic`.
</Note>
### OpenAI Handler
The OpenAI handler works with the OpenAI API and compatible providers:
```python
from openai import OpenAI
from fastmcp import Client
from fastmcp.server.sampling.openai import OpenAISamplingHandler
handler = OpenAISamplingHandler(
default_model="gpt-4o-mini",
client=OpenAI(), # Uses OPENAI_API_KEY env var
)
async with Client("server.py", sampling_handler=handler) as client:
result = await client.call_tool("analyze", {"data": "..."})
```
The handler automatically:
- Converts MCP messages to OpenAI's chat format
- Translates tool definitions to OpenAI's function calling format
- Maps stop reasons (`tool_calls` → `toolUse`, `stop` → `endTurn`)
- Selects models based on server preferences
<Note>
Requires the `openai` package. Install with `pip install fastmcp[openai]` or `pip install openai`.
</Note>
### Using with Compatible Providers
The OpenAI handler works with any OpenAI-compatible API:
```python
from openai import OpenAI
from fastmcp.server.sampling.openai import OpenAISamplingHandler
# Azure OpenAI
handler = OpenAISamplingHandler(
default_model="gpt-4o",
client=OpenAI(
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
base_url="https://your-resource.openai.azure.com/openai/deployments/your-deployment",
),
)
# Local models (Ollama, vLLM, etc.)
handler = OpenAISamplingHandler(
default_model="llama3",
client=OpenAI(
api_key="not-needed",
base_url="http://localhost:11434/v1",
),
)
```

View file

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

View file

@ -1,52 +0,0 @@
"""
Example of using sampling to request an LLM completion via Marvin
"""
import asyncio
import marvin
from mcp.types import TextContent
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
# -- Create a server that sends a sampling request to the LLM
mcp = FastMCP("Sampling Example")
@mcp.tool
async def example_tool(prompt: str, context: Context) -> str:
"""Sample a completion from the LLM."""
response = await context.sample(
"What is your favorite programming language?",
system_prompt="You love languages named after snakes.",
)
assert isinstance(response, TextContent)
return response.text
# -- Create a client that can handle the sampling request
async def sampling_fn(
messages: list[SamplingMessage],
params: SamplingParams,
ctx: RequestContext,
) -> str:
return await marvin.say_async(
message=[m.content.text for m in messages],
instructions=params.systemPrompt,
)
async def run():
async with Client(mcp, sampling_handler=sampling_fn) as client:
result = await client.call_tool(
"example_tool", {"prompt": "What is the best programming language?"}
)
print(result)
if __name__ == "__main__":
asyncio.run(run())

View file

@ -0,0 +1,151 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Server-Side Sampling Fallback Example
When the CLIENT has no sampling handler, the SERVER's handler is used instead.
MCP Flow with Fallback:
1. Client calls server tool (client has NO sampling handler)
2. Server tool calls ctx.sample()
3. Client can't handle sampling → server's fallback handler is used
4. Server's handler calls the LLM directly
5. Response returns to server, then to client
Run:
python examples/sampling/server_fallback.py
"""
import asyncio
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from fastmcp import Client, Context, FastMCP
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
console = Console()
# ============================================================================
# SERVER - Note: sampling_handler is on the SERVER, not the client
# ============================================================================
class LoggingAnthropicHandler(AnthropicSamplingHandler):
async def __call__(self, messages, params, context):
console.print(
" [bold blue]⚡ FALLBACK[/] Server's handler calling Claude...",
highlight=False,
)
result = await super().__call__(messages, params, context)
console.print(
" [bold blue]⚡ FALLBACK[/] Response received", highlight=False
)
return result
mcp = FastMCP(
"Server with Fallback",
sampling_handler=LoggingAnthropicHandler(default_model="claude-sonnet-4-5-20250929"),
)
@mcp.tool
async def get_fun_fact(topic: str, ctx: Context) -> str:
"""Get a fun fact about a topic."""
console.print(" [bold yellow]📦 SERVER[/] Tool 'get_fun_fact' called")
console.print(
" [bold yellow]📦 SERVER[/] Requesting sampling (client has no handler!)"
)
result = await ctx.sample(
messages=f"Tell me one fun fact about: {topic}",
system_prompt="Share a fascinating fact in 1-2 sentences.",
)
console.print(" [bold yellow]📦 SERVER[/] Got response via fallback handler")
return result.text # type: ignore[return-value]
@mcp.tool
async def translate(text: str, language: str, ctx: Context) -> str:
"""Translate text to another language."""
console.print(" [bold yellow]📦 SERVER[/] Tool 'translate' called")
console.print(
" [bold yellow]📦 SERVER[/] Requesting sampling (client has no handler!)"
)
result = await ctx.sample(
messages=f"Translate to {language}: {text}",
system_prompt="Provide only the translation.",
)
console.print(" [bold yellow]📦 SERVER[/] Got response via fallback handler")
return result.text # type: ignore[return-value]
# ============================================================================
# CLIENT - Note: NO sampling_handler provided!
# ============================================================================
async def main():
console.print()
console.print(
Panel.fit(
"[bold]Server Fallback Example[/]\n\n"
"The [green]CLIENT[/] has [bold red]no sampling handler[/].\n"
"The [yellow]SERVER's[/] fallback handler is used instead.",
border_style="bright_black",
)
)
console.print()
# IMPORTANT: No sampling_handler!
async with Client(mcp) as client:
console.rule(style="dim")
console.print(
"[bold green]🖥️ CLIENT[/] Calling 'get_fun_fact' [dim](no sampling handler!)[/]"
)
console.print()
result = await client.call_tool("get_fun_fact", {"topic": "octopuses"})
console.print()
console.print("[bold green]🖥️ CLIENT[/] Result:")
console.print(Panel(result.data, border_style="green"))
console.print()
console.rule(style="dim")
console.print(
"[bold green]🖥️ CLIENT[/] Calling 'translate' [dim](no sampling handler!)[/]"
)
console.print()
result = await client.call_tool(
"translate", {"text": "Hello, how are you?", "language": "Spanish"}
)
console.print()
console.print("[bold green]🖥️ CLIENT[/] Result:")
console.print(Panel(result.data, border_style="green"))
console.print()
# Summary
summary = Text()
summary.append("Key concept: ", style="bold")
summary.append(
"When the client lacks sampling support, the server's\n"
"fallback handler steps in. This lets servers guarantee\n"
"sampling works regardless of client capabilities."
)
console.print(Panel(summary, border_style="bright_black"))
console.print()
if __name__ == "__main__":
asyncio.run(main())

View file

@ -0,0 +1,163 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Structured Output Example
Use `result_type` to get validated Pydantic models from an LLM.
MCP Flow:
1. Client calls server tool
2. Server makes sampling request with result_type=YourModel
3. LLM response is parsed and validated against the Pydantic schema
4. Server returns structured data to client
Run:
python examples/sampling/structured_output.py
"""
import asyncio
from enum import Enum
from pydantic import BaseModel, Field
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from fastmcp import Client, Context, FastMCP
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
console = Console()
# ============================================================================
# PYDANTIC MODELS - Define the schema for LLM responses
# ============================================================================
class Sentiment(str, Enum):
POSITIVE = "positive"
NEGATIVE = "negative"
NEUTRAL = "neutral"
MIXED = "mixed"
class SentimentAnalysis(BaseModel):
sentiment: Sentiment = Field(description="Overall sentiment")
confidence: float = Field(ge=0.0, le=1.0, description="Confidence 0.0-1.0")
reasoning: str = Field(description="Brief explanation")
key_phrases: list[str] = Field(description="Influential phrases")
# ============================================================================
# SAMPLING HANDLER WITH LOGGING
# ============================================================================
class LoggingAnthropicHandler(AnthropicSamplingHandler):
async def __call__(self, messages, params, context):
console.print(
" [bold blue]⚡ SAMPLING[/] Calling Claude API...", highlight=False
)
result = await super().__call__(messages, params, context)
console.print(
" [bold blue]⚡ SAMPLING[/] Response received", highlight=False
)
return result
# ============================================================================
# SERVER
# ============================================================================
mcp = FastMCP("Sentiment Analyzer")
@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> dict:
"""Analyze sentiment and return structured results."""
console.print(" [bold yellow]📦 SERVER[/] Tool 'analyze_sentiment' called")
console.print(
" [bold yellow]📦 SERVER[/] Sampling with result_type=SentimentAnalysis"
)
result = await ctx.sample(
messages=f"Analyze the sentiment of this text:\n\n{text}",
system_prompt="You are a sentiment analysis expert.",
result_type=SentimentAnalysis,
)
console.print(
f" [bold yellow]📦 SERVER[/] Got validated {type(result.result).__name__}"
)
return result.result.model_dump() # type: ignore[union-attr]
# ============================================================================
# CLIENT
# ============================================================================
async def main():
console.print()
console.print(
Panel.fit(
"[bold]Structured Output Example[/]\n\n"
"The [cyan]result_type[/] parameter ensures LLM responses\n"
"match your Pydantic schema.",
border_style="bright_black",
)
)
console.print()
handler = LoggingAnthropicHandler(default_model="claude-sonnet-4-5-20250929")
test_texts = [
("I love this! Best purchase ever!", "😊"),
("Terrible. Would not recommend.", "😞"),
("It's okay. Nothing special.", "😐"),
]
async with Client(mcp, sampling_handler=handler) as client:
for text, _ in test_texts:
console.rule(style="dim")
console.print(f"[bold green]🖥️ CLIENT[/] Analyzing: [italic]{text}[/]")
console.print()
result = await client.call_tool("analyze_sentiment", {"text": text})
data = result.data
# Display result in a nice table
console.print()
console.print("[bold green]🖥️ CLIENT[/] Received structured result:")
emoji = {"positive": "😊", "negative": "😞", "neutral": "😐", "mixed": "🤔"}
table = Table(show_header=False, box=None, padding=(0, 2))
table.add_column(style="bold")
table.add_column()
table.add_row(
"Sentiment", f"{emoji.get(data['sentiment'], '')} {data['sentiment']}"
)
table.add_row(
"Confidence",
f"[green]{'' * int(data['confidence'] * 10)}[/][dim]{'' * (10 - int(data['confidence'] * 10))}[/] {data['confidence']:.0%}",
)
table.add_row("Reasoning", data["reasoning"])
table.add_row("Key phrases", ", ".join(data["key_phrases"]))
console.print(Panel(table, border_style="green"))
console.print()
console.print(
Panel(
"[bold]Key concept:[/] The result_type parameter enforces a Pydantic schema.\n"
"Invalid LLM responses are automatically rejected and retried.",
border_style="bright_black",
)
)
console.print()
if __name__ == "__main__":
asyncio.run(main())

152
examples/sampling/text.py Normal file
View file

@ -0,0 +1,152 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Text Sampling Example
The simplest form of sampling: send text to an LLM and get text back.
MCP Sampling Flow:
1. Client calls a tool on the server
2. Server tool needs LLM help, makes a sampling request
3. The sampling handler (on client) calls the LLM
4. Response flows back through the chain
Run:
python examples/sampling/text.py
"""
import asyncio
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from fastmcp import Client, Context, FastMCP
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
console = Console()
# ============================================================================
# SAMPLING HANDLER WITH LOGGING
# Wraps AnthropicSamplingHandler to show when LLM calls happen
# ============================================================================
class LoggingAnthropicHandler(AnthropicSamplingHandler):
"""Sampling handler that logs when it calls the LLM."""
async def __call__(self, messages, params, context):
console.print(
" [bold blue]⚡ SAMPLING[/] Calling Claude API...", highlight=False
)
result = await super().__call__(messages, params, context)
console.print(
" [bold blue]⚡ SAMPLING[/] Response received", highlight=False
)
return result
# ============================================================================
# SERVER
# ============================================================================
mcp = FastMCP("Creative Writer")
@mcp.tool
async def write_haiku(topic: str, ctx: Context) -> str:
"""Write a haiku about the given topic."""
console.print(
f" [bold yellow]📦 SERVER[/] Tool 'write_haiku' called with topic={topic!r}"
)
console.print(" [bold yellow]📦 SERVER[/] Requesting LLM completion...")
result = await ctx.sample(
messages=f"Write a haiku about: {topic}",
system_prompt="You are a poet. Write only the haiku, nothing else.",
)
console.print(" [bold yellow]📦 SERVER[/] Returning result to client")
return result.text # type: ignore[return-value]
@mcp.tool
async def explain_simply(concept: str, ctx: Context) -> str:
"""Explain a concept in simple terms."""
console.print(
f" [bold yellow]📦 SERVER[/] Tool 'explain_simply' called with concept={concept!r}"
)
console.print(" [bold yellow]📦 SERVER[/] Requesting LLM completion...")
result = await ctx.sample(
messages=f"Explain this concept: {concept}",
system_prompt="Explain in 1-2 simple sentences a child could understand.",
)
console.print(" [bold yellow]📦 SERVER[/] Returning result to client")
return result.text # type: ignore[return-value]
# ============================================================================
# CLIENT
# ============================================================================
async def main():
console.print()
console.print(
Panel.fit(
"[bold]Text Sampling Example[/]\n\n"
"Watch the flow: [green]CLIENT[/] → [yellow]SERVER[/] → [blue]SAMPLING[/] → [yellow]SERVER[/] → [green]CLIENT[/]",
border_style="bright_black",
)
)
console.print()
handler = LoggingAnthropicHandler(default_model="claude-sonnet-4-5-20250929")
async with Client(mcp, sampling_handler=handler) as client:
# Example 1
console.rule("[bold green]Example 1: Write a Haiku", style="green")
console.print("[bold green]🖥️ CLIENT[/] Calling tool 'write_haiku'")
console.print()
result = await client.call_tool("write_haiku", {"topic": "async programming"})
console.print()
console.print("[bold green]🖥️ CLIENT[/] Got result:")
console.print(Panel(result.data, border_style="green", padding=(0, 2)))
console.print()
# Example 2
console.rule("[bold green]Example 2: Simple Explanation", style="green")
console.print("[bold green]🖥️ CLIENT[/] Calling tool 'explain_simply'")
console.print()
result = await client.call_tool("explain_simply", {"concept": "recursion"})
console.print()
console.print("[bold green]🖥️ CLIENT[/] Got result:")
console.print(Panel(result.data, border_style="green", padding=(0, 2)))
console.print()
# Summary
summary = Text()
summary.append("Flow: ", style="bold")
summary.append("CLIENT", style="green")
summary.append(" calls tool → ")
summary.append("SERVER", style="yellow")
summary.append(" needs LLM → ")
summary.append("SAMPLING", style="blue")
summary.append(" calls Claude → response flows back")
console.print(
Panel(summary, title="[bold]How it works[/]", border_style="bright_black")
)
console.print()
if __name__ == "__main__":
asyncio.run(main())

View file

@ -0,0 +1,169 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Tool Use Example
Give the LLM tools to use during sampling.
MCP Flow with Tools:
1. Client calls server tool
2. Server makes sampling request with tools=[...]
3. LLM decides to call a tool tool executes result fed back to LLM
4. Loop continues until LLM gives final answer (or max_iterations)
5. Server returns final response to client
Run:
python examples/sampling/tool_use.py
"""
import asyncio
import random
from datetime import datetime
from rich.console import Console
from rich.panel import Panel
from fastmcp import Client, Context, FastMCP
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
console = Console()
# ============================================================================
# TOOLS - Functions the LLM can call during sampling
# ============================================================================
def calculate(expression: str) -> str:
"""Evaluate a math expression. Use Python syntax (e.g., 2**10 for power)."""
console.print(
f" [bold magenta]🔧 TOOL[/] calculate({expression!r})", highlight=False
)
try:
allowed = {"abs": abs, "round": round, "min": min, "max": max}
result = eval(expression, {"__builtins__": {}}, allowed)
console.print(f" [bold magenta]🔧 TOOL[/] → {result}", highlight=False)
return str(result)
except Exception as e:
return f"Error: {e}"
def get_current_time() -> str:
"""Get the current date and time."""
console.print(
" [bold magenta]🔧 TOOL[/] get_current_time()", highlight=False
)
result = datetime.now().strftime("%A, %B %d, %Y at %I:%M %p")
console.print(f" [bold magenta]🔧 TOOL[/] → {result}", highlight=False)
return result
def roll_dice(sides: int = 6, count: int = 1) -> str:
"""Roll dice and return results."""
console.print(
f" [bold magenta]🔧 TOOL[/] roll_dice(sides={sides}, count={count})",
highlight=False,
)
rolls = [random.randint(1, sides) for _ in range(count)]
result = f"Rolled {count}d{sides}: {rolls} (total: {sum(rolls)})"
console.print(f" [bold magenta]🔧 TOOL[/] → {result}", highlight=False)
return result
# ============================================================================
# SAMPLING HANDLER WITH LOGGING
# ============================================================================
class LoggingAnthropicHandler(AnthropicSamplingHandler):
async def __call__(self, messages, params, context):
console.print(
" [bold blue]⚡ SAMPLING[/] Calling Claude API...", highlight=False
)
result = await super().__call__(messages, params, context)
console.print(
" [bold blue]⚡ SAMPLING[/] Response received", highlight=False
)
return result
# ============================================================================
# SERVER
# ============================================================================
mcp = FastMCP("Assistant with Tools")
@mcp.tool
async def ask(question: str, ctx: Context) -> str:
"""Ask a question. The LLM can use tools to help answer."""
console.print(" [bold yellow]📦 SERVER[/] Tool 'ask' called")
console.print(
" [bold yellow]📦 SERVER[/] Sampling with tools=[calculate, get_current_time, roll_dice]"
)
console.print()
result = await ctx.sample(
messages=question,
system_prompt="You have tools available. Use them when helpful. Be concise.",
tools=[calculate, get_current_time, roll_dice],
max_iterations=5,
)
console.print()
console.print(" [bold yellow]📦 SERVER[/] Tool loop complete, returning answer")
return result.text # type: ignore[return-value]
# ============================================================================
# CLIENT
# ============================================================================
async def main():
console.print()
console.print(
Panel.fit(
"[bold]Tool Use Example[/]\n\n"
"Watch the LLM use [magenta]TOOLS[/] during sampling.\n"
"The tool loop runs until the LLM has a final answer.",
border_style="bright_black",
)
)
console.print()
handler = LoggingAnthropicHandler(default_model="claude-sonnet-4-5-20250929")
questions = [
"What's 15% tip on a $47.50 bill?",
"What time is it right now?",
"Roll 2 dice and tell me if I got doubles.",
]
async with Client(mcp, sampling_handler=handler) as client:
for question in questions:
console.rule(style="dim")
console.print(f"[bold green]🖥️ CLIENT[/] Question: [italic]{question}[/]")
console.print()
result = await client.call_tool("ask", {"question": question})
console.print()
console.print("[bold green]🖥️ CLIENT[/] Answer:")
console.print(Panel(result.data, border_style="green"))
console.print()
console.print(
Panel(
"[bold]Key concept:[/] The [cyan]tools[/] parameter lets the LLM call functions.\n"
"FastMCP handles the tool execution loop automatically.\n"
"[cyan]max_iterations[/] prevents infinite loops.",
border_style="bright_black",
)
)
console.print()
if __name__ == "__main__":
asyncio.run(main())

View file

@ -1,38 +0,0 @@
# /// script
# dependencies = ["openai", "fastmcp"]
# ///
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=os.getenv("MODEL") or "gpt-4o-mini", # pyright: ignore[reportArgumentType]
client=OpenAI(
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL"),
),
),
)
@server.tool
async def test_sample_fallback(ctx: Context) -> ContentBlock:
return await ctx.sample(
messages=["hello world!"],
)
await server.run_http_async()
if __name__ == "__main__":
asyncio.run(async_main())

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.23.3",
"openapi-pydantic>=0.5.1",
"platformdirs>=4.0.0",
"pydocket>=0.14.0",
@ -47,12 +47,13 @@ classifiers = [
]
[project.optional-dependencies]
anthropic = ["anthropic>=0.40.0"]
openai = ["openai>=1.102.0"]
[dependency-groups]
dev = [
"dirty-equals>=0.9.0",
"fastmcp[openai]",
"fastmcp[anthropic,openai]",
# add optional dependencies for fastmcp dev
"fastapi>=0.115.12",
"inline-snapshot[dirty-equals]>=0.27.2",

View file

@ -250,6 +250,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,
@ -307,6 +308,13 @@ 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
or mcp.types.SamplingCapability(
tools=mcp.types.SamplingToolsCapability()
)
)
if elicitation_handler is not None:
self._session_kwargs["elicitation_callback"] = create_elicitation_callback(
@ -344,11 +352,20 @@ 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
or mcp.types.SamplingCapability(tools=mcp.types.SamplingToolsCapability())
)
def set_elicitation_callback(
self, elicitation_callback: ElicitationHandler

View file

@ -65,6 +65,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

View file

@ -1,21 +0,0 @@
from abc import ABC, abstractmethod
from collections.abc import Awaitable
from mcp import ClientSession, 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,
)
class BaseLLMSamplingHandler(ABC):
@abstractmethod
def __call__(
self,
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext[ServerSession, LifespanContextT]
| RequestContext[ClientSession, LifespanContextT],
) -> str | CreateMessageResult | Awaitable[str | CreateMessageResult]: ...

View file

@ -1,170 +0,0 @@
from collections.abc import Iterator, Sequence
from typing import 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,
ModelPreferences,
SamplingMessage,
TextContent,
)
try:
from openai import NOT_GIVEN, OpenAI
from openai.types.chat import (
ChatCompletion,
ChatCompletionAssistantMessageParam,
ChatCompletionMessageParam,
ChatCompletionSystemMessageParam,
ChatCompletionUserMessageParam,
)
from openai.types.shared.chat_model import ChatModel
except ImportError as e:
raise ImportError(
"The `openai` package is not installed. Please install `fastmcp[openai]` or add `openai` to your dependencies manually."
) from e
from typing_extensions import override
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()
self.default_model: ChatModel = default_model
@override
async def __call__(
self,
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext[ServerSession, LifespanContextT]
| RequestContext[ClientSession, LifespanContextT],
) -> CreateMessageResult:
openai_messages: list[ChatCompletionMessageParam] = (
self._convert_to_openai_messages(
system_prompt=params.systemPrompt,
messages=messages,
)
)
model: ChatModel = self._select_model_from_preferences(params.modelPreferences)
response = self.client.chat.completions.create(
model=model,
messages=openai_messages,
temperature=params.temperature or NOT_GIVEN,
max_tokens=params.maxTokens,
stop=params.stopSequences or NOT_GIVEN,
)
return self._chat_completion_to_create_message_result(response)
@staticmethod
def _iter_models_from_preferences(
model_preferences: ModelPreferences | str | list[str] | None,
) -> Iterator[str]:
if model_preferences is None:
return
if isinstance(model_preferences, str) and model_preferences in get_args(
ChatModel
):
yield model_preferences
if isinstance(model_preferences, list):
yield from model_preferences
if isinstance(model_preferences, ModelPreferences):
if not (hints := model_preferences.hints):
return
for hint in hints:
if not (name := hint.name):
continue
yield name
@staticmethod
def _convert_to_openai_messages(
system_prompt: str | None, messages: Sequence[SamplingMessage]
) -> list[ChatCompletionMessageParam]:
openai_messages: list[ChatCompletionMessageParam] = []
if system_prompt:
openai_messages.append(
ChatCompletionSystemMessageParam(
role="system",
content=system_prompt,
)
)
if isinstance(messages, str):
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=messages,
)
)
if isinstance(messages, list):
for message in messages:
if isinstance(message, str):
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=message,
)
)
continue
if not isinstance(message.content, TextContent):
raise ValueError("Only text content is supported")
if message.role == "user":
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=message.content.text,
)
)
else:
openai_messages.append(
ChatCompletionAssistantMessageParam(
role="assistant",
content=message.content.text,
)
)
return openai_messages
@staticmethod
def _chat_completion_to_create_message_result(
chat_completion: ChatCompletion,
) -> CreateMessageResult:
if len(chat_completion.choices) == 0:
raise ValueError("No response for completion")
first_choice = chat_completion.choices[0]
if content := first_choice.message.content:
return CreateMessageResult(
content=TextContent(type="text", text=content),
role="assistant",
model=chat_completion.model,
)
raise ValueError("No content in response from completion")
def _select_model_from_preferences(
self, model_preferences: ModelPreferences | str | list[str] | None
) -> ChatModel:
for model_option in self._iter_models_from_preferences(model_preferences):
if model_option in get_args(ChatModel):
chosen_model: ChatModel = model_option # pyright: ignore[reportAssignmentType]
return chosen_model
return self.default_model

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
@ -94,10 +94,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,
@ -105,7 +105,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,

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,
@ -132,10 +132,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),

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,
@ -201,10 +201,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),
@ -218,7 +218,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.

View file

@ -2,15 +2,16 @@ 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 enum import Enum
from logging import Logger
from typing import Any, Literal, cast, get_origin, overload
from typing import Any, Generic, Literal, cast, get_origin, overload
import anyio
from mcp import LoggingLevel, ServerSession
@ -20,6 +21,7 @@ from mcp.shared.context import RequestContext
from mcp.types import (
ClientCapabilities,
CreateMessageResult,
CreateMessageResultWithTools,
GetPromptResult,
IncludeContext,
ModelHint,
@ -28,11 +30,16 @@ from mcp.types import (
SamplingCapability,
SamplingMessage,
SamplingMessageContentBlock,
SamplingToolsCapability,
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.networks import AnyUrl
from starlette.requests import Request
from typing_extensions import TypeVar
@ -44,7 +51,10 @@ from fastmcp.server.elicitation import (
ScalarElicitationType,
get_elicitation_schema,
)
from fastmcp.server.sampling import SamplingTool
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import Tool as FastMCPTool
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
@ -58,8 +68,36 @@ _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]
@dataclass
class SamplingResult(Generic[ResultT]):
"""Result from ctx.sample() containing the response and history.
This is a generic class where ResultT defaults to str for text responses.
When a result_type is specified, ResultT is that type.
Attributes:
text: The text representation of the result. For text responses, this is
the raw text. For structured responses, this is the JSON representation.
None if the response was a tool use (for manual loop building).
result: The typed result. When ResultT is str, this equals text. When ResultT
is a custom type, this is the validated/parsed object.
history: List of all messages exchanged during sampling, including tool calls
and results. Useful for continuing conversations or debugging.
"""
text: str | None
result: ResultT
history: list[SamplingMessage]
_flush_lock = anyio.Lock()
@ -247,7 +285,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:
@ -255,7 +293,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:
@ -486,21 +524,96 @@ class Context:
"""Send a prompt list changed notification to the client."""
await self.session.send_prompt_list_changed()
@overload
async def sample(
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 | FastMCPTool | Callable[..., Any]] | None = None,
tool_choice: ToolChoiceOption | None = None,
max_iterations: int = 10,
result_type: type[ResultT],
) -> SamplingResult[ResultT]:
"""Overload: With result_type, returns SamplingResult[ResultT]."""
@overload
async def sample(
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,
tools: Sequence[SamplingTool | FastMCPTool | Callable[..., Any]] | None = None,
tool_choice: ToolChoiceOption | None = None,
max_iterations: int = 10,
result_type: 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,
include_context: IncludeContext | None = None,
temperature: float | None = None,
max_tokens: int | None = None,
model_preferences: ModelPreferences | str | list[str] | None = None,
tools: Sequence[SamplingTool | FastMCPTool | Callable[..., Any]] | None = None,
tool_choice: ToolChoiceOption | None = None,
max_iterations: int = 10,
result_type: type[ResultT] | 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.
When tools are provided, the method automatically 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 response or max_iterations is reached.
When result_type is specified (not str), 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`.
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.
include_context: Optional context inclusion setting.
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, SamplingTools, or FastMCP Tools (all are auto-converted).
When provided, the method automatically handles tool execution
and returns the final response after all tool calls complete.
tool_choice: Optional control over tool usage behavior. Only valid
when tools are provided.
max_iterations: Maximum number of LLM calls before returning.
Defaults to 10. Set to 1 for single-iteration mode where you
can build your own loop using the history.
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.
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
"""
if max_tokens is None:
@ -520,6 +633,40 @@ class Context:
for m in messages
]
# Convert tools to SamplingTools, then to SDK tools
sampling_tools: list[SamplingTool] = []
if tools is not None:
for t in tools:
if isinstance(t, SamplingTool):
sampling_tools.append(t)
elif isinstance(t, FastMCPTool):
sampling_tools.append(SamplingTool.from_mcp_tool(t))
elif callable(t):
sampling_tools.append(SamplingTool.from_function(t))
else:
raise TypeError(
f"Expected SamplingTool, FastMCP Tool, or callable, got {type(t)}"
)
# Create synthetic final_response tool for structured output
final_response_tool: SamplingTool | 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.append(final_response_tool)
# Add hint to system prompt
hint = "Call final_response when you have completed the task."
system_prompt = f"{system_prompt}\n\n{hint}" if system_prompt else hint
sdk_tools: list[SDKTool] | None = None
if sampling_tools:
sdk_tools = [t._to_sdk_tool() for t in sampling_tools]
# Build a tool lookup for execution
tool_map: dict[str, SamplingTool] = {}
if sampling_tools:
tool_map = {t.name: t for t in sampling_tools}
should_fallback = (
self.fastmcp.sampling_handler_behavior == "fallback"
and not self.session.check_client_capability(
@ -531,33 +678,50 @@ class Context:
if self.fastmcp.sampling_handler is None:
raise ValueError("Client does not support sampling")
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,
return await self._sample_with_fallback_handler(
sampling_messages=sampling_messages,
system_prompt=system_prompt,
temperature=temperature,
max_tokens=max_tokens,
model_preferences=model_preferences,
sdk_tools=sdk_tools,
tool_choice=tool_choice,
tool_map=tool_map,
max_iterations=max_iterations,
result_type=result_type,
final_response_tool=final_response_tool,
)
if inspect.isawaitable(create_message_result):
create_message_result = await create_message_result
if isinstance(create_message_result, str):
return TextContent(text=create_message_result, type="text")
if isinstance(create_message_result, CreateMessageResult):
return create_message_result.content
else:
# When tools are provided, we need the client to support sampling.tools
if sampling_tools:
has_tools_capability = self.session.check_client_capability(
capability=ClientCapabilities(
sampling=SamplingCapability(tools=SamplingToolsCapability())
)
)
if not has_tools_capability:
raise ValueError(
f"Unexpected sampling handler result: {create_message_result}"
"Client does not support sampling with tools. "
"The client must advertise the sampling.tools capability."
)
result: CreateMessageResult = await self.session.create_message(
return await self._sample_with_client(
sampling_messages=sampling_messages,
system_prompt=system_prompt,
include_context=include_context,
temperature=temperature,
max_tokens=max_tokens,
model_preferences=model_preferences,
sdk_tools=sdk_tools,
tool_choice=tool_choice,
tool_map=tool_map,
max_iterations=max_iterations,
result_type=result_type,
final_response_tool=final_response_tool,
)
# Simple case: no tools, no structured output
result = await self.session.create_message(
messages=sampling_messages,
system_prompt=system_prompt,
include_context=include_context,
@ -567,7 +731,355 @@ class Context:
related_request_id=self.request_id,
)
return result.content
# Extract text from content
text = _extract_text_from_content(result.content)
return SamplingResult(
text=text,
result=cast(ResultT, text),
history=list(sampling_messages),
)
async def _sample_with_fallback_handler(
self,
sampling_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: ToolChoiceOption | None,
tool_map: dict[str, SamplingTool],
max_iterations: int,
result_type: type[ResultT] | None,
final_response_tool: SamplingTool | None,
) -> SamplingResult[ResultT] | SamplingResult[str]:
"""Execute sampling with fallback handler, including automatic tool loop."""
assert self.fastmcp.sampling_handler is not None
current_messages = list(sampling_messages)
iteration = 0
has_tools = bool(sdk_tools)
has_result_type = result_type is not None and result_type is not str
while True:
# On last iteration, force completion
effective_tool_choice: ToolChoice | None = (
ToolChoice(mode=tool_choice) if tool_choice else None
)
if iteration == max_iterations - 1:
if has_result_type and final_response_tool is not None:
# Force final_response tool on last iteration
effective_tool_choice = ToolChoice(
mode="required", name="final_response"
)
elif has_tools:
# Force text response (no tools) on last iteration
effective_tool_choice = ToolChoice(mode="none")
create_message_result = self.fastmcp.sampling_handler(
current_messages,
SamplingParams(
systemPrompt=system_prompt,
messages=current_messages,
temperature=temperature,
maxTokens=max_tokens,
modelPreferences=_parse_model_preferences(model_preferences),
tools=sdk_tools,
toolChoice=effective_tool_choice,
),
self.request_context,
)
if inspect.isawaitable(create_message_result):
create_message_result = await create_message_result
iteration += 1
# Handle simple string/content responses
if isinstance(create_message_result, str):
if has_tools:
raise ValueError(
"Sampling handler returned a string, but tools were provided. "
"Handler must return CreateMessageResultWithTools when tools are used."
)
return SamplingResult(
text=create_message_result,
result=cast(ResultT, create_message_result),
history=current_messages,
)
if isinstance(create_message_result, CreateMessageResult):
if has_tools:
raise ValueError(
"Sampling handler returned CreateMessageResult, but tools were provided. "
"Handler must return CreateMessageResultWithTools when tools are used."
)
text = _extract_text_from_content(create_message_result.content)
return SamplingResult(
text=text,
result=cast(ResultT, text),
history=current_messages,
)
if isinstance(create_message_result, CreateMessageResultWithTools):
# If not a tool use, this is the final response
if create_message_result.stopReason != "toolUse" or not tool_map:
text = _extract_text_from_content(create_message_result.content)
return SamplingResult(
text=text,
result=cast(ResultT, text),
history=current_messages,
)
# Check if we've hit max iterations
if iteration >= max_iterations:
# Return what we have with None text (tool use without completion)
return SamplingResult(
text=None,
result=cast(ResultT, None),
history=current_messages,
)
# Execute tool calls
result = await self._execute_tool_calls(
result=create_message_result,
current_messages=current_messages,
tool_map=tool_map,
result_type=result_type,
final_response_tool=final_response_tool,
)
if isinstance(result, SamplingResult):
# final_response was called, return the structured result
return result
# result is a tuple of (messages, should_continue)
current_messages = cast(list[SamplingMessage], result[0])
should_continue = result[1]
if not should_continue:
text = _extract_text_from_content(create_message_result.content)
return SamplingResult(
text=text,
result=cast(ResultT, text),
history=current_messages,
)
continue
raise ValueError(
f"Unexpected sampling handler result: {create_message_result}"
)
async def _sample_with_client(
self,
sampling_messages: list[SamplingMessage],
system_prompt: str | None,
include_context: IncludeContext | None,
temperature: float | None,
max_tokens: int,
model_preferences: ModelPreferences | str | list[str] | None,
sdk_tools: list[SDKTool] | None,
tool_choice: ToolChoiceOption | None,
tool_map: dict[str, SamplingTool],
max_iterations: int,
result_type: type[ResultT] | None,
final_response_tool: SamplingTool | None,
) -> SamplingResult[ResultT] | SamplingResult[str]:
"""Execute sampling with client, including automatic tool loop."""
current_messages = list(sampling_messages)
iteration = 0
has_tools = bool(sdk_tools)
has_result_type = result_type is not None and result_type is not str
while True:
# On last iteration, force completion
effective_tool_choice: ToolChoice | None = (
ToolChoice(mode=tool_choice) if tool_choice else None
)
if iteration == max_iterations - 1:
if has_result_type and final_response_tool is not None:
# Force final_response tool on last iteration
effective_tool_choice = ToolChoice(
mode="required", name="final_response"
)
elif has_tools:
# Force text response (no tools) on last iteration
effective_tool_choice = ToolChoice(mode="none")
result = await self.session.create_message(
messages=current_messages,
system_prompt=system_prompt,
include_context=include_context,
temperature=temperature,
max_tokens=max_tokens,
model_preferences=_parse_model_preferences(model_preferences),
tools=sdk_tools,
tool_choice=effective_tool_choice,
related_request_id=self.request_id,
)
iteration += 1
# If not a tool use or no tools to execute, return text result
if result.stopReason != "toolUse" or not tool_map:
text = _extract_text_from_content(result.content)
return SamplingResult(
text=text,
result=cast(ResultT, text),
history=current_messages,
)
# Check if we've hit max iterations
if iteration >= max_iterations:
# Return what we have with None text (tool use without completion)
return SamplingResult(
text=None,
result=cast(ResultT, None),
history=current_messages,
)
# Execute tool calls
tool_result = await self._execute_tool_calls(
result=result,
current_messages=current_messages,
tool_map=tool_map,
result_type=result_type,
final_response_tool=final_response_tool,
)
if isinstance(tool_result, SamplingResult):
# final_response was called, return the structured result
return tool_result
# tool_result is a tuple of (messages, should_continue)
current_messages = cast(list[SamplingMessage], tool_result[0])
should_continue = tool_result[1]
if not should_continue:
text = _extract_text_from_content(result.content)
return SamplingResult(
text=text,
result=cast(ResultT, text),
history=current_messages,
)
async def _execute_tool_calls(
self,
result: CreateMessageResultWithTools,
current_messages: list[SamplingMessage],
tool_map: dict[str, SamplingTool],
result_type: type[ResultT] | None,
final_response_tool: SamplingTool | None,
) -> (
tuple[list[SamplingMessage], bool]
| SamplingResult[ResultT]
| SamplingResult[str]
):
"""Execute tool calls from an LLM response and return updated messages.
Returns:
- Tuple of (updated_messages, should_continue) for normal tool calls
- SamplingResult when final_response tool is called with valid input
"""
# Find all tool use content blocks
content_list = (
result.content if isinstance(result.content, list) else [result.content]
)
tool_use_blocks = [c for c in content_list if isinstance(c, ToolUseContent)]
if not tool_use_blocks:
return current_messages, False
# Add the assistant's response to messages
current_messages.append(
SamplingMessage(role="assistant", content=result.content)
)
# Execute each tool and collect results into a single list
tool_results: list[ToolResultContent] = []
for tool_use in tool_use_blocks:
# Check if this is the final_response tool
if (
tool_use.name == "final_response"
and final_response_tool is not None
and result_type is not None
and result_type is not str
):
# Validate and parse the input as the result type
try:
type_adapter = get_cached_typeadapter(result_type)
validated_result = type_adapter.validate_python(tool_use.input)
# Convert to JSON for .text
text = json.dumps(
type_adapter.dump_python(validated_result, mode="json")
)
return SamplingResult(
text=text,
result=validated_result,
history=current_messages,
)
except Exception as e:
# Validation failed - add error and continue the loop
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=tool_use.id,
content=[
TextContent(
type="text",
text=f"Validation error: {e}. Please try again.",
)
],
isError=True,
)
)
continue
tool = tool_map.get(tool_use.name)
if tool is None:
# Tool not found - add error result
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 Exception as e:
tool_results.append(
ToolResultContent(
type="tool_result",
toolUseId=tool_use.id,
content=[TextContent(type="text", text=f"Error: {e}")],
isError=True,
)
)
# Add all tool results as a single user message with list content
if tool_results:
current_messages.append(
SamplingMessage(
role="user",
content=tool_results, # type: ignore[arg-type]
)
)
return current_messages, True
@overload
async def elicit(
@ -604,11 +1116,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
):
@ -668,6 +1181,25 @@ class Context:
if result.action == "accept":
if response_type is not None:
# Handle dict-based enum responses (direct value extraction)
if isinstance(response_type, dict):
# Single-select: result.content is {"value": "selected_value"}
value = result.content.get("value") if result.content else None
return AcceptedElicitation[str](data=cast(str, value))
elif isinstance(response_type, list) and len(response_type) == 1:
if isinstance(response_type[0], dict):
# Multi-select titled: result.content is {"value": ["selected1", "selected2"]}
value = result.content.get("value") if result.content else None
return AcceptedElicitation[list[str]](
data=cast(list[str], value)
)
elif isinstance(response_type[0], list):
# Multi-select untitled: result.content is {"value": ["selected1", "selected2"]}
value = result.content.get("value") if result.content else None
return AcceptedElicitation[list[str]](
data=cast(list[str], value)
)
type_adapter = get_cached_typeadapter(response_type)
validated_data = cast(
T | ScalarElicitationType[T],
@ -798,3 +1330,46 @@ 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)
# 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 hasattr(block, "text"):
return block.text
return None
elif hasattr(content, "text"):
return content.text
return None

View file

@ -555,15 +555,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",

View file

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

View file

@ -0,0 +1,391 @@
"""Anthropic sampling handler for FastMCP servers."""
from collections.abc import Awaitable, Callable, Iterator, Sequence
from typing import Any
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 anthropic import Anthropic, NotGiven
from anthropic._types import NOT_GIVEN
from anthropic.types import (
Message,
MessageParam,
TextBlock,
TextBlockParam,
ToolParam,
ToolResultBlockParam,
ToolUseBlock,
ToolUseBlockParam,
)
from anthropic.types.model_param import ModelParam
from anthropic.types.tool_choice_param import ToolChoiceParam
except ImportError as e:
raise ImportError(
"The `anthropic` package is not installed. "
"Please install `fastmcp[anthropic]` or add `anthropic` to your dependencies manually."
) from e
SamplingHandlerResult = str | CreateMessageResult | CreateMessageResultWithTools
ServerSamplingHandler = Callable[
[
list[SamplingMessage],
SamplingParams,
RequestContext[ServerSession, LifespanContextT],
],
SamplingHandlerResult | Awaitable[SamplingHandlerResult],
]
class AnthropicSamplingHandler:
"""Sampling handler that uses the Anthropic API."""
def __init__(self, default_model: ModelParam, client: Anthropic | None = None):
self.client: Anthropic = client or Anthropic()
self.default_model: ModelParam = default_model
async def __call__(
self,
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext[ServerSession, LifespanContextT]
| RequestContext[ClientSession, LifespanContextT],
) -> CreateMessageResult | CreateMessageResultWithTools:
anthropic_messages: list[MessageParam] = self._convert_to_anthropic_messages(
messages=messages,
)
model: ModelParam = self._select_model_from_preferences(params.modelPreferences)
# Convert MCP tools to Anthropic format
anthropic_tools: list[ToolParam] | NotGiven = NOT_GIVEN
if params.tools:
anthropic_tools = self._convert_tools_to_anthropic(params.tools)
# Convert tool_choice to Anthropic format
anthropic_tool_choice: ToolChoiceParam | NotGiven = NOT_GIVEN
if params.toolChoice:
anthropic_tool_choice = self._convert_tool_choice_to_anthropic(
params.toolChoice
)
response = self.client.messages.create(
model=model,
messages=anthropic_messages,
system=params.systemPrompt or NOT_GIVEN,
temperature=params.temperature or NOT_GIVEN,
max_tokens=params.maxTokens,
stop_sequences=params.stopSequences or NOT_GIVEN,
tools=anthropic_tools,
tool_choice=anthropic_tool_choice,
)
# Return appropriate result type based on whether tools were provided
if params.tools:
return self._message_to_result_with_tools(response)
return self._message_to_create_message_result(response)
@staticmethod
def _iter_models_from_preferences(
model_preferences: ModelPreferences | str | list[str] | None,
) -> Iterator[str]:
if model_preferences is None:
return
if isinstance(model_preferences, str):
yield model_preferences
if isinstance(model_preferences, list):
yield from model_preferences
if isinstance(model_preferences, ModelPreferences):
if not (hints := model_preferences.hints):
return
for hint in hints:
if not (name := hint.name):
continue
yield name
@staticmethod
def _convert_to_anthropic_messages(
messages: Sequence[SamplingMessage],
) -> list[MessageParam]:
anthropic_messages: list[MessageParam] = []
if isinstance(messages, str):
anthropic_messages.append(
MessageParam(
role="user",
content=messages,
)
)
if isinstance(messages, list):
for message in messages:
if isinstance(message, str):
anthropic_messages.append(
MessageParam(
role="user",
content=message,
)
)
continue
content = message.content
# Handle list content (from CreateMessageResultWithTools)
if isinstance(content, list):
content_blocks: list[
TextBlockParam | ToolUseBlockParam | ToolResultBlockParam
] = []
for item in content:
if isinstance(item, ToolUseContent):
content_blocks.append(
ToolUseBlockParam(
type="tool_use",
id=item.id,
name=item.name,
input=item.input,
)
)
elif isinstance(item, TextContent):
content_blocks.append(
TextBlockParam(type="text", text=item.text)
)
elif isinstance(item, ToolResultContent):
# Extract text content from the result
result_content: str | list[TextBlockParam] = ""
if item.content:
text_blocks: list[TextBlockParam] = []
for sub_item in item.content:
if isinstance(sub_item, TextContent):
text_blocks.append(
TextBlockParam(
type="text", text=sub_item.text
)
)
if len(text_blocks) == 1:
result_content = text_blocks[0]["text"]
elif text_blocks:
result_content = text_blocks
content_blocks.append(
ToolResultBlockParam(
type="tool_result",
tool_use_id=item.toolUseId,
content=result_content,
)
)
if content_blocks:
anthropic_messages.append(
MessageParam(
role=message.role,
content=content_blocks, # type: ignore[arg-type]
)
)
continue
# Handle ToolUseContent (assistant's tool calls)
if isinstance(content, ToolUseContent):
anthropic_messages.append(
MessageParam(
role="assistant",
content=[
ToolUseBlockParam(
type="tool_use",
id=content.id,
name=content.name,
input=content.input,
)
],
)
)
continue
# Handle ToolResultContent (user's tool results)
if isinstance(content, ToolResultContent):
result_content_str: str | list[TextBlockParam] = ""
if content.content:
text_parts: list[TextBlockParam] = []
for item in content.content:
if isinstance(item, TextContent):
text_parts.append(
TextBlockParam(type="text", text=item.text)
)
if len(text_parts) == 1:
result_content_str = text_parts[0]["text"]
elif text_parts:
result_content_str = text_parts
anthropic_messages.append(
MessageParam(
role="user",
content=[
ToolResultBlockParam(
type="tool_result",
tool_use_id=content.toolUseId,
content=result_content_str,
)
],
)
)
continue
# Handle TextContent
if isinstance(content, TextContent):
anthropic_messages.append(
MessageParam(
role=message.role,
content=content.text,
)
)
continue
raise ValueError(f"Unsupported content type: {type(content)}")
return anthropic_messages
@staticmethod
def _message_to_create_message_result(
message: Message,
) -> CreateMessageResult:
if len(message.content) == 0:
raise ValueError("No content in response from Anthropic")
first_block = message.content[0]
if isinstance(first_block, TextBlock):
return CreateMessageResult(
content=TextContent(type="text", text=first_block.text),
role="assistant",
model=message.model,
)
raise ValueError(f"Unexpected content type in response: {type(first_block)}")
def _select_model_from_preferences(
self, model_preferences: ModelPreferences | str | list[str] | None
) -> ModelParam:
# Anthropic model names to check against
known_models = {
"claude-3-opus-20240229",
"claude-3-sonnet-20240229",
"claude-3-haiku-20240307",
"claude-3-5-sonnet-20240620",
"claude-3-5-sonnet-20241022",
"claude-3-5-haiku-20241022",
"claude-sonnet-4-5-20250929",
"claude-opus-4-5-20251101",
}
for model_option in self._iter_models_from_preferences(model_preferences):
# Accept any model that starts with "claude" or is in known models
if model_option.startswith("claude") or model_option in known_models:
return model_option
return self.default_model
@staticmethod
def _convert_tools_to_anthropic(tools: list[Tool]) -> list[ToolParam]:
"""Convert MCP tools to Anthropic tool format."""
anthropic_tools: list[ToolParam] = []
for tool in tools:
# Build input_schema dict, ensuring required fields
input_schema: dict[str, Any] = dict(tool.inputSchema)
if "type" not in input_schema:
input_schema["type"] = "object"
anthropic_tools.append(
ToolParam(
name=tool.name,
description=tool.description or "",
input_schema=input_schema, # type: ignore[arg-type]
)
)
return anthropic_tools
@staticmethod
def _convert_tool_choice_to_anthropic(
tool_choice: ToolChoice,
) -> ToolChoiceParam:
"""Convert MCP tool_choice to Anthropic format."""
if tool_choice.mode == "auto":
return {"type": "auto"}
elif tool_choice.mode == "required":
return {"type": "any"}
elif tool_choice.mode == "none":
# Anthropic doesn't have a "none" option, use auto
return {"type": "auto"}
else:
return {"type": "auto"}
@staticmethod
def _message_to_result_with_tools(
message: Message,
) -> CreateMessageResultWithTools:
"""Convert Anthropic response to CreateMessageResultWithTools."""
if len(message.content) == 0:
raise ValueError("No content in response from Anthropic")
# Determine stop reason
stop_reason: StopReason
if message.stop_reason == "tool_use":
stop_reason = "toolUse"
elif message.stop_reason == "end_turn":
stop_reason = "endTurn"
elif message.stop_reason == "max_tokens":
stop_reason = "maxTokens"
elif message.stop_reason == "stop_sequence":
stop_reason = "endTurn"
else:
stop_reason = "endTurn"
# Build content list
content: list[TextContent | ToolUseContent] = []
for block in message.content:
if isinstance(block, TextBlock):
content.append(TextContent(type="text", text=block.text))
elif isinstance(block, ToolUseBlock):
# Anthropic returns input as dict directly
arguments = block.input if isinstance(block.input, dict) else {}
content.append(
ToolUseContent(
type="tool_use",
id=block.id,
name=block.name,
input=arguments,
)
)
# Must have at least some content
if not content:
raise ValueError("No content in response from Anthropic")
return CreateMessageResultWithTools(
content=content,
role="assistant",
model=message.model,
stopReason=stop_reason,
)

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],
]

View file

@ -0,0 +1,409 @@
"""OpenAI sampling handler for FastMCP servers."""
import json
from collections.abc import Awaitable, Callable, Iterator, Sequence
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, NotGiven, OpenAI
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."
) from e
SamplingHandlerResult = str | CreateMessageResult | CreateMessageResultWithTools
ServerSamplingHandler = Callable[
[
list[SamplingMessage],
SamplingParams,
RequestContext[ServerSession, LifespanContextT],
],
SamplingHandlerResult | Awaitable[SamplingHandlerResult],
]
class OpenAISamplingHandler:
"""Sampling handler that uses the OpenAI API."""
def __init__(self, default_model: ChatModel, client: OpenAI | None = None):
self.client: OpenAI = client or OpenAI()
self.default_model: ChatModel = default_model
async def __call__(
self,
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext[ServerSession, LifespanContextT]
| RequestContext[ClientSession, LifespanContextT],
) -> CreateMessageResult | CreateMessageResultWithTools:
openai_messages: list[ChatCompletionMessageParam] = (
self._convert_to_openai_messages(
system_prompt=params.systemPrompt,
messages=messages,
)
)
model: ChatModel = self._select_model_from_preferences(params.modelPreferences)
# 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 = self.client.chat.completions.create(
model=model,
messages=openai_messages,
temperature=params.temperature or NOT_GIVEN,
max_tokens=params.maxTokens,
stop=params.stopSequences or 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
def _iter_models_from_preferences(
model_preferences: ModelPreferences | str | list[str] | None,
) -> Iterator[str]:
if model_preferences is None:
return
if isinstance(model_preferences, str) and model_preferences in get_args(
ChatModel
):
yield model_preferences
if isinstance(model_preferences, list):
yield from model_preferences
if isinstance(model_preferences, ModelPreferences):
if not (hints := model_preferences.hints):
return
for hint in hints:
if not (name := hint.name):
continue
yield name
@staticmethod
def _convert_to_openai_messages(
system_prompt: str | None, messages: Sequence[SamplingMessage]
) -> list[ChatCompletionMessageParam]:
openai_messages: list[ChatCompletionMessageParam] = []
if system_prompt:
openai_messages.append(
ChatCompletionSystemMessageParam(
role="system",
content=system_prompt,
)
)
if isinstance(messages, str):
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=messages,
)
)
if isinstance(messages, list):
for message in messages:
if isinstance(message, str):
openai_messages.append(
ChatCompletionUserMessageParam(
role="user",
content=message,
)
)
continue
content = message.content
# 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] = []
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):
# Each tool result becomes a separate tool message
content_parts: list[dict[str, str]] = []
if item.content:
for sub_item in item.content:
if isinstance(sub_item, TextContent):
content_parts.append(
{"type": "text", "text": sub_item.text}
)
openai_messages.append(
ChatCompletionToolMessageParam(
role="tool",
tool_call_id=item.toolUseId,
content=content_parts if content_parts else "",
)
)
# Add assistant message with tool calls if present
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,
)
)
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,
)
)
continue
# Handle ToolUseContent (assistant's tool calls)
if isinstance(content, ToolUseContent):
openai_messages.append(
ChatCompletionAssistantMessageParam(
role="assistant",
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_parts: list[dict[str, str]] = []
if content.content:
for item in content.content:
if isinstance(item, TextContent):
result_parts.append({"type": "text", "text": item.text})
openai_messages.append(
ChatCompletionToolMessageParam(
role="tool",
tool_call_id=content.toolUseId,
content=result_parts if result_parts else "",
)
)
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
@staticmethod
def _chat_completion_to_create_message_result(
chat_completion: ChatCompletion,
) -> CreateMessageResult:
if len(chat_completion.choices) == 0:
raise ValueError("No response for completion")
first_choice = chat_completion.choices[0]
if content := first_choice.message.content:
return CreateMessageResult(
content=TextContent(type="text", text=content),
role="assistant",
model=chat_completion.model,
)
raise ValueError("No content in response from completion")
def _select_model_from_preferences(
self, model_preferences: ModelPreferences | str | list[str] | None
) -> ChatModel:
for model_option in self._iter_models_from_preferences(model_preferences):
if model_option in get_args(ChatModel):
chosen_model: ChatModel = model_option # pyright: ignore[reportAssignmentType]
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:
# Unknown mode, default to auto
return "auto"
@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:
# Parse the arguments JSON string
try:
arguments = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
arguments = {}
content.append(
ToolUseContent(
type="tool_use",
id=tool_call.id,
name=tool_call.function.name,
input=arguments,
)
)
# Must have at least some content
if not content:
raise ValueError("No content in response from completion")
return CreateMessageResultWithTools(
content=content,
role="assistant",
model=chat_completion.model,
stopReason=stop_reason,
)

View file

@ -0,0 +1,141 @@
"""SamplingTool for use during LLM sampling requests."""
from __future__ import annotations
import inspect
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from mcp.types import Tool as SDKTool
from pydantic import BaseModel, ConfigDict
from fastmcp.tools.tool import ParsedFunction
if TYPE_CHECKING:
from fastmcp.tools.tool import Tool
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")
Or from an existing FastMCP Tool:
sampling_tool = SamplingTool.from_mcp_tool(existing_tool)
"""
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_mcp_tool(cls, tool: Tool) -> SamplingTool:
"""Create a SamplingTool from a FastMCP Tool.
Args:
tool: A FastMCP Tool instance (must have an fn attribute).
Returns:
A SamplingTool with the same schema and executor.
Raises:
ValueError: If the tool doesn't have an fn attribute.
"""
if not hasattr(tool, "fn"):
raise ValueError(
f"Tool {tool.name!r} does not have an fn attribute. "
"Only FunctionTools can be converted to SamplingTools."
)
return cls(
name=tool.name,
description=tool.description,
parameters=tool.parameters,
fn=tool.fn,
)
@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,
)

View file

@ -42,10 +42,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
@ -953,7 +953,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.
@ -1037,7 +1037,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.
@ -1126,7 +1126,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.
@ -1220,7 +1220,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.

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,775 @@ 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_must_return_correct_type(self):
"""Test that fallback handler must return CreateMessageResultWithTools when tools provided."""
from fastmcp.exceptions import ToolError
# This handler returns a string, which is invalid when tools are provided
invalid_handler = AsyncMock(return_value="Fallback response")
mcp = FastMCP(sampling_handler=invalid_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 str(result)
# Client without sampling handler - will use server's fallback
async with Client(mcp) as client:
# Handler returns string but tools were provided - should error
with pytest.raises(
ToolError, match="must return CreateMessageResultWithTools"
):
await client.call_tool("sample_with_tool", {})
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_sampling_tool_from_mcp_tool(self):
"""Test creating SamplingTool from FastMCP Tool."""
from fastmcp.tools.tool import Tool
def original_fn(x: int, y: int) -> int:
"""Add x and y."""
return x + y
mcp_tool = Tool.from_function(original_fn)
sampling = SamplingTool.from_mcp_tool(mcp_tool)
assert sampling.name == "original_fn"
assert sampling.description == "Add x and y."
assert sampling.fn is mcp_tool.fn
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_max_iterations(self):
"""Test that max_iterations prevents infinite loops.
On the last iteration, tool_choice is set to 'none' to force text response.
"""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
call_count = 0
received_tool_choices: list = []
def looping_tool() -> str:
"""A tool that always gets called again."""
return "keep going"
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
nonlocal call_count
call_count += 1
received_tool_choices.append(params.toolChoice)
# On last iteration (when tool_choice=none), return text
if params.toolChoice and params.toolChoice.mode == "none":
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Forced to stop")],
model="test-model",
stopReason="endTurn",
)
# Otherwise keep returning tool use
return CreateMessageResultWithTools(
role="assistant",
content=[
ToolUseContent(
type="tool_use",
id=f"call_{call_count}",
name="looping_tool",
input={},
)
],
model="test-model",
stopReason="toolUse",
)
mcp = FastMCP(sampling_handler=sampling_handler)
@mcp.tool
async def infinite_loop(context: Context) -> str:
result = await context.sample(
messages="Start", tools=[looping_tool], max_iterations=3
)
return result.text or "no text"
async with Client(mcp) as client:
result = await client.call_tool("infinite_loop", {})
# Should complete after 3 iterations with forced text response
assert call_count == 3
assert result.data == "Forced to stop"
# Last call should have tool_choice=none
assert received_tool_choices[-1].mode == "none"
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"
async def test_max_iterations_one_for_manual_loop(self):
"""Test that max_iterations=1 forces text on first call for manual loop building.
With max_iterations=1, tool_choice is set to 'none' on the first (and only) call,
forcing a text response so users can build their own loop using history.
"""
from mcp.types import CreateMessageResultWithTools, ToolUseContent
received_tool_choices: list = []
def my_tool() -> str:
"""A tool."""
return "result"
def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
) -> CreateMessageResultWithTools:
received_tool_choices.append(params.toolChoice)
# With tool_choice=none, return text response
if params.toolChoice and params.toolChoice.mode == "none":
return CreateMessageResultWithTools(
role="assistant",
content=[TextContent(type="text", text="Forced text response")],
model="test-model",
stopReason="endTurn",
)
# Otherwise return tool use (shouldn't happen with max_iterations=1)
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 single_iteration(context: Context) -> str:
result = await context.sample(
messages="Test", tools=[my_tool], max_iterations=1
)
return result.text or "no text"
async with Client(mcp) as client:
result = await client.call_tool("single_iteration", {})
# Should get forced text response
assert result.data == "Forced text response"
# First and only call should have tool_choice=none
assert len(received_tool_choices) == 1
assert received_tool_choices[0].mode == "none"
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"

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

View file

@ -0,0 +1,238 @@
from unittest.mock import MagicMock
import pytest
from anthropic import Anthropic
from anthropic.types import Message, TextBlock, ToolUseBlock, Usage
from mcp.types import (
CreateMessageResult,
CreateMessageResultWithTools,
ModelHint,
ModelPreferences,
SamplingMessage,
TextContent,
ToolUseContent,
)
from fastmcp.server.sampling.anthropic import AnthropicSamplingHandler
def test_convert_sampling_messages_to_anthropic_messages():
msgs = AnthropicSamplingHandler._convert_to_anthropic_messages(
messages=[
SamplingMessage(
role="user", content=TextContent(type="text", text="hello")
),
SamplingMessage(
role="assistant", content=TextContent(type="text", text="ok")
),
],
)
assert msgs == [
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "ok"},
]
def test_convert_to_anthropic_messages_raises_on_non_text():
from fastmcp.utilities.types import Image
with pytest.raises(ValueError):
AnthropicSamplingHandler._convert_to_anthropic_messages(
messages=[
SamplingMessage(
role="user",
content=Image(data=b"abc").to_image_content(),
)
],
)
@pytest.mark.parametrize(
"prefs,expected",
[
("claude-3-5-sonnet-20241022", "claude-3-5-sonnet-20241022"),
(
ModelPreferences(hints=[ModelHint(name="claude-3-5-sonnet-20241022")]),
"claude-3-5-sonnet-20241022",
),
(["claude-3-5-sonnet-20241022", "other"], "claude-3-5-sonnet-20241022"),
(None, "fallback-model"),
(["unknown-model"], "fallback-model"),
],
)
def test_select_model_from_preferences(prefs, expected):
mock_client = MagicMock(spec=Anthropic)
handler = AnthropicSamplingHandler(
default_model="fallback-model", client=mock_client
)
assert handler._select_model_from_preferences(prefs) == expected
def test_message_to_create_message_result():
mock_client = MagicMock(spec=Anthropic)
handler = AnthropicSamplingHandler(
default_model="fallback-model", client=mock_client
)
message = Message(
id="msg_123",
type="message",
role="assistant",
content=[TextBlock(type="text", text="HELPFUL CONTENT FROM A VERY SMART LLM")],
model="claude-3-5-sonnet-20241022",
stop_reason="end_turn",
stop_sequence=None,
usage=Usage(input_tokens=10, output_tokens=20),
)
result: CreateMessageResult = handler._message_to_create_message_result(message)
assert result == CreateMessageResult(
content=TextContent(type="text", text="HELPFUL CONTENT FROM A VERY SMART LLM"),
role="assistant",
model="claude-3-5-sonnet-20241022",
)
def test_message_to_result_with_tools():
message = Message(
id="msg_123",
type="message",
role="assistant",
content=[
TextBlock(type="text", text="I'll help you with that."),
ToolUseBlock(
type="tool_use",
id="toolu_123",
name="get_weather",
input={"location": "San Francisco"},
),
],
model="claude-3-5-sonnet-20241022",
stop_reason="tool_use",
stop_sequence=None,
usage=Usage(input_tokens=10, output_tokens=20),
)
result: CreateMessageResultWithTools = (
AnthropicSamplingHandler._message_to_result_with_tools(message)
)
assert result.role == "assistant"
assert result.model == "claude-3-5-sonnet-20241022"
assert result.stopReason == "toolUse"
assert len(result.content) == 2
assert result.content[0] == TextContent(
type="text", text="I'll help you with that."
)
assert result.content[1] == ToolUseContent(
type="tool_use",
id="toolu_123",
name="get_weather",
input={"location": "San Francisco"},
)
def test_convert_tool_choice_auto():
result = AnthropicSamplingHandler._convert_tool_choice_to_anthropic(
MagicMock(mode="auto")
)
assert result == {"type": "auto"}
def test_convert_tool_choice_required():
result = AnthropicSamplingHandler._convert_tool_choice_to_anthropic(
MagicMock(mode="required")
)
assert result == {"type": "any"}
def test_convert_tool_choice_none():
result = AnthropicSamplingHandler._convert_tool_choice_to_anthropic(
MagicMock(mode="none")
)
# Anthropic doesn't have "none", falls back to "auto"
assert result == {"type": "auto"}
def test_convert_tools_to_anthropic():
from mcp.types import Tool
tools = [
Tool(
name="get_weather",
description="Get the current weather",
inputSchema={
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"],
},
)
]
result = AnthropicSamplingHandler._convert_tools_to_anthropic(tools)
assert len(result) == 1
assert result[0]["name"] == "get_weather"
assert result[0]["description"] == "Get the current weather"
assert result[0]["input_schema"] == {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"],
}
def test_convert_messages_with_tool_use_content():
"""Test converting messages that include tool use content from assistant."""
msgs = AnthropicSamplingHandler._convert_to_anthropic_messages(
messages=[
SamplingMessage(
role="assistant",
content=ToolUseContent(
type="tool_use",
id="toolu_123",
name="get_weather",
input={"location": "NYC"},
),
),
],
)
assert len(msgs) == 1
assert msgs[0]["role"] == "assistant"
assert msgs[0]["content"] == [
{
"type": "tool_use",
"id": "toolu_123",
"name": "get_weather",
"input": {"location": "NYC"},
}
]
def test_convert_messages_with_tool_result_content():
"""Test converting messages that include tool result content from user."""
from mcp.types import ToolResultContent
msgs = AnthropicSamplingHandler._convert_to_anthropic_messages(
messages=[
SamplingMessage(
role="user",
content=ToolResultContent(
type="tool_result",
toolUseId="toolu_123",
content=[TextContent(type="text", text="72F and sunny")],
),
),
],
)
assert len(msgs) == 1
assert msgs[0]["role"] == "user"
assert msgs[0]["content"] == [
{
"type": "tool_result",
"tool_use_id": "toolu_123",
"content": "72F and sunny",
}
]

View file

@ -18,7 +18,7 @@ from openai.types.chat import (
)
from openai.types.chat.chat_completion import Choice
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
from fastmcp.server.sampling.openai import OpenAISamplingHandler
def test_convert_sampling_messages_to_openai_messages():

View file

@ -0,0 +1,150 @@
"""Tests for SamplingTool."""
import pytest
from fastmcp.server.sampling import SamplingTool
from fastmcp.tools.tool import Tool
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 TestSamplingToolFromMCPTool:
"""Tests for SamplingTool.from_mcp_tool()."""
def test_from_function_tool(self):
def original(x: int) -> int:
"""Double a number."""
return x * 2
mcp_tool = Tool.from_function(original)
sampling = SamplingTool.from_mcp_tool(mcp_tool)
assert sampling.name == "original"
assert sampling.description == "Double a number."
assert sampling.fn is mcp_tool.fn
def test_from_tool_without_fn_raises(self):
# Create a base Tool without fn (not a FunctionTool)
tool = Tool(
name="test",
description="Test tool",
parameters={"type": "object"},
tags=set(),
)
with pytest.raises(ValueError, match="does not have an fn attribute"):
SamplingTool.from_mcp_tool(tool)
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", {})

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=[])

1673
uv.lock generated

File diff suppressed because it is too large Load diff