ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Consolidate sampling examples and fix tool_choice bug (#2618)

Browse source 
* Fix tool_choice to always require tools when result_type is set

* Consolidate sampling examples with rich output

* Replace eval() with explicit add/multiply tools
This commit is contained in:
Jeremiah Lowin 2025-12-14 20:47:00 -05:00 • committed by GitHub
commit 577f4d1bdf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 466 additions and 445 deletions
Download patch file Download diff file Expand all files Collapse all files

40
examples/advanced_sampling/README.md
View file

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

159
examples/advanced_sampling/client_sampling_test.py
View file

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

68
examples/advanced_sampling/structured_output.py
View file

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

84
examples/advanced_sampling/tool_use.py
View file

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

52
examples/sampling.py
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())

62
examples/sampling/README.md Normal file
View file

@ -0,0 +1,62 @@
# Sampling Examples
These examples demonstrate FastMCP's sampling API, which allows server tools to request LLM completions from the client.
## Prerequisites
```bash
pip install fastmcp[anthropic]
export ANTHROPIC_API_KEY=your-key
```
Or run directly with `uv`:
```bash
uv run examples/sampling/text.py
```
## Examples
### Simple Text Sampling (`text.py`)
Basic sampling flow where a server tool requests an LLM completion:
```bash
uv run examples/sampling/text.py
```
### Structured Output (`structured_output.py`)
Uses `result_type` to get validated Pydantic models from the LLM:
```bash
uv run examples/sampling/structured_output.py
```
### Tool Use (`tool_use.py`)
Gives the LLM tools to use during sampling (calculator, time, dice):
```bash
uv run examples/sampling/tool_use.py
```
### Server Fallback (`server_fallback.py`)
Configures a fallback sampling handler on the server, enabling sampling even when clients don't support it:
```bash
uv run examples/sampling/server_fallback.py
```
## Using OpenAI Instead
To use OpenAI instead of Anthropic, change the handler:
```python
from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
handler = OpenAISamplingHandler(default_model="gpt-4o-mini")
```
And install with `pip install fastmcp[openai]`.

88
examples/sampling/server_fallback.py Normal file
View file

@ -0,0 +1,88 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Server-Side Fallback Handler
Demonstrates configuring a sampling handler on the server. This ensures
sampling works even when the client doesn't provide a handler.
The server runs as an HTTP server that can be connected to by any MCP client.
Run:
uv run examples/sampling/server_fallback.py
Then connect with any MCP client (e.g., Claude Desktop) or test with:
curl http://localhost:8000/mcp/
"""
import asyncio
from rich.console import Console
from rich.panel import Panel
from fastmcp import FastMCP
from fastmcp.client.sampling.handlers.anthropic import AnthropicSamplingHandler
from fastmcp.server.context import Context
console = Console()
# Create server with a fallback sampling handler
# This handler is used when the client doesn't support sampling
mcp = FastMCP(
"Server with Fallback Handler",
sampling_handler=AnthropicSamplingHandler(default_model="claude-sonnet-4-5"),
sampling_handler_behavior="fallback", # Use only if client lacks sampling
)
@mcp.tool
async def summarize(text: str, ctx: Context) -> str:
"""Summarize the given text."""
console.print(f"[bold cyan]SERVER[/] Summarizing text ({len(text)} chars)...")
result = await ctx.sample(
messages=f"Summarize this text in 1-2 sentences:\n\n{text}",
system_prompt="You are a concise summarizer.",
max_tokens=150,
)
console.print("[bold cyan]SERVER[/] Summary complete")
return result.text or ""
@mcp.tool
async def translate(text: str, target_language: str, ctx: Context) -> str:
"""Translate text to the target language."""
console.print(f"[bold cyan]SERVER[/] Translating to {target_language}...")
result = await ctx.sample(
messages=f"Translate to {target_language}:\n\n{text}",
system_prompt=f"You are a translator. Output only the {target_language} translation.",
max_tokens=500,
)
console.print("[bold cyan]SERVER[/] Translation complete")
return result.text or ""
async def main():
console.print(
Panel.fit(
"[bold]Server-Side Fallback Handler Demo[/]\n\n"
"This server has a built-in Anthropic handler that activates\n"
"when clients don't provide their own sampling support.",
subtitle="server_fallback.py",
)
)
console.print()
console.print("[bold yellow]Starting HTTP server on http://localhost:8000[/]")
console.print("Connect with an MCP client or press Ctrl+C to stop")
console.print()
await mcp.run_http_async(host="localhost", port=8000)
if __name__ == "__main__":
asyncio.run(main())

110
examples/sampling/structured_output.py Normal file
View file

@ -0,0 +1,110 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Structured Output Sampling
Demonstrates using `result_type` to get validated Pydantic models from an LLM.
The server exposes a sentiment analysis tool that returns structured data.
Run:
uv run examples/sampling/structured_output.py
"""
import asyncio
from pydantic import BaseModel
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import SamplingMessage, SamplingParams
from fastmcp.client.sampling.handlers.anthropic import AnthropicSamplingHandler
console = Console()
class LoggingAnthropicHandler(AnthropicSamplingHandler):
async def __call__(
self, messages: list[SamplingMessage], params: SamplingParams, context
): # type: ignore[override]
console.print(" [bold blue]SAMPLING[/] Calling Claude API...")
result = await super().__call__(messages, params, context)
console.print(" [bold blue]SAMPLING[/] Response received")
return result
# Define a structured output model
class SentimentAnalysis(BaseModel):
sentiment: str # "positive", "negative", or "neutral"
confidence: float # 0.0 to 1.0
keywords: list[str] # Key words that influenced the analysis
explanation: str # Brief explanation of the analysis
# Create the MCP server
mcp = FastMCP("Sentiment Analyzer")
@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> dict:
"""Analyze the sentiment of the given text."""
console.print(" [bold cyan]SERVER[/] Analyzing sentiment...")
result = await ctx.sample(
messages=f"Analyze the sentiment of this text:\n\n{text}",
system_prompt="You are a sentiment analysis expert. Analyze text carefully.",
result_type=SentimentAnalysis,
)
console.print(" [bold cyan]SERVER[/] Analysis complete")
return result.result.model_dump() # type: ignore[attr-defined]
async def main():
console.print(
Panel.fit("[bold]MCP Sampling Flow Demo[/]", subtitle="structured_output.py")
)
console.print()
handler = LoggingAnthropicHandler(default_model="claude-sonnet-4-5")
async with Client(mcp, sampling_handler=handler) as client:
texts = [
"I absolutely love this product! It exceeded all my expectations.",
"The service was okay, nothing special but got the job done.",
"This is the worst experience I've ever had. Never again.",
]
for text in texts:
console.print(f"[bold green]CLIENT[/] Analyzing: [italic]{text[:50]}...[/]")
console.print()
result = await client.call_tool("analyze_sentiment", {"text": text})
data = result.data
# Display results in a table
table = Table(show_header=False, box=None, padding=(0, 2))
table.add_column(style="bold")
table.add_column()
sentiment_color = {
"positive": "green",
"negative": "red",
"neutral": "yellow",
}.get(
data["sentiment"],
"white", # type: ignore[union-attr]
)
table.add_row("Sentiment", f"[{sentiment_color}]{data['sentiment']}[/]") # type: ignore[index]
table.add_row("Confidence", f"{data['confidence']:.0%}") # type: ignore[index]
table.add_row("Keywords", ", ".join(data["keywords"])) # type: ignore[index]
table.add_row("Explanation", data["explanation"]) # type: ignore[index]
console.print(Panel(table, border_style=sentiment_color))
console.print()
if __name__ == "__main__":
asyncio.run(main())

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

@ -0,0 +1,78 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Simple Text Sampling
Demonstrates the basic MCP sampling flow where a server tool requests
an LLM completion from the client.
Run:
uv run examples/sampling/text.py
"""
import asyncio
from rich.console import Console
from rich.panel import Panel
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import SamplingMessage, SamplingParams
from fastmcp.client.sampling.handlers.anthropic import AnthropicSamplingHandler
console = Console()
# Create a wrapper handler that logs when the LLM is called
class LoggingAnthropicHandler(AnthropicSamplingHandler):
async def __call__(
self, messages: list[SamplingMessage], params: SamplingParams, context
): # type: ignore[override]
console.print(" [bold blue]SAMPLING[/] Calling Claude API...")
result = await super().__call__(messages, params, context)
console.print(" [bold blue]SAMPLING[/] Response received")
return result
# Create the MCP server
mcp = FastMCP("Haiku Generator")
@mcp.tool
async def write_haiku(topic: str, ctx: Context) -> str:
"""Write a haiku about any topic."""
console.print(
f" [bold cyan]SERVER[/] Tool 'write_haiku' called with topic: {topic}"
)
result = await ctx.sample(
messages=f"Write a haiku about: {topic}",
system_prompt="You are a poet. Write only the haiku, nothing else.",
max_tokens=100,
)
console.print(" [bold cyan]SERVER[/] Returning haiku to client")
return result.text or ""
async def main():
console.print(Panel.fit("[bold]MCP Sampling Flow Demo[/]", subtitle="text.py"))
console.print()
# Create the sampling handler
handler = LoggingAnthropicHandler(default_model="claude-sonnet-4-5")
# Connect client to server with the sampling handler
async with Client(mcp, sampling_handler=handler) as client:
console.print("[bold green]CLIENT[/] Calling tool 'write_haiku'...")
console.print()
result = await client.call_tool("write_haiku", {"topic": "Python programming"})
console.print()
console.print("[bold green]CLIENT[/] Received result:")
console.print(Panel(result.data, title="Haiku", border_style="green")) # type: ignore[arg-type]
if __name__ == "__main__":
asyncio.run(main())

125
examples/sampling/tool_use.py Normal file
View file

@ -0,0 +1,125 @@
# /// script
# dependencies = ["anthropic", "fastmcp", "rich"]
# ///
"""
Sampling with Tools
Demonstrates giving an LLM tools to use during sampling. The LLM can call
helper functions to gather information before responding.
Run:
uv run examples/sampling/tool_use.py
"""
import asyncio
import random
from datetime import datetime
from pydantic import BaseModel, Field
from rich.console import Console
from rich.panel import Panel
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import SamplingMessage, SamplingParams
from fastmcp.client.sampling.handlers.anthropic import AnthropicSamplingHandler
console = Console()
class LoggingAnthropicHandler(AnthropicSamplingHandler):
async def __call__(
self, messages: list[SamplingMessage], params: SamplingParams, context
): # type: ignore[override]
console.print(" [bold blue]SAMPLING[/] Calling Claude API...")
result = await super().__call__(messages, params, context)
console.print(" [bold blue]SAMPLING[/] Response received")
return result
# Define tools available to the LLM during sampling
def add(a: float, b: float) -> str:
"""Add two numbers together."""
result = a + b
console.print(f" [bold magenta]TOOL[/] add({a}, {b}) = {result}")
return str(result)
def multiply(a: float, b: float) -> str:
"""Multiply two numbers together."""
result = a * b
console.print(f" [bold magenta]TOOL[/] multiply({a}, {b}) = {result}")
return str(result)
def get_current_time() -> str:
"""Get the current date and time."""
console.print(" [bold magenta]TOOL[/] get_current_time()")
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def roll_dice(sides: int = 6) -> str:
"""Roll a dice with the specified number of sides."""
result = random.randint(1, sides)
console.print(f" [bold magenta]TOOL[/] roll_dice({sides}) = {result}")
return str(result)
# Structured output for the response
class AssistantResponse(BaseModel):
answer: str = Field(description="The answer to the user's question")
tools_used: list[str] = Field(description="List of tools that were used")
reasoning: str = Field(
description="Brief explanation of how the answer was determined"
)
# Create the MCP server
mcp = FastMCP("Smart Assistant")
@mcp.tool
async def ask_assistant(question: str, ctx: Context) -> dict:
"""Ask the assistant a question. It can use tools to help answer."""
console.print(" [bold cyan]SERVER[/] Processing question...")
result = await ctx.sample(
messages=question,
system_prompt="You are a helpful assistant with access to tools. Use them when needed to answer questions accurately.",
tools=[add, multiply, get_current_time, roll_dice],
result_type=AssistantResponse,
)
console.print(" [bold cyan]SERVER[/] Response ready")
return result.result.model_dump() # type: ignore[attr-defined]
async def main():
console.print(Panel.fit("[bold]MCP Sampling Flow Demo[/]", subtitle="tool_use.py"))
console.print()
handler = LoggingAnthropicHandler(default_model="claude-sonnet-4-5")
async with Client(mcp, sampling_handler=handler) as client:
questions = [
"What is 15 times 7, plus 23?",
"Roll a 20-sided dice for me",
"What time is it right now?",
]
for question in questions:
console.print(f"[bold green]CLIENT[/] Question: {question}")
console.print()
result = await client.call_tool("ask_assistant", {"question": question})
data = result.data
console.print(f"[bold green]CLIENT[/] Answer: {data['answer']}") # type: ignore[index]
console.print(
f" Tools used: {', '.join(data['tools_used']) or 'none'}"
) # type: ignore[index]
console.print(f" Reasoning: {data['reasoning']}") # type: ignore[index]
console.print()
if __name__ == "__main__":
asyncio.run(main())

38
examples/sampling_fallback.py
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.client.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())

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

@ -777,7 +777,6 @@ class Context:
# Convert tools to SamplingTools
sampling_tools = _prepare_tools(tools)
has_user_tools = bool(sampling_tools)
# Handle structured output with result_type
tool_choice: str | None = None
@ -786,9 +785,9 @@ class Context:
sampling_tools = list(sampling_tools) if sampling_tools else []
sampling_tools.append(final_response_tool)
# If no user tools, force the LLM to call a tool (which is final_response)
if not has_user_tools:
tool_choice = "required"
# Always require tool calls when result_type is set - the LLM must
# eventually call final_response (text responses are not accepted)
tool_choice = "required"
# Convert messages for the loop
current_messages: str | Sequence[str | SamplingMessage] = messages