From 577f4d1bdf71f61a4812ced218774e325a2a0c87 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 14 Dec 2025 20:47:00 -0500 Subject: [PATCH] Consolidate sampling examples and fix tool_choice bug (#2618) * 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 --- examples/advanced_sampling/README.md | 40 ----- .../advanced_sampling/client_sampling_test.py | 159 ------------------ .../advanced_sampling/structured_output.py | 68 -------- examples/advanced_sampling/tool_use.py | 84 --------- examples/sampling.py | 52 ------ examples/sampling/README.md | 62 +++++++ examples/sampling/server_fallback.py | 88 ++++++++++ examples/sampling/structured_output.py | 110 ++++++++++++ examples/sampling/text.py | 78 +++++++++ examples/sampling/tool_use.py | 125 ++++++++++++++ examples/sampling_fallback.py | 38 ----- src/fastmcp/server/context.py | 7 +- 12 files changed, 466 insertions(+), 445 deletions(-) delete mode 100644 examples/advanced_sampling/README.md delete mode 100644 examples/advanced_sampling/client_sampling_test.py delete mode 100644 examples/advanced_sampling/structured_output.py delete mode 100644 examples/advanced_sampling/tool_use.py delete mode 100644 examples/sampling.py create mode 100644 examples/sampling/README.md create mode 100644 examples/sampling/server_fallback.py create mode 100644 examples/sampling/structured_output.py create mode 100644 examples/sampling/text.py create mode 100644 examples/sampling/tool_use.py delete mode 100644 examples/sampling_fallback.py diff --git a/examples/advanced_sampling/README.md b/examples/advanced_sampling/README.md deleted file mode 100644 index 12b01ae19..000000000 --- a/examples/advanced_sampling/README.md +++ /dev/null @@ -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 -``` diff --git a/examples/advanced_sampling/client_sampling_test.py b/examples/advanced_sampling/client_sampling_test.py deleted file mode 100644 index 31c3b5312..000000000 --- a/examples/advanced_sampling/client_sampling_test.py +++ /dev/null @@ -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()) diff --git a/examples/advanced_sampling/structured_output.py b/examples/advanced_sampling/structured_output.py deleted file mode 100644 index 9fe31112b..000000000 --- a/examples/advanced_sampling/structured_output.py +++ /dev/null @@ -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()) diff --git a/examples/advanced_sampling/tool_use.py b/examples/advanced_sampling/tool_use.py deleted file mode 100644 index 3c6998dd5..000000000 --- a/examples/advanced_sampling/tool_use.py +++ /dev/null @@ -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()) diff --git a/examples/sampling.py b/examples/sampling.py deleted file mode 100644 index cfb9c395a..000000000 --- a/examples/sampling.py +++ /dev/null @@ -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()) diff --git a/examples/sampling/README.md b/examples/sampling/README.md new file mode 100644 index 000000000..3f9225b19 --- /dev/null +++ b/examples/sampling/README.md @@ -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]`. diff --git a/examples/sampling/server_fallback.py b/examples/sampling/server_fallback.py new file mode 100644 index 000000000..1c3fb10af --- /dev/null +++ b/examples/sampling/server_fallback.py @@ -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()) diff --git a/examples/sampling/structured_output.py b/examples/sampling/structured_output.py new file mode 100644 index 000000000..a866d6610 --- /dev/null +++ b/examples/sampling/structured_output.py @@ -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()) diff --git a/examples/sampling/text.py b/examples/sampling/text.py new file mode 100644 index 000000000..6d354e7c2 --- /dev/null +++ b/examples/sampling/text.py @@ -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()) diff --git a/examples/sampling/tool_use.py b/examples/sampling/tool_use.py new file mode 100644 index 000000000..e745feae8 --- /dev/null +++ b/examples/sampling/tool_use.py @@ -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()) diff --git a/examples/sampling_fallback.py b/examples/sampling_fallback.py deleted file mode 100644 index edabd2d9e..000000000 --- a/examples/sampling_fallback.py +++ /dev/null @@ -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()) diff --git a/src/fastmcp/server/context.py b/src/fastmcp/server/context.py index 2e5cea6a4..8cd4f7307 100644 --- a/src/fastmcp/server/context.py +++ b/src/fastmcp/server/context.py @@ -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