Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 23 additions & 3 deletions src/fastmcp/server/sampling/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -334,20 +334,40 @@ def prepare_messages(


def prepare_tools(
tools: Sequence[SamplingTool | Callable[..., Any]] | None,
tools: Sequence[SamplingTool | Callable[..., Any] | Any] | None,
) -> list[SamplingTool] | None:
"""Convert tools to SamplingTool objects."""
"""Convert tools to SamplingTool objects.

Accepts SamplingTool instances, FunctionTool instances, TransformedTool instances,
or plain callable functions. FunctionTool and TransformedTool are converted using
from_callable_tool(), while plain functions use from_function().

Args:
tools: Sequence of tools to prepare. Can be SamplingTool, FunctionTool,
TransformedTool, or plain callable functions.

Returns:
List of SamplingTool instances, or None if tools is None.
"""
if tools is None:
return None

# Import here to avoid circular dependencies and check for tool types
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool_transform import TransformedTool

sampling_tools: list[SamplingTool] = []
for t in tools:
if isinstance(t, SamplingTool):
sampling_tools.append(t)
elif isinstance(t, (FunctionTool, TransformedTool)):
sampling_tools.append(SamplingTool.from_callable_tool(t))
elif callable(t):
sampling_tools.append(SamplingTool.from_function(t))
else:
raise TypeError(f"Expected SamplingTool or callable, got {type(t)}")
raise TypeError(
f"Expected SamplingTool, FunctionTool, TransformedTool, or callable, got {type(t)}"
)

return sampling_tools if sampling_tools else None

Expand Down
94 changes: 94 additions & 0 deletions src/fastmcp/server/sampling/sampling_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,3 +107,97 @@ def from_function(
parameters=parsed.input_schema,
fn=parsed.fn,
)

@classmethod
def from_callable_tool(
cls,
tool: Any,
*,
name: str | None = None,
description: str | None = None,
) -> SamplingTool:
"""Create a SamplingTool from a FunctionTool or TransformedTool.

This helper enables reusing existing server tools in sampling contexts
without duplication. Both FunctionTool and TransformedTool have callable
.fn attributes that can be directly used for sampling.

For TransformedTool instances, the tool's .run() method is used instead
of .fn to ensure proper argument transformation and execution. The result
is automatically unwrapped from ToolResult if needed.

Args:
tool: A FunctionTool or TransformedTool with a callable .fn attribute.
name: Optional name override. Defaults to tool.name.
description: Optional description override. Defaults to tool.description.

Returns:
A SamplingTool that wraps the tool's functionality.

Raises:
AttributeError: If the tool doesn't have required attributes (.fn, .name, etc.).

Examples:
Convert a FunctionTool to SamplingTool:

@mcp.tool
def search(query: str) -> str:
return do_search(query)

sampling_tool = SamplingTool.from_callable_tool(search)

Use in sampling context:

result = await ctx.sample(
"Research Python",
tools=[SamplingTool.from_callable_tool(search)]
)
"""
# Import here to avoid circular dependencies
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import ToolResult
from fastmcp.tools.tool_transform import TransformedTool

# Validate that the tool is a supported type
if not isinstance(tool, (FunctionTool, TransformedTool)):
raise TypeError(
f"Expected FunctionTool or TransformedTool, got {type(tool).__name__}. "
"Only callable tools can be converted to SamplingTools."
)

# For TransformedTool, we need to use .run() and unwrap ToolResult
# because .fn might be a forwarding function that returns ToolResult
if isinstance(tool, TransformedTool):

async def wrapper(**kwargs: Any) -> Any:
result = await tool.run(kwargs)
# Unwrap ToolResult - extract the actual value
if isinstance(result, ToolResult):
# If there's structured_content, use that
if result.structured_content is not None:
# Handle wrapped results
if (
isinstance(result.structured_content, dict)
and "result" in result.structured_content
):
return result.structured_content["result"]
return result.structured_content

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Only unwrap ToolResult when wrap-result is enabled

The new TransformedTool wrapper treats any structured_content dict containing a "result" key as a wrapped payload and returns only that value. This loses data for legitimate schemas that include a result field alongside other properties (e.g., {result: ..., source: ...}), so sampling will drop fields and stringify only the inner value. This only occurs for TransformedTool outputs with structured content that happen to include a result key; consider checking the tool’s output schema (e.g., x-fastmcp-wrap-result) before unwrapping, otherwise return the dict unchanged.

Useful? React with 👍 / 👎.

# Otherwise, extract from text content
if result.content and len(result.content) > 0:
first_content = result.content[0]
if hasattr(first_content, "text"):
return first_content.text
return result

fn = wrapper
else:
# FunctionTool.fn can be used directly
fn = tool.fn
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated

# Extract the callable function, name, description, and parameters
return cls(
name=name or tool.name,
description=description or tool.description,
parameters=tool.parameters,
fn=fn,
)
111 changes: 111 additions & 0 deletions tests/server/sampling/test_prepare_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
"""Tests for prepare_tools helper function."""

import pytest

from fastmcp.server.sampling.run import prepare_tools
from fastmcp.server.sampling.sampling_tool import SamplingTool
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool_transform import ArgTransform, TransformedTool


class TestPrepareTools:
"""Tests for prepare_tools()."""

def test_prepare_tools_with_none(self):
"""Test that None returns None."""
result = prepare_tools(None)
assert result is None

def test_prepare_tools_with_sampling_tool(self):
"""Test that SamplingTool instances pass through."""

def search(query: str) -> str:
return f"Results: {query}"

sampling_tool = SamplingTool.from_function(search)
result = prepare_tools([sampling_tool])

assert result is not None
assert len(result) == 1
assert result[0] is sampling_tool

def test_prepare_tools_with_function(self):
"""Test that plain functions are converted."""

def search(query: str) -> str:
"""Search function."""
return f"Results: {query}"

result = prepare_tools([search])

assert result is not None
assert len(result) == 1
assert isinstance(result[0], SamplingTool)
assert result[0].name == "search"

def test_prepare_tools_with_function_tool(self):
"""Test that FunctionTool instances are converted."""

def search(query: str) -> str:
"""Search the web."""
return f"Results: {query}"

function_tool = FunctionTool.from_function(search)
result = prepare_tools([function_tool])

assert result is not None
assert len(result) == 1
assert isinstance(result[0], SamplingTool)
assert result[0].name == "search"
assert result[0].description == "Search the web."

def test_prepare_tools_with_transformed_tool(self):
"""Test that TransformedTool instances are converted."""

def original(query: str) -> str:
"""Original tool."""
return f"Results: {query}"

function_tool = FunctionTool.from_function(original)
transformed_tool = TransformedTool.from_tool(
function_tool,
name="search_v2",
transform_args={"query": ArgTransform(name="q")},
)

result = prepare_tools([transformed_tool])

assert result is not None
assert len(result) == 1
assert isinstance(result[0], SamplingTool)
assert result[0].name == "search_v2"
assert "q" in result[0].parameters.get("properties", {})

def test_prepare_tools_with_mixed_types(self):
"""Test that mixed tool types are all converted."""

def plain_fn(x: int) -> int:
return x * 2

def fn_for_tool(y: int) -> int:
return y * 3

function_tool = FunctionTool.from_function(fn_for_tool)
sampling_tool = SamplingTool.from_function(lambda z: z * 4, name="lambda_tool")

result = prepare_tools([plain_fn, function_tool, sampling_tool])

assert result is not None
assert len(result) == 3
assert all(isinstance(t, SamplingTool) for t in result)

def test_prepare_tools_with_invalid_type(self):
"""Test that invalid types raise TypeError."""

with pytest.raises(TypeError, match="Expected SamplingTool, FunctionTool"):
prepare_tools(["not a tool"])

def test_prepare_tools_empty_list(self):
"""Test that empty list returns None."""
result = prepare_tools([])
assert result is None
114 changes: 114 additions & 0 deletions tests/server/sampling/test_sampling_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
import pytest

from fastmcp.server.sampling import SamplingTool
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool_transform import ArgTransform, TransformedTool


class TestSamplingToolFromFunction:
Expand Down Expand Up @@ -119,3 +121,115 @@ def search(query: str) -> str:
assert sdk_tool.name == "search"
assert sdk_tool.description == "Search the web."
assert "query" in sdk_tool.inputSchema.get("properties", {})


class TestSamplingToolFromCallableTool:
"""Tests for SamplingTool.from_callable_tool()."""

def test_from_function_tool(self):
"""Test converting a FunctionTool to SamplingTool."""

def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"

function_tool = FunctionTool.from_function(search)
sampling_tool = SamplingTool.from_callable_tool(function_tool)

assert sampling_tool.name == "search"
assert sampling_tool.description == "Search the web."
assert "query" in sampling_tool.parameters.get("properties", {})
assert sampling_tool.fn is function_tool.fn

def test_from_function_tool_with_overrides(self):
"""Test converting FunctionTool with name/description overrides."""

def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"

function_tool = FunctionTool.from_function(search)
sampling_tool = SamplingTool.from_callable_tool(
function_tool,
name="web_search",
description="Search the internet",
)

assert sampling_tool.name == "web_search"
assert sampling_tool.description == "Search the internet"

def test_from_transformed_tool(self):
"""Test converting a TransformedTool to SamplingTool."""

def original(query: str, limit: int) -> str:
"""Original tool."""
return f"Results for: {query} (limit: {limit})"

function_tool = FunctionTool.from_function(original)
transformed_tool = TransformedTool.from_tool(
function_tool,
name="search_transformed",
transform_args={"query": ArgTransform(name="q")},
)

sampling_tool = SamplingTool.from_callable_tool(transformed_tool)

assert sampling_tool.name == "search_transformed"
assert sampling_tool.description == "Original tool."
# The transformed tool should have 'q' instead of 'query'
assert "q" in sampling_tool.parameters.get("properties", {})
assert "limit" in sampling_tool.parameters.get("properties", {})

async def test_from_function_tool_execution(self):
"""Test that converted FunctionTool executes correctly."""

def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b

function_tool = FunctionTool.from_function(add)
sampling_tool = SamplingTool.from_callable_tool(function_tool)

result = await sampling_tool.run({"a": 2, "b": 3})
assert result == 5

async def test_from_transformed_tool_execution(self):
"""Test that converted TransformedTool executes correctly."""

def multiply(x: int, y: int) -> int:
"""Multiply two numbers."""
return x * y

function_tool = FunctionTool.from_function(multiply)
transformed_tool = TransformedTool.from_tool(
function_tool,
transform_args={"x": ArgTransform(name="a"), "y": ArgTransform(name="b")},
)

sampling_tool = SamplingTool.from_callable_tool(transformed_tool)

# Use the transformed parameter names
result = await sampling_tool.run({"a": 3, "b": 4})
# Result should be unwrapped from ToolResult
assert result == 12

def test_from_invalid_tool_type(self):
"""Test that from_callable_tool rejects non-tool objects."""

class NotATool:
pass

with pytest.raises(
TypeError,
match="Expected FunctionTool or TransformedTool",
):
SamplingTool.from_callable_tool(NotATool())

def test_from_plain_function_fails(self):
"""Test that plain functions are rejected by from_callable_tool."""

def my_function():
pass

with pytest.raises(TypeError, match="Expected FunctionTool or TransformedTool"):
SamplingTool.from_callable_tool(my_function)
Loading