ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 05:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Split Tool into Tool and FunctionTool

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-04 10:19:14 -04:00
commit 3a3c01708d
17 changed files with 207 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/fastmcp/resources/resource.py
View file

@ -8,7 +8,6 @@ from typing import TYPE_CHECKING, Annotated, Any
from mcp.types import Resource as MCPResource
from pydantic import (
AnyUrl,
BaseModel,
BeforeValidator,
ConfigDict,
Field,
@ -17,13 +16,13 @@ from pydantic import (
field_validator,
)
from fastmcp.utilities.types import _convert_set_defaults
from fastmcp.utilities.types import FastMCPBaseModel, _convert_set_defaults
if TYPE_CHECKING:
pass
class Resource(BaseModel, abc.ABC):
class Resource(FastMCPBaseModel, abc.ABC):
"""Base class for all resources."""
model_config = ConfigDict(validate_default=True)

9
src/fastmcp/resources/template.py
View file

@ -11,7 +11,6 @@ from urllib.parse import unquote
from mcp.types import ResourceTemplate as MCPResourceTemplate
from pydantic import (
AnyUrl,
BaseModel,
BeforeValidator,
Field,
field_validator,
@ -22,6 +21,7 @@ from fastmcp.resources.types import FunctionResource, Resource
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.types import (
FastMCPBaseModel,
_convert_set_defaults,
find_kwarg_by_type,
get_cached_typeadapter,
@ -52,12 +52,7 @@ def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
return None
class MyModel(BaseModel):
key: str
value: int
class ResourceTemplate(BaseModel):
class ResourceTemplate(FastMCPBaseModel):
"""A template for dynamically creating resources."""
uri_template: str = Field(

48
src/fastmcp/server/openapi.py
View file

@ -233,7 +233,6 @@ class OpenAPITool(Tool):
name=name,
description=description,
parameters=parameters,
fn=self._execute_request, # We'll use an instance method instead of a global function
tags=tags,
annotations=annotations,
exclude_args=exclude_args,
@ -247,9 +246,10 @@ class OpenAPITool(Tool):
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPITool(name={self.name!r}, method={self._route.method}, path={self._route.path})"
async def _execute_request(self, *args, **kwargs):
async def run(
self, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Execute the HTTP request based on the route configuration."""
context = kwargs.get("context")
# Prepare URL
path = self._route.path
@ -258,11 +258,11 @@ class OpenAPITool(Tool):
# Path parameters should never be None as they're typically required
# but we'll handle that case anyway
path_params = {
p.name: kwargs.get(p.name)
p.name: arguments.get(p.name)
for p in self._route.parameters
if p.location == "path"
and p.name in kwargs
and kwargs.get(p.name) is not None
and p.name in arguments
and arguments.get(p.name) is not None
}
# Ensure all path parameters are provided
@ -340,11 +340,11 @@ class OpenAPITool(Tool):
for p in self._route.parameters:
if (
p.location == "query"
and p.name in kwargs
and kwargs.get(p.name) is not None
and kwargs.get(p.name) != ""
and p.name in arguments
and arguments.get(p.name) is not None
and arguments.get(p.name) != ""
):
param_value = kwargs.get(p.name)
param_value = arguments.get(p.name)
# Format array query parameters as comma-separated strings
# following OpenAPI form style (default for query parameters)
@ -399,10 +399,10 @@ class OpenAPITool(Tool):
for p in self._route.parameters:
if (
p.location == "header"
and p.name in kwargs
and kwargs[p.name] is not None
and p.name in arguments
and arguments[p.name] is not None
):
openapi_headers[p.name.lower()] = str(kwargs[p.name])
openapi_headers[p.name.lower()] = str(arguments[p.name])
headers.update(openapi_headers)
# Add headers from the current MCP client HTTP request (these take precedence)
@ -420,21 +420,13 @@ class OpenAPITool(Tool):
}
body_params = {
k: v
for k, v in kwargs.items()
for k, v in arguments.items()
if k not in path_query_header_params and k != "context"
}
if body_params:
json_data = body_params
# Log the request details if a context is available
if context:
try:
await context.info(f"Making {self._route.method} request to {path}")
except (ValueError, AttributeError):
# Silently continue if context logging is not available
pass
# Execute the request
try:
response = await self._client.request(
@ -451,10 +443,11 @@ class OpenAPITool(Tool):
# Try to parse as JSON first
try:
return response.json()
result = response.json()
except (json.JSONDecodeError, ValueError):
# Return text content if not JSON
return response.text
result = response.text
return _convert_to_content(result)
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx)
@ -474,13 +467,6 @@ class OpenAPITool(Tool):
# Handle request errors (connection, timeout, etc.)
raise ValueError(f"Request error: {str(e)}")
async def run(
self, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Run the tool with arguments and optional context."""
response = await self._execute_request(**arguments)
return _convert_to_content(response)
class OpenAPIResource(Resource):
"""Resource implementation for OpenAPI endpoints."""

5
src/fastmcp/server/proxy.py
View file

@ -48,7 +48,6 @@ class ProxyTool(Tool):
name=tool.name,
description=tool.description,
parameters=tool.inputSchema,
fn=_proxy_passthrough,
)
async def run(
@ -69,6 +68,9 @@ class ProxyTool(Tool):
class ProxyResource(Resource):
_client: Client
_value: str | bytes | None = None
def __init__(self, client: Client, *, _value: str | bytes | None = None, **kwargs):
super().__init__(**kwargs)
self._client = client
@ -146,7 +148,6 @@ class ProxyTemplate(ResourceTemplate):
name=self.name,
description=self.description,
mime_type=result[0].mimeType,
contents=result,
_value=value,
)

6
src/fastmcp/server/server.py
View file

@ -55,7 +55,7 @@ from fastmcp.server.http import (
create_streamable_http_app,
)
from fastmcp.tools import ToolManager
from fastmcp.tools.tool import Tool
from fastmcp.tools.tool import FunctionTool, Tool
from fastmcp.utilities.cache import TimedCache
from fastmcp.utilities.decorators import DecoratedFunction
from fastmcp.utilities.logging import get_logger
@ -508,7 +508,7 @@ class FastMCP(Generic[LifespanResultT]):
if isinstance(annotations, dict):
annotations = ToolAnnotations(**annotations)
self._tool_manager.add_tool_from_fn(
tool = FunctionTool.from_function(
fn,
name=name,
description=description,
@ -516,6 +516,8 @@ class FastMCP(Generic[LifespanResultT]):
annotations=annotations,
exclude_args=exclude_args,
)
self._tool_manager.add_tool(tool)
self._cache.clear()
def remove_tool(self, name: str) -> None:

4
src/fastmcp/tools/__init__.py
View file

@ -1,4 +1,4 @@
from .tool import Tool
from .tool import Tool, FunctionTool
from .tool_manager import ToolManager
__all__ = ["Tool", "ToolManager"]
__all__ = ["Tool", "ToolManager", "FunctionTool"]

57
src/fastmcp/tools/tool.py
View file

@ -2,19 +2,22 @@ from __future__ import annotations
import inspect
import json
import warnings
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import TYPE_CHECKING, Annotated, Any
import pydantic_core
from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
from mcp.types import Tool as MCPTool
from pydantic import BaseModel, BeforeValidator, Field
from pydantic import BeforeValidator, Field
import fastmcp
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
FastMCPBaseModel,
Image,
_convert_set_defaults,
find_kwarg_by_type,
@ -31,10 +34,9 @@ def default_serializer(data: Any) -> str:
return pydantic_core.to_json(data, fallback=str, indent=2).decode()
class Tool(BaseModel):
class Tool(FastMCPBaseModel, ABC):
"""Internal tool registration info."""
fn: Callable[..., Any]
name: str = Field(description="Name of the tool")
description: str | None = Field(
default=None, description="Description of what the tool does"
@ -54,6 +56,39 @@ class Tool(BaseModel):
None, description="Optional custom serializer for tool results"
)
def to_mcp_tool(self, **overrides: Any) -> MCPTool:
kwargs = {
"name": self.name,
"description": self.description,
"inputSchema": self.parameters,
"annotations": self.annotations,
}
return MCPTool(**kwargs | overrides)
@staticmethod
def from_function(fn: Callable[..., Any], **overrides: Any) -> FunctionTool:
# deprecated in 2.6.2
warnings.warn(
"Tool.from_function() is deprecated. Use FunctionTool.from_function() instead."
)
return FunctionTool.from_function(fn, **overrides)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Tool):
return False
return self.model_dump() == other.model_dump()
@abstractmethod
async def run(
self, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Run the tool with arguments."""
raise NotImplementedError("Subclasses must implement run()")
class FunctionTool(Tool):
fn: Callable[..., Any]
@classmethod
def from_function(
cls,
@ -64,7 +99,7 @@ class Tool(BaseModel):
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
serializer: Callable[[Any], str] | None = None,
) -> Tool:
) -> FunctionTool:
"""Create a Tool from a function."""
from fastmcp.server.context import Context
@ -170,20 +205,6 @@ class Tool(BaseModel):
return _convert_to_content(result, serializer=self.serializer)
def to_mcp_tool(self, **overrides: Any) -> MCPTool:
kwargs = {
"name": self.name,
"description": self.description,
"inputSchema": self.parameters,
"annotations": self.annotations,
}
return MCPTool(**kwargs | overrides)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Tool):
return False
return self.model_dump() == other.model_dump()
def _convert_to_content(
result: Any,

4
src/fastmcp/tools/tool_manager.py
View file

@ -7,7 +7,7 @@ from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotatio
from fastmcp.exceptions import NotFoundError, ToolError
from fastmcp.settings import DuplicateBehavior
from fastmcp.tools.tool import Tool
from fastmcp.tools.tool import FunctionTool, Tool
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
@ -69,7 +69,7 @@ class ToolManager:
exclude_args: list[str] | None = None,
) -> Tool:
"""Add a tool to the server."""
tool = Tool.from_function(
tool = FunctionTool.from_function(
fn,
name=name,
description=description,

10
src/fastmcp/utilities/mcp_config.py
View file

@ -3,7 +3,9 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, Literal
from urllib.parse import urlparse
from pydantic import AnyUrl, BaseModel, Field
from pydantic import AnyUrl, Field
from fastmcp.utilities.types import FastMCPBaseModel
if TYPE_CHECKING:
from fastmcp.client.transports import (
@ -32,7 +34,7 @@ def infer_transport_type_from_url(
return "streamable-http"
class StdioMCPServer(BaseModel):
class StdioMCPServer(FastMCPBaseModel):
command: str
args: list[str] = Field(default_factory=list)
env: dict[str, Any] = Field(default_factory=dict)
@ -50,7 +52,7 @@ class StdioMCPServer(BaseModel):
)
class RemoteMCPServer(BaseModel):
class RemoteMCPServer(FastMCPBaseModel):
url: str
headers: dict[str, str] = Field(default_factory=dict)
transport: Literal["streamable-http", "sse", "http"] | None = None
@ -69,7 +71,7 @@ class RemoteMCPServer(BaseModel):
return StreamableHttpTransport(self.url, headers=self.headers)
class MCPConfig(BaseModel):
class MCPConfig(FastMCPBaseModel):
mcpServers: dict[str, StdioMCPServer | RemoteMCPServer]
@classmethod

9
src/fastmcp/utilities/openapi.py
View file

@ -25,6 +25,7 @@ from openapi_pydantic.v3.v3_0 import Schema as Schema_30
from pydantic import BaseModel, Field, ValidationError
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.types import FastMCPBaseModel
logger = logging.getLogger(__name__)
@ -38,7 +39,7 @@ ParameterLocation = Literal["path", "query", "header", "cookie"]
JsonSchema = dict[str, Any]
class ParameterInfo(BaseModel):
class ParameterInfo(FastMCPBaseModel):
"""Represents a single parameter for an HTTP operation in our IR."""
name: str
@ -48,7 +49,7 @@ class ParameterInfo(BaseModel):
description: str | None = None
class RequestBodyInfo(BaseModel):
class RequestBodyInfo(FastMCPBaseModel):
"""Represents the request body for an HTTP operation in our IR."""
required: bool = False
@ -58,7 +59,7 @@ class RequestBodyInfo(BaseModel):
description: str | None = None
class ResponseInfo(BaseModel):
class ResponseInfo(FastMCPBaseModel):
"""Represents response information in our IR."""
description: str | None = None
@ -66,7 +67,7 @@ class ResponseInfo(BaseModel):
content_schema: dict[str, JsonSchema] = Field(default_factory=dict)
class HTTPRoute(BaseModel):
class HTTPRoute(FastMCPBaseModel):
"""Intermediate Representation for a single OpenAPI operation."""
path: str

8
src/fastmcp/utilities/types.py
View file

@ -9,11 +9,17 @@ from types import UnionType
from typing import Annotated, TypeVar, Union, get_args, get_origin
from mcp.types import ImageContent
from pydantic import TypeAdapter
from pydantic import BaseModel, ConfigDict, TypeAdapter
T = TypeVar("T")
class FastMCPBaseModel(BaseModel):
"""Base model for FastMCP models."""
model_config = ConfigDict(extra="forbid")
@lru_cache(maxsize=5000)
def get_cached_typeadapter(cls: T) -> TypeAdapter[T]:
"""

76
tests/deprecated/test_tool_from_function_deprecated.py Normal file
View file

@ -0,0 +1,76 @@
"""Tests for deprecated Tool.from_function() method.
The Tool.from_function() method was deprecated in version 2.6.2 in favor of
FunctionTool.from_function().
"""
import warnings
import pytest
from fastmcp.tools.tool import FunctionTool, Tool
def test_tool_from_function_deprecation_warning():
"""Test that Tool.from_function() raises a deprecation warning."""
def example_function(x: int) -> str:
"""Example function for testing."""
return f"Result: {x}"
with pytest.warns(
UserWarning,
match="Tool.from_function\\(\\) is deprecated. Use FunctionTool.from_function\\(\\) instead.",
):
tool = Tool.from_function(example_function)
# Verify the tool was created correctly despite the warning
assert isinstance(tool, FunctionTool)
assert tool.name == "example_function"
assert tool.description == "Example function for testing."
def test_tool_from_function_produces_same_result_as_function_tool():
"""Test that Tool.from_function() produces the same result as FunctionTool.from_function()."""
def example_function(x: int, y: str = "default") -> dict:
"""Example function with parameters."""
return {"x": x, "y": y}
# Create tool using the deprecated method (with warning suppressed)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
deprecated_tool = Tool.from_function(example_function)
# Create tool using the new method
new_tool = FunctionTool.from_function(example_function)
# They should be equivalent
assert deprecated_tool == new_tool
assert deprecated_tool.name == new_tool.name
assert deprecated_tool.description == new_tool.description
assert deprecated_tool.parameters == new_tool.parameters
def test_tool_from_function_with_overrides():
"""Test that Tool.from_function() works with parameter overrides."""
def example_function() -> str:
"""Original description."""
return "test"
custom_name = "custom_tool_name"
custom_description = "Custom description"
custom_tags = {"test", "deprecated"}
with pytest.warns(UserWarning, match="Tool.from_function\\(\\) is deprecated"):
tool = Tool.from_function(
example_function,
name=custom_name,
description=custom_description,
tags=custom_tags,
)
assert tool.name == custom_name
assert tool.description == custom_description
assert tool.tags == custom_tags

14
tests/server/openapi/test_openapi_path_parameters.py
View file

@ -128,7 +128,7 @@ async def test_array_path_parameter_handling(mock_client):
)
# Test with a single value
await tool._execute_request(days=["monday"])
await tool.run({"days": ["monday"]})
# Check that the path parameter is formatted correctly
# This is where the bug is: it should be '/select/monday' not '/select/[\'monday\']'
@ -143,7 +143,7 @@ async def test_array_path_parameter_handling(mock_client):
mock_client.request.reset_mock()
# Test with multiple values
await tool._execute_request(days=["monday", "tuesday"])
await tool.run({"days": ["monday", "tuesday"]})
# Check that the path parameter is formatted correctly
# It should be '/select/monday,tuesday' not '/select/[\'monday\', \'tuesday\']'
@ -234,7 +234,7 @@ async def test_complex_nested_array_path_parameter(mock_client):
]
# Execute the request with complex filters
await tool._execute_request(filters=complex_filters)
await tool.run({"filters": complex_filters})
# The complex object should be properly serialized in the URL
# For path parameters, this would typically need a custom serialization strategy
@ -359,7 +359,7 @@ async def test_array_query_parameter_format(mock_client):
)
# Test with a single value
await tool._execute_request(days=["monday"])
await tool.run({"days": ["monday"]})
# Check that the query parameter is formatted correctly
mock_client.request.assert_called_with(
@ -373,7 +373,7 @@ async def test_array_query_parameter_format(mock_client):
mock_client.request.reset_mock()
# Test with multiple values
await tool._execute_request(days=["monday", "tuesday"])
await tool.run({"days": ["monday", "tuesday"]})
# Check that the query parameter is formatted correctly
# It should be 'days=monday,tuesday' not 'days=["monday","tuesday"]'
@ -429,7 +429,7 @@ async def test_array_query_parameter_exploded_format(mock_client):
)
# Test with a single value
await tool._execute_request(days=["monday"])
await tool.run({"days": ["monday"]})
# Check that the query parameter is formatted correctly
mock_client.request.assert_called_with(
@ -443,7 +443,7 @@ async def test_array_query_parameter_exploded_format(mock_client):
mock_client.request.reset_mock()
# Test with multiple values
await tool._execute_request(days=["monday", "tuesday"])
await tool.run({"days": ["monday", "tuesday"]})
# Check that the query parameter is formatted correctly
# It should be passed as an array, which httpx will serialize as days=monday&days=tuesday

4
tests/server/test_import_server.py
View file

@ -3,6 +3,7 @@ from urllib.parse import quote
from fastmcp.client.client import Client
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import FunctionTool
async def test_import_basic_functionality():
@ -27,6 +28,7 @@ async def test_import_basic_functionality():
tool = main_app._tool_manager.get_tool("sub_sub_tool")
assert tool is not None
assert tool.name == "sub_tool"
assert isinstance(tool, FunctionTool)
assert callable(tool.fn)
@ -205,6 +207,7 @@ async def test_tool_custom_name_preserved_when_imported():
assert tool is not None
# Check that the function name is preserved
assert isinstance(tool, FunctionTool)
assert tool.fn.__name__ == "fetch_data"
@ -238,6 +241,7 @@ async def test_first_level_importing_with_custom_name():
# Tool is accessible in the service app with the first prefix
tool = service_app._tool_manager.get_tool("provider_compute")
assert tool is not None
assert isinstance(tool, FunctionTool)
assert tool.fn.__name__ == "calculate_value"

4
tests/server/test_server.py
View file

@ -12,7 +12,7 @@ from fastmcp.server.server import (
has_resource_prefix,
remove_resource_prefix,
)
from fastmcp.tools.tool import Tool
from fastmcp.tools import FunctionTool
class TestCreateServer:
@ -102,7 +102,7 @@ class TestTools:
"""add two to a number"""
return x + 2
g_tool = Tool.from_function(g, name="g-tool")
g_tool = FunctionTool.from_function(g, name="g-tool")
mcp = FastMCP(tools=[f, g_tool])

42
tests/tools/test_tool.py
View file

@ -5,7 +5,7 @@ from pydantic import BaseModel
from fastmcp import FastMCP, Image
from fastmcp.client import Client
from fastmcp.exceptions import ToolError
from fastmcp.tools.tool import Tool
from fastmcp.tools import FunctionTool
from fastmcp.utilities.tests import temporary_settings
@ -17,7 +17,7 @@ class TestToolFromFunction:
"""Add two numbers."""
return a + b
tool = Tool.from_function(add)
tool = FunctionTool.from_function(add)
assert tool.name == "add"
assert tool.description == "Add two numbers."
@ -32,7 +32,7 @@ class TestToolFromFunction:
"""Fetch data from URL."""
return f"Data from {url}"
tool = Tool.from_function(fetch_data)
tool = FunctionTool.from_function(fetch_data)
assert tool.name == "fetch_data"
assert tool.description == "Fetch data from URL."
@ -46,7 +46,7 @@ class TestToolFromFunction:
"""ignore this"""
return x + y
tool = Tool.from_function(Adder())
tool = FunctionTool.from_function(Adder())
assert tool.name == "Adder"
assert tool.description == "Adds two numbers."
assert len(tool.parameters["properties"]) == 2
@ -61,7 +61,7 @@ class TestToolFromFunction:
"""ignore this"""
return x + y
tool = Tool.from_function(Adder())
tool = FunctionTool.from_function(Adder())
assert tool.name == "Adder"
assert tool.description == "Adds two numbers."
assert len(tool.parameters["properties"]) == 2
@ -79,7 +79,7 @@ class TestToolFromFunction:
"""Create a new user."""
return {"id": 1, **user.model_dump()}
tool = Tool.from_function(create_user)
tool = FunctionTool.from_function(create_user)
assert tool.name == "create_user"
assert tool.description == "Create a new user."
@ -91,7 +91,7 @@ class TestToolFromFunction:
def image_tool(data: bytes) -> Image:
return Image(data=data)
tool = Tool.from_function(image_tool)
tool = FunctionTool.from_function(image_tool)
result = await tool.run({"data": "test.png"})
assert tool.parameters["properties"]["data"]["type"] == "string"
@ -99,24 +99,24 @@ class TestToolFromFunction:
def test_non_callable_fn(self):
with pytest.raises(TypeError, match="not a callable object"):
Tool.from_function(1) # type: ignore
FunctionTool.from_function(1) # type: ignore
def test_lambda(self):
tool = Tool.from_function(lambda x: x, name="my_tool")
tool = FunctionTool.from_function(lambda x: x, name="my_tool")
assert tool.name == "my_tool"
def test_lambda_with_no_name(self):
with pytest.raises(
ValueError, match="You must provide a name for lambda functions"
):
Tool.from_function(lambda x: x)
FunctionTool.from_function(lambda x: x)
def test_private_arguments(self):
def add(_a: int, _b: int) -> int:
"""Add two numbers."""
return _a + _b
tool = Tool.from_function(add)
tool = FunctionTool.from_function(add)
assert tool.parameters["properties"]["_a"]["type"] == "integer"
assert tool.parameters["properties"]["_b"]["type"] == "integer"
@ -128,7 +128,7 @@ class TestToolFromFunction:
with pytest.raises(
ValueError, match=r"Functions with \*args are not supported as tools"
):
Tool.from_function(func)
FunctionTool.from_function(func)
def test_tool_with_varkwargs_not_allowed(self):
def func(a: int, b: int, **kwargs: int) -> int:
@ -138,7 +138,7 @@ class TestToolFromFunction:
with pytest.raises(
ValueError, match=r"Functions with \*\*kwargs are not supported as tools"
):
Tool.from_function(func)
FunctionTool.from_function(func)
async def test_instance_method(self):
class MyClass:
@ -148,7 +148,7 @@ class TestToolFromFunction:
obj = MyClass()
tool = Tool.from_function(obj.add)
tool = FunctionTool.from_function(obj.add)
assert tool.name == "add"
assert tool.description == "Add two numbers."
assert "self" not in tool.parameters["properties"]
@ -164,7 +164,7 @@ class TestToolFromFunction:
with pytest.raises(
ValueError, match=r"Functions with \*args are not supported as tools"
):
Tool.from_function(obj.add)
FunctionTool.from_function(obj.add)
async def test_instance_method_with_varkwargs_not_allowed(self):
class MyClass:
@ -177,7 +177,7 @@ class TestToolFromFunction:
with pytest.raises(
ValueError, match=r"Functions with \*\*kwargs are not supported as tools"
):
Tool.from_function(obj.add)
FunctionTool.from_function(obj.add)
async def test_classmethod(self):
class MyClass:
@ -199,7 +199,7 @@ class TestLegacyToolJsonParsing:
return f"{x}-{','.join(y)}"
# Create a tool to use its JSON pre-parsing logic
tool = Tool.from_function(simple_func)
tool = FunctionTool.from_function(simple_func)
# Prepare arguments where some are JSON strings
json_args = {
@ -217,7 +217,7 @@ class TestLegacyToolJsonParsing:
def func_with_str_types(str_or_list: str | list[str]) -> str | list[str]:
return str_or_list
tool = Tool.from_function(func_with_str_types)
tool = FunctionTool.from_function(func_with_str_types)
# Test regular string input (should remain a string)
result = await tool.run({"str_or_list": "hello"})
@ -243,7 +243,7 @@ class TestLegacyToolJsonParsing:
def func_with_str_types(string: str) -> str:
return string
tool = Tool.from_function(func_with_str_types)
tool = FunctionTool.from_function(func_with_str_types)
# Invalid JSON should remain a string
invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}"
@ -258,7 +258,7 @@ class TestLegacyToolJsonParsing:
) -> str | dict[int, str] | None:
return string
tool = Tool.from_function(func_with_str_types)
tool = FunctionTool.from_function(func_with_str_types)
# Invalid JSON for the union type should remain a string
invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}"
@ -275,7 +275,7 @@ class TestLegacyToolJsonParsing:
def func_with_complex_type(data: SomeModel) -> SomeModel:
return data
tool = Tool.from_function(func_with_complex_type)
tool = FunctionTool.from_function(func_with_complex_type)
# Valid JSON for the model
valid_json = '{"x": 1, "y": {"1": "hello"}}'

14
tests/tools/test_tool_manager.py
View file

@ -10,8 +10,7 @@ from pydantic import BaseModel
from fastmcp import Context, FastMCP, Image
from fastmcp.exceptions import NotFoundError, ToolError
from fastmcp.tools import ToolManager
from fastmcp.tools.tool import Tool
from fastmcp.tools import FunctionTool, ToolManager
from fastmcp.utilities.tests import temporary_settings
@ -212,6 +211,7 @@ class TestAddTools:
# Should have replaced with the new function
tool = manager.get_tool("test_tool")
assert tool is not None
assert isinstance(tool, FunctionTool)
assert tool.fn.__name__ == "replacement_fn"
def test_ignore_duplicate_tools(self):
@ -230,8 +230,10 @@ class TestAddTools:
# Should keep the original
tool = manager.get_tool("test_tool")
assert tool is not None
assert isinstance(tool, FunctionTool)
assert tool.fn.__name__ == "original_fn"
# Result should be the original tool
assert isinstance(result, FunctionTool)
assert result.fn.__name__ == "original_fn"
@ -566,7 +568,7 @@ class TestContextHandling:
def test_context_parameter_detection(self):
"""Test that context parameters are properly detected in
Tool.from_function()."""
FunctionTool.from_function()."""
def tool_with_context(x: int, ctx: Context) -> str:
return str(x)
@ -632,7 +634,7 @@ class TestContextHandling:
def test_parameterized_context_parameter_detection(self):
"""Test that context parameters are properly detected in
Tool.from_function()."""
FunctionTool.from_function()."""
def tool_with_context(x: int, ctx: Context) -> str:
return str(x)
@ -649,7 +651,7 @@ class TestContextHandling:
def test_parameterized_union_context_parameter_detection(self):
"""Test that context parameters are properly detected in
Tool.from_function()."""
FunctionTool.from_function()."""
def tool_with_context(x: int, ctx: Context | None) -> str:
return str(x)
@ -703,7 +705,7 @@ class TestCustomToolNames:
return x + 1
# Create a tool with a specific name
tool = Tool.from_function(fn, name="my_tool")
tool = FunctionTool.from_function(fn, name="my_tool")
manager = ToolManager()
# Store it under a different name
manager.add_tool(tool, key="proxy_tool")