mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 13:34:17 +02:00
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
This commit is contained in:
Jeremiah Lowin
parent
b544cddca7
commit
7e02fa29b2
14 changed files with 1297 additions and 281 deletions
|
|
@ -1,28 +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
|
||||
```
|
||||
|
|
@ -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.experimental.sampling.handlers.openai import OpenAISamplingHandler
|
||||
|
||||
|
||||
# Define a structured output model
|
||||
class SentimentAnalysis(BaseModel):
|
||||
sentiment: str # "positive", "negative", or "neutral"
|
||||
confidence: float # 0.0 to 1.0
|
||||
keywords: list[str] # Key words that influenced the analysis
|
||||
|
||||
|
||||
# Create an MCP server with a sampling tool
|
||||
mcp = FastMCP("Sentiment Analyzer")
|
||||
|
||||
|
||||
@mcp.tool
|
||||
async def analyze_sentiment(text: str, ctx: Context) -> dict:
|
||||
"""Analyze the sentiment of the given text."""
|
||||
result = await ctx.sample(
|
||||
messages=f"Analyze the sentiment of this text:\n\n{text}",
|
||||
system_prompt="You are a sentiment analysis expert. Analyze text and return structured results.",
|
||||
result_type=SentimentAnalysis,
|
||||
)
|
||||
# result.result is a validated SentimentAnalysis instance
|
||||
return result.result.model_dump() # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def main():
|
||||
# Create an OpenAI-backed sampling handler
|
||||
sampling_handler = OpenAISamplingHandler(default_model="gpt-4o-mini")
|
||||
|
||||
# Connect to the server with the sampling handler
|
||||
async with Client(mcp, sampling_handler=sampling_handler) as client:
|
||||
# Call the tool with some test text
|
||||
result = await client.call_tool(
|
||||
"analyze_sentiment",
|
||||
{
|
||||
"text": "I absolutely love this product! It exceeded all my expectations."
|
||||
},
|
||||
)
|
||||
|
||||
print("Analysis Result:")
|
||||
print(f" Sentiment: {result.data['sentiment']}")
|
||||
print(f" Confidence: {result.data['confidence']:.1%}")
|
||||
print(f" Keywords: {', '.join(result.data['keywords'])}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
|
@ -1,85 +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
|
||||
|
||||
from fastmcp import Client, Context, FastMCP
|
||||
from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
|
||||
|
||||
|
||||
# Define tools (available to the LLM during sampling)
|
||||
def search_web(query: str) -> str:
|
||||
"""Search the web for information."""
|
||||
# Simulated search results
|
||||
results = {
|
||||
"python async": "Python's asyncio provides async/await syntax for concurrent code.",
|
||||
"fastmcp": "FastMCP is a framework for building MCP servers and clients in Python.",
|
||||
"mcp protocol": "MCP (Model Context Protocol) enables AI models to use tools and resources.",
|
||||
}
|
||||
for key, value in results.items():
|
||||
if key in query.lower():
|
||||
return value
|
||||
return f"No results found for: {query}"
|
||||
|
||||
|
||||
def get_word_count(text: str) -> str:
|
||||
"""Count words in text."""
|
||||
return str(len(text.split()))
|
||||
|
||||
|
||||
# Structured output for the final response
|
||||
class ResearchReport(BaseModel):
|
||||
summary: str
|
||||
sources_used: list[str]
|
||||
confidence: float
|
||||
|
||||
|
||||
# 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,
|
||||
max_iterations=5,
|
||||
)
|
||||
|
||||
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']}%")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
|
@ -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())
|
||||
151
examples/sampling/server_fallback.py
Normal file
151
examples/sampling/server_fallback.py
Normal 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-20250514"),
|
||||
)
|
||||
|
||||
|
||||
@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())
|
||||
163
examples/sampling/structured_output.py
Normal file
163
examples/sampling/structured_output.py
Normal 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-20250514")
|
||||
|
||||
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
152
examples/sampling/text.py
Normal 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-20250514")
|
||||
|
||||
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())
|
||||
169
examples/sampling/tool_use.py
Normal file
169
examples/sampling/tool_use.py
Normal 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-20250514")
|
||||
|
||||
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())
|
||||
|
|
@ -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())
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -241,9 +241,7 @@ class OpenAISamplingHandler(BaseLLMSamplingHandler):
|
|||
if content.content:
|
||||
for item in content.content:
|
||||
if isinstance(item, TextContent):
|
||||
result_parts.append(
|
||||
{"type": "text", "text": item.text}
|
||||
)
|
||||
result_parts.append({"type": "text", "text": item.text})
|
||||
openai_messages.append(
|
||||
ChatCompletionToolMessageParam(
|
||||
role="tool",
|
||||
|
|
|
|||
391
src/fastmcp/server/sampling/anthropic.py
Normal file
391
src/fastmcp/server/sampling/anthropic.py
Normal 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-20250514",
|
||||
"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,
|
||||
)
|
||||
238
tests/server/sampling/test_anthropic_handler.py
Normal file
238
tests/server/sampling/test_anthropic_handler.py
Normal 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",
|
||||
}
|
||||
]
|
||||
36
uv.lock
generated
36
uv.lock
generated
|
|
@ -24,6 +24,25 @@ wheels = [
|
|||
{ url = "https://files.pythonhosted.org/packages/78/b6/6307fbef88d9b5ee7421e68d78a9f162e0da4900bc5f5793f6d3d0e34fb8/annotated_types-0.7.0-py3-none-any.whl", hash = "sha256:1f02e8b43a8fbbc3f3e0d4f0f4bfc8131bcb4eebe8849b8e5c773f3a1c582a53", size = 13643, upload-time = "2024-05-20T21:33:24.1Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anthropic"
|
||||
version = "0.75.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "anyio" },
|
||||
{ name = "distro" },
|
||||
{ name = "docstring-parser" },
|
||||
{ name = "httpx" },
|
||||
{ name = "jiter" },
|
||||
{ name = "pydantic" },
|
||||
{ name = "sniffio" },
|
||||
{ name = "typing-extensions" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/04/1f/08e95f4b7e2d35205ae5dcbb4ae97e7d477fc521c275c02609e2931ece2d/anthropic-0.75.0.tar.gz", hash = "sha256:e8607422f4ab616db2ea5baacc215dd5f028da99ce2f022e33c7c535b29f3dfb", size = 439565, upload-time = "2025-11-24T20:41:45.28Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/60/1c/1cd02b7ae64302a6e06724bf80a96401d5313708651d277b1458504a1730/anthropic-0.75.0-py3-none-any.whl", hash = "sha256:ea8317271b6c15d80225a9f3c670152746e88805a7a61e14d4a374577164965b", size = 388164, upload-time = "2025-11-24T20:41:43.587Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anyio"
|
||||
version = "4.12.0"
|
||||
|
|
@ -685,6 +704,9 @@ dependencies = [
|
|||
]
|
||||
|
||||
[package.optional-dependencies]
|
||||
anthropic = [
|
||||
{ name = "anthropic" },
|
||||
]
|
||||
openai = [
|
||||
{ name = "openai" },
|
||||
]
|
||||
|
|
@ -693,7 +715,7 @@ openai = [
|
|||
dev = [
|
||||
{ name = "dirty-equals" },
|
||||
{ name = "fastapi" },
|
||||
{ name = "fastmcp", extra = ["openai"] },
|
||||
{ name = "fastmcp", extra = ["anthropic", "openai"] },
|
||||
{ name = "inline-snapshot", extra = ["dirty-equals"] },
|
||||
{ name = "ipython", version = "8.37.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
|
||||
{ name = "ipython", version = "9.8.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
|
||||
|
|
@ -718,12 +740,13 @@ dev = [
|
|||
|
||||
[package.metadata]
|
||||
requires-dist = [
|
||||
{ name = "anthropic", marker = "extra == 'anthropic'", specifier = ">=0.40.0" },
|
||||
{ name = "authlib", specifier = ">=1.6.5" },
|
||||
{ name = "cyclopts", specifier = ">=4.0.0" },
|
||||
{ name = "exceptiongroup", specifier = ">=1.2.2" },
|
||||
{ name = "httpx", specifier = ">=0.28.1" },
|
||||
{ name = "jsonschema-path", specifier = ">=0.3.4" },
|
||||
{ name = "mcp", specifier = ">=1.23.1" },
|
||||
{ name = "mcp", specifier = ">=1.23.3" },
|
||||
{ name = "openai", marker = "extra == 'openai'", specifier = ">=1.102.0" },
|
||||
{ name = "openapi-pydantic", specifier = ">=0.5.1" },
|
||||
{ name = "platformdirs", specifier = ">=4.0.0" },
|
||||
|
|
@ -736,12 +759,13 @@ requires-dist = [
|
|||
{ name = "uvicorn", specifier = ">=0.35" },
|
||||
{ name = "websockets", specifier = ">=15.0.1" },
|
||||
]
|
||||
provides-extras = ["openai"]
|
||||
provides-extras = ["anthropic", "openai"]
|
||||
|
||||
[package.metadata.requires-dev]
|
||||
dev = [
|
||||
{ name = "dirty-equals", specifier = ">=0.9.0" },
|
||||
{ name = "fastapi", specifier = ">=0.115.12" },
|
||||
{ name = "fastmcp", extras = ["anthropic"] },
|
||||
{ name = "fastmcp", extras = ["openai"] },
|
||||
{ name = "inline-snapshot", extras = ["dirty-equals"], specifier = ">=0.27.2" },
|
||||
{ name = "ipython", specifier = ">=8.12.3" },
|
||||
|
|
@ -1237,7 +1261,7 @@ wheels = [
|
|||
|
||||
[[package]]
|
||||
name = "mcp"
|
||||
version = "1.23.1"
|
||||
version = "1.23.3"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "anyio" },
|
||||
|
|
@ -1255,9 +1279,9 @@ dependencies = [
|
|||
{ name = "typing-inspection" },
|
||||
{ name = "uvicorn", marker = "sys_platform != 'emscripten'" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/12/42/10c0c09ca27aceacd8c428956cfabdd67e3d328fe55c4abc16589285d294/mcp-1.23.1.tar.gz", hash = "sha256:7403e053e8e2283b1e6ae631423cb54736933fea70b32422152e6064556cd298", size = 596519, upload-time = "2025-12-02T18:41:12.807Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/a7/a4/d06a303f45997e266f2c228081abe299bbcba216cb806128e2e49095d25f/mcp-1.23.3.tar.gz", hash = "sha256:b3b0da2cc949950ce1259c7bfc1b081905a51916fcd7c8182125b85e70825201", size = 600697, upload-time = "2025-12-09T16:04:37.351Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/9f/9e/26e1d2d2c6afe15dfba5ca6799eeeea7656dce625c22766e4c57305e9cc2/mcp-1.23.1-py3-none-any.whl", hash = "sha256:3ce897fcc20a41bd50b4c58d3aa88085f11f505dcc0eaed48930012d34c731d8", size = 231433, upload-time = "2025-12-02T18:41:11.195Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/32/c6/13c1a26b47b3f3a3b480783001ada4268917c9f42d78a079c336da2e75e5/mcp-1.23.3-py3-none-any.whl", hash = "sha256:32768af4b46a1b4f7df34e2bfdf5c6011e7b63d7f1b0e321d0fdef4cd6082031", size = 231570, upload-time = "2025-12-09T16:04:35.56Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue