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

Merge pull request #701 from jlowin/prompt-strict

Browse source 
Use strict basemodel for Prompt; relax from_function deprecation
This commit is contained in:
Jeremiah Lowin 2025-06-04 10:53:58 -04:00 • committed by GitHub
commit 7cf2adc136
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 122 additions and 145 deletions
Download patch file Download diff file Expand all files Collapse all files

89
src/fastmcp/prompts/prompt.py
View file

@ -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)

10
src/fastmcp/prompts/prompt_manager.py
View file

@ -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."""

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

@ -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]:

4
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 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,

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

@ -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):

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 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,

76
tests/deprecated/test_tool_from_function_deprecated.py
View file

@ -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

5
tests/prompts/test_prompt_manager.py
View file

@ -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):

46
tests/tools/test_tool.py
View file

@ -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"}}'

6
tests/tools/test_tool_manager.py
View file

@ -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)