diff --git a/src/fastmcp/prompts/prompt.py b/src/fastmcp/prompts/prompt.py index d9bae2b9a..ed7ee53bc 100644 --- a/src/fastmcp/prompts/prompt.py +++ b/src/fastmcp/prompts/prompt.py @@ -3,6 +3,7 @@ from __future__ import annotations as _annotations import inspect +from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable, Sequence from typing import TYPE_CHECKING, Annotated, Any @@ -10,13 +11,14 @@ import pydantic_core from mcp.types import EmbeddedResource, ImageContent, PromptMessage, Role, TextContent from mcp.types import Prompt as MCPPrompt from mcp.types import PromptArgument as MCPPromptArgument -from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_call +from pydantic import BeforeValidator, Field, TypeAdapter, validate_call from fastmcp.exceptions import PromptError 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, _convert_set_defaults, find_kwarg_by_type, get_cached_typeadapter, @@ -52,7 +54,7 @@ SyncPromptResult = ( PromptResult = SyncPromptResult | Awaitable[SyncPromptResult] -class PromptArgument(BaseModel): +class PromptArgument(FastMCPBaseModel): """An argument that can be passed to a prompt.""" name: str = Field(description="Name of the argument") @@ -64,7 +66,7 @@ class PromptArgument(BaseModel): ) -class Prompt(BaseModel): +class Prompt(FastMCPBaseModel, ABC): """A prompt template that can be rendered with parameters.""" name: str = Field(description="Name of the prompt") @@ -77,6 +79,61 @@ class Prompt(BaseModel): arguments: list[PromptArgument] | None = Field( None, description="Arguments that can be passed to the prompt" ) + + def __eq__(self, other: object) -> bool: + if type(self) is not type(other): + return False + assert isinstance(other, type(self)) + return self.model_dump() == other.model_dump() + + def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt: + """Convert the prompt to an MCP prompt.""" + arguments = [ + MCPPromptArgument( + name=arg.name, + description=arg.description, + required=arg.required, + ) + for arg in self.arguments or [] + ] + kwargs = { + "name": self.name, + "description": self.description, + "arguments": arguments, + } + return MCPPrompt(**kwargs | overrides) + + @staticmethod + def from_function( + fn: Callable[..., PromptResult | Awaitable[PromptResult]], + name: str | None = None, + description: str | None = None, + tags: set[str] | None = None, + ) -> FunctionPrompt: + """Create a Prompt from a function. + + The function can return: + - A string (converted to a message) + - A Message object + - A dict (converted to a message) + - A sequence of any of the above + """ + return FunctionPrompt.from_function( + fn=fn, name=name, description=description, tags=tags + ) + + @abstractmethod + async def render( + self, + arguments: dict[str, Any] | None = None, + ) -> list[PromptMessage]: + """Render the prompt with arguments.""" + raise NotImplementedError("Prompt.render() must be implemented by subclasses") + + +class FunctionPrompt(Prompt): + """A prompt that is a function.""" + fn: Callable[..., PromptResult | Awaitable[PromptResult]] @classmethod @@ -86,7 +143,7 @@ class Prompt(BaseModel): name: str | None = None, description: str | None = None, tags: set[str] | None = None, - ) -> Prompt: + ) -> FunctionPrompt: """Create a Prompt from a function. The function can return: @@ -147,8 +204,8 @@ class Prompt(BaseModel): name=func_name, description=description, arguments=arguments, - fn=fn, tags=tags or set(), + fn=fn, ) async def render( @@ -212,25 +269,3 @@ class Prompt(BaseModel): except Exception as e: logger.exception(f"Error rendering prompt {self.name}: {e}") raise PromptError(f"Error rendering prompt {self.name}.") - - def __eq__(self, other: object) -> bool: - if not isinstance(other, Prompt): - return False - return self.model_dump() == other.model_dump() - - def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt: - """Convert the prompt to an MCP prompt.""" - arguments = [ - MCPPromptArgument( - name=arg.name, - description=arg.description, - required=arg.required, - ) - for arg in self.arguments or [] - ] - kwargs = { - "name": self.name, - "description": self.description, - "arguments": arguments, - } - return MCPPrompt(**kwargs | overrides) diff --git a/src/fastmcp/prompts/prompt_manager.py b/src/fastmcp/prompts/prompt_manager.py index bfd96b573..e0760be29 100644 --- a/src/fastmcp/prompts/prompt_manager.py +++ b/src/fastmcp/prompts/prompt_manager.py @@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any from mcp import GetPromptResult from fastmcp.exceptions import NotFoundError, PromptError -from fastmcp.prompts.prompt import Prompt, PromptResult +from fastmcp.prompts.prompt import FunctionPrompt, Prompt, PromptResult from fastmcp.settings import DuplicateBehavior from fastmcp.utilities.logging import get_logger @@ -55,10 +55,12 @@ class PromptManager: name: str | None = None, description: str | None = None, tags: set[str] | None = None, - ) -> Prompt: + ) -> FunctionPrompt: """Create a prompt from a function.""" - prompt = Prompt.from_function(fn, name=name, description=description, tags=tags) - return self.add_prompt(prompt) + prompt = FunctionPrompt.from_function( + fn, name=name, description=description, tags=tags + ) + return self.add_prompt(prompt) # type: ignore def add_prompt(self, prompt: Prompt, key: str | None = None) -> Prompt: """Add a prompt to the manager.""" diff --git a/src/fastmcp/server/proxy.py b/src/fastmcp/server/proxy.py index 96d5d1a41..603826c0f 100644 --- a/src/fastmcp/server/proxy.py +++ b/src/fastmcp/server/proxy.py @@ -153,6 +153,8 @@ class ProxyTemplate(ResourceTemplate): class ProxyPrompt(Prompt): + _client: Client + def __init__(self, client: Client, **kwargs): super().__init__(**kwargs) self._client = client @@ -164,7 +166,6 @@ class ProxyPrompt(Prompt): name=prompt.name, description=prompt.description, arguments=[a.model_dump() for a in prompt.arguments or []], - fn=_proxy_passthrough, ) async def render(self, arguments: dict[str, Any]) -> list[PromptMessage]: diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index 0729efbb2..c9e2a3d43 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -55,7 +55,7 @@ from fastmcp.server.http import ( create_streamable_http_app, ) from fastmcp.tools import ToolManager -from fastmcp.tools.tool import FunctionTool, Tool +from fastmcp.tools.tool import 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) - tool = FunctionTool.from_function( + tool = Tool.from_function( fn, name=name, description=description, diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py index b95a0b579..434ba9748 100644 --- a/src/fastmcp/tools/tool.py +++ b/src/fastmcp/tools/tool.py @@ -2,7 +2,6 @@ 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 @@ -66,12 +65,25 @@ class Tool(FastMCPBaseModel, ABC): 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." + def from_function( + fn: Callable[..., Any], + name: str | None = None, + description: str | None = None, + tags: set[str] | None = None, + annotations: ToolAnnotations | None = None, + exclude_args: list[str] | None = None, + serializer: Callable[[Any], str] | None = None, + ) -> FunctionTool: + """Create a Tool from a function.""" + return FunctionTool.from_function( + fn=fn, + name=name, + description=description, + tags=tags, + annotations=annotations, + exclude_args=exclude_args, + serializer=serializer, ) - return FunctionTool.from_function(fn, **overrides) def __eq__(self, other: object) -> bool: if not isinstance(other, Tool): diff --git a/src/fastmcp/tools/tool_manager.py b/src/fastmcp/tools/tool_manager.py index b5000a840..e89ba0ef9 100644 --- a/src/fastmcp/tools/tool_manager.py +++ b/src/fastmcp/tools/tool_manager.py @@ -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 FunctionTool, Tool +from fastmcp.tools.tool import 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 = FunctionTool.from_function( + tool = Tool.from_function( fn, name=name, description=description, diff --git a/tests/deprecated/test_tool_from_function_deprecated.py b/tests/deprecated/test_tool_from_function_deprecated.py deleted file mode 100644 index 3c81d6616..000000000 --- a/tests/deprecated/test_tool_from_function_deprecated.py +++ /dev/null @@ -1,76 +0,0 @@ -"""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 diff --git a/tests/prompts/test_prompt_manager.py b/tests/prompts/test_prompt_manager.py index 51710792c..c33751061 100644 --- a/tests/prompts/test_prompt_manager.py +++ b/tests/prompts/test_prompt_manager.py @@ -5,7 +5,7 @@ import pytest from fastmcp import Context from fastmcp.exceptions import NotFoundError, PromptError from fastmcp.prompts import Prompt -from fastmcp.prompts.prompt import PromptMessage, TextContent +from fastmcp.prompts.prompt import FunctionPrompt, PromptMessage, TextContent from fastmcp.prompts.prompt_manager import PromptManager @@ -97,6 +97,7 @@ class TestPromptManager: # Should have replaced with the new prompt prompt = manager.get_prompt("test_prompt") assert prompt is not None + assert isinstance(prompt, FunctionPrompt) assert prompt.fn.__name__ == "replacement_fn" def test_ignore_duplicate_prompts(self): @@ -118,8 +119,10 @@ class TestPromptManager: # Should keep the original prompt = manager.get_prompt("test_prompt") assert prompt is not None + assert isinstance(prompt, FunctionPrompt) assert prompt.fn.__name__ == "original_fn" # Result should be the original prompt + assert isinstance(result, FunctionPrompt) assert result.fn.__name__ == "original_fn" def test_get_prompts(self): diff --git a/tests/tools/test_tool.py b/tests/tools/test_tool.py index fb4a95b64..97f6fdf81 100644 --- a/tests/tools/test_tool.py +++ b/tests/tools/test_tool.py @@ -5,7 +5,7 @@ from pydantic import AnyUrl, BaseModel from fastmcp import FastMCP, Image from fastmcp.client import Client from fastmcp.exceptions import ToolError -from fastmcp.tools.tool import FunctionTool, _convert_to_content +from fastmcp.tools.tool import Tool, _convert_to_content from fastmcp.utilities.tests import temporary_settings @@ -17,7 +17,7 @@ class TestToolFromFunction: """Add two numbers.""" return a + b - tool = FunctionTool.from_function(add) + tool = Tool.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 = FunctionTool.from_function(fetch_data) + tool = Tool.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 = FunctionTool.from_function(Adder()) + tool = Tool.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 = FunctionTool.from_function(Adder()) + tool = Tool.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 = FunctionTool.from_function(create_user) + tool = Tool.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 = FunctionTool.from_function(image_tool) + tool = Tool.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"): - FunctionTool.from_function(1) # type: ignore + Tool.from_function(1) # type: ignore def test_lambda(self): - tool = FunctionTool.from_function(lambda x: x, name="my_tool") + tool = Tool.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" ): - FunctionTool.from_function(lambda x: x) + Tool.from_function(lambda x: x) def test_private_arguments(self): def add(_a: int, _b: int) -> int: """Add two numbers.""" return _a + _b - tool = FunctionTool.from_function(add) + tool = Tool.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" ): - FunctionTool.from_function(func) + Tool.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" ): - FunctionTool.from_function(func) + Tool.from_function(func) async def test_instance_method(self): class MyClass: @@ -148,7 +148,7 @@ class TestToolFromFunction: obj = MyClass() - tool = FunctionTool.from_function(obj.add) + tool = Tool.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" ): - FunctionTool.from_function(obj.add) + Tool.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" ): - FunctionTool.from_function(obj.add) + Tool.from_function(obj.add) async def test_classmethod(self): class MyClass: @@ -188,7 +188,7 @@ class TestToolFromFunction: """Add two numbers.""" return x + y - tool = FunctionTool.from_function(MyClass.call) + tool = Tool.from_function(MyClass.call) assert tool.name == "call" assert tool.description == "Add two numbers." assert "x" in tool.parameters["properties"] @@ -203,7 +203,7 @@ class TestToolFromFunction: def process_list(items: list[int]) -> int: return sum(items) - tool = FunctionTool.from_function(process_list, serializer=custom_serializer) + tool = Tool.from_function(process_list, serializer=custom_serializer) result = await tool.run(arguments={"items": [1, 2, 3, 4, 5]}) assert isinstance(result[0], TextContent) @@ -225,7 +225,7 @@ class TestLegacyToolJsonParsing: return f"{x}-{','.join(y)}" # Create a tool to use its JSON pre-parsing logic - tool = FunctionTool.from_function(simple_func) + tool = Tool.from_function(simple_func) # Prepare arguments where some are JSON strings json_args = { @@ -243,7 +243,7 @@ class TestLegacyToolJsonParsing: def func_with_str_types(str_or_list: str | list[str]) -> str | list[str]: return str_or_list - tool = FunctionTool.from_function(func_with_str_types) + tool = Tool.from_function(func_with_str_types) # Test regular string input (should remain a string) result = await tool.run({"str_or_list": "hello"}) @@ -269,7 +269,7 @@ class TestLegacyToolJsonParsing: def func_with_str_types(string: str) -> str: return string - tool = FunctionTool.from_function(func_with_str_types) + tool = Tool.from_function(func_with_str_types) # Invalid JSON should remain a string invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}" @@ -284,7 +284,7 @@ class TestLegacyToolJsonParsing: ) -> str | dict[int, str] | None: return string - tool = FunctionTool.from_function(func_with_str_types) + tool = Tool.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}" @@ -301,7 +301,7 @@ class TestLegacyToolJsonParsing: def func_with_complex_type(data: SomeModel) -> SomeModel: return data - tool = FunctionTool.from_function(func_with_complex_type) + tool = Tool.from_function(func_with_complex_type) # Valid JSON for the model valid_json = '{"x": 1, "y": {"1": "hello"}}' diff --git a/tests/tools/test_tool_manager.py b/tests/tools/test_tool_manager.py index 8f2bb32fa..9117d200a 100644 --- a/tests/tools/test_tool_manager.py +++ b/tests/tools/test_tool_manager.py @@ -568,7 +568,7 @@ class TestContextHandling: def test_context_parameter_detection(self): """Test that context parameters are properly detected in - FunctionTool.from_function().""" + Tool.from_function().""" def tool_with_context(x: int, ctx: Context) -> str: return str(x) @@ -634,7 +634,7 @@ class TestContextHandling: def test_parameterized_context_parameter_detection(self): """Test that context parameters are properly detected in - FunctionTool.from_function().""" + Tool.from_function().""" def tool_with_context(x: int, ctx: Context) -> str: return str(x) @@ -651,7 +651,7 @@ class TestContextHandling: def test_parameterized_union_context_parameter_detection(self): """Test that context parameters are properly detected in - FunctionTool.from_function().""" + Tool.from_function().""" def tool_with_context(x: int, ctx: Context | None) -> str: return str(x)