mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 06:24:18 +02:00
Remove DecoratedFunction
This commit is contained in:
Jeremiah Lowin
parent
c3a1010d9b
commit
31fd9abee0
3 changed files with 1 additions and 330 deletions
|
|
@ -57,7 +57,6 @@ from fastmcp.server.http import (
|
|||
from fastmcp.tools import ToolManager
|
||||
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
|
||||
from fastmcp.utilities.mcp_config import MCPConfig
|
||||
|
||||
|
|
@ -868,12 +867,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
)
|
||||
self.add_prompt(prompt)
|
||||
|
||||
# If name is provided, this is a direct call, return original function for consistency with tools
|
||||
# If name is None, this is @prompt without parens, return DecoratedFunction for proper method handling
|
||||
if name is not None:
|
||||
return fn # Direct function call
|
||||
else:
|
||||
return DecoratedFunction(fn) # Decorator usage
|
||||
return fn
|
||||
|
||||
elif isinstance(name_or_fn, str):
|
||||
# Case 3: @prompt("custom_name") - name passed as first argument
|
||||
|
|
|
|||
|
|
@ -1,101 +0,0 @@
|
|||
import inspect
|
||||
from collections.abc import Callable
|
||||
from typing import Generic, ParamSpec, TypeVar, cast, overload
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
R = TypeVar("R")
|
||||
P = ParamSpec("P")
|
||||
|
||||
|
||||
class DecoratedFunction(Generic[P, R]):
|
||||
"""Descriptor for decorated functions.
|
||||
|
||||
You can return this object from a decorator to ensure that it works across
|
||||
all types of functions: vanilla, instance methods, class methods, and static
|
||||
methods; both synchronous and asynchronous.
|
||||
|
||||
This class is used to store the original function and metadata about how to
|
||||
register it as a tool.
|
||||
|
||||
Example usage:
|
||||
|
||||
```python
|
||||
def my_decorator(fn: Callable[P, R]) -> DecoratedFunction[P, R]:
|
||||
return DecoratedFunction(fn)
|
||||
```
|
||||
|
||||
On a function:
|
||||
```python
|
||||
@my_decorator
|
||||
def my_function(a: int, b: int) -> int:
|
||||
return a + b
|
||||
```
|
||||
|
||||
On an instance method:
|
||||
```python
|
||||
class Test:
|
||||
@my_decorator
|
||||
def my_function(self, a: int, b: int) -> int:
|
||||
return a + b
|
||||
```
|
||||
|
||||
On a class method:
|
||||
```python
|
||||
class Test:
|
||||
@classmethod
|
||||
@my_decorator
|
||||
def my_function(cls, a: int, b: int) -> int:
|
||||
return a + b
|
||||
```
|
||||
|
||||
Note that for classmethods, the decorator must be applied first, then
|
||||
`@classmethod` on top.
|
||||
|
||||
On a static method:
|
||||
```python
|
||||
class Test:
|
||||
@staticmethod
|
||||
@my_decorator
|
||||
def my_function(a: int, b: int) -> int:
|
||||
return a + b
|
||||
```
|
||||
"""
|
||||
|
||||
def __init__(self, fn: Callable[P, R]):
|
||||
self.fn = fn
|
||||
|
||||
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
|
||||
"""Call the original function."""
|
||||
try:
|
||||
return self.fn(*args, **kwargs)
|
||||
except TypeError as e:
|
||||
if "'classmethod' object is not callable" in str(e):
|
||||
raise TypeError(
|
||||
"To apply this decorator to a classmethod, apply the decorator first, then @classmethod on top."
|
||||
)
|
||||
raise
|
||||
|
||||
@overload
|
||||
def __get__(self, instance: None, owner: type | None = None) -> Self: ...
|
||||
|
||||
@overload
|
||||
def __get__(
|
||||
self, instance: object, owner: type | None = None
|
||||
) -> Callable[P, R]: ...
|
||||
|
||||
def __get__(
|
||||
self, instance: object | None, owner: type | None = None
|
||||
) -> Self | Callable[P, R]:
|
||||
"""Return the original function when accessed from an instance, or self when accessed from the class."""
|
||||
if instance is None:
|
||||
return self
|
||||
# Return the original function bound to the instance
|
||||
return cast(Callable[P, R], self.fn.__get__(instance, owner))
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Return a representation that matches Python's function representation."""
|
||||
module = getattr(self.fn, "__module__", "unknown")
|
||||
qualname = getattr(self.fn, "__qualname__", str(self.fn))
|
||||
sig_str = str(inspect.signature(self.fn))
|
||||
return f"<function {module}.{qualname}{sig_str}>"
|
||||
|
|
@ -1,222 +0,0 @@
|
|||
import functools
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from fastmcp.utilities.decorators import DecoratedFunction
|
||||
|
||||
DECORATOR_CALLED = []
|
||||
|
||||
|
||||
def decorator(fn: Callable[..., Any]) -> DecoratedFunction[..., Any]:
|
||||
@functools.wraps(fn)
|
||||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
DECORATOR_CALLED.append((args, kwargs))
|
||||
return fn(*args, **kwargs)
|
||||
|
||||
return DecoratedFunction(wrapper)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_decorator_called():
|
||||
DECORATOR_CALLED.clear()
|
||||
yield
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
|
||||
@decorator
|
||||
def add(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
|
||||
@decorator
|
||||
async def add_async(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
|
||||
class DecoratedClass:
|
||||
def __init__(self, x: int):
|
||||
self.x = x
|
||||
|
||||
@decorator
|
||||
def add(self, a: int, b: int) -> int:
|
||||
return a + b + self.x
|
||||
|
||||
@decorator
|
||||
async def add_async(self, a: int, b: int) -> int:
|
||||
return a + b + self.x
|
||||
|
||||
@classmethod
|
||||
@decorator
|
||||
def add_classmethod(cls, a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@staticmethod
|
||||
@decorator
|
||||
def add_staticmethod(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@classmethod
|
||||
@decorator
|
||||
async def add_classmethod_async(cls, a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@staticmethod
|
||||
@decorator
|
||||
async def add_staticmethod_async(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@decorator
|
||||
@classmethod
|
||||
def add_classmethod_reverse_decorator_order(cls, a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@decorator
|
||||
@staticmethod
|
||||
def add_staticmethod_reverse_decorator_order(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@decorator
|
||||
@classmethod
|
||||
async def add_classmethod_async_reverse_decorator_order(cls, a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
@decorator
|
||||
@staticmethod
|
||||
async def add_staticmethod_async_reverse_decorator_order(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
|
||||
def test_add():
|
||||
assert add(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((1, 2), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert add(a=3, b=4) == 7
|
||||
assert DECORATOR_CALLED == [((), {"a": 3, "b": 4})]
|
||||
|
||||
|
||||
async def test_add_async():
|
||||
assert await add_async(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((1, 2), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert await add_async(a=3, b=4) == 7
|
||||
assert DECORATOR_CALLED == [((), {"a": 3, "b": 4})]
|
||||
|
||||
|
||||
def test_instance_method():
|
||||
obj = DecoratedClass(10)
|
||||
assert obj.add(2, 3) == 15
|
||||
assert DECORATOR_CALLED == [((obj, 2, 3), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert obj.add(a=4, b=5) == 19
|
||||
assert DECORATOR_CALLED == [((obj,), {"a": 4, "b": 5})]
|
||||
|
||||
|
||||
async def test_instance_method_async():
|
||||
obj = DecoratedClass(10)
|
||||
assert await obj.add_async(2, 3) == 15
|
||||
assert DECORATOR_CALLED == [((obj, 2, 3), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert await obj.add_async(a=4, b=5) == 19
|
||||
assert DECORATOR_CALLED == [((obj,), {"a": 4, "b": 5})]
|
||||
|
||||
|
||||
def test_classmethod():
|
||||
assert DecoratedClass.add_classmethod(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((DecoratedClass, 1, 2), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert DecoratedClass.add_classmethod(a=3, b=4) == 7
|
||||
assert DECORATOR_CALLED == [((DecoratedClass,), {"a": 3, "b": 4})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test via instance
|
||||
obj = DecoratedClass(10)
|
||||
assert obj.add_classmethod(5, 6) == 11
|
||||
assert DECORATOR_CALLED == [((DecoratedClass, 5, 6), {})]
|
||||
|
||||
|
||||
async def test_classmethod_async():
|
||||
assert await DecoratedClass.add_classmethod_async(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((DecoratedClass, 1, 2), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert await DecoratedClass.add_classmethod_async(a=3, b=4) == 7
|
||||
assert DECORATOR_CALLED == [((DecoratedClass,), {"a": 3, "b": 4})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test via instance
|
||||
obj = DecoratedClass(10)
|
||||
assert await obj.add_classmethod_async(5, 6) == 11
|
||||
assert DECORATOR_CALLED == [((DecoratedClass, 5, 6), {})]
|
||||
|
||||
|
||||
def test_classmethod_wrong_order():
|
||||
with pytest.raises(
|
||||
TypeError,
|
||||
match="To apply this decorator to a classmethod, apply the decorator first, then @classmethod on top.",
|
||||
):
|
||||
DecoratedClass.add_classmethod_reverse_decorator_order(1, 2)
|
||||
|
||||
|
||||
async def test_classmethod_async_wrong_order():
|
||||
with pytest.raises(
|
||||
TypeError,
|
||||
match="To apply this decorator to a classmethod, apply the decorator first, then @classmethod on top.",
|
||||
):
|
||||
await DecoratedClass.add_classmethod_async_reverse_decorator_order(1, 2)
|
||||
|
||||
|
||||
def test_staticmethod():
|
||||
assert DecoratedClass.add_staticmethod(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((1, 2), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert DecoratedClass.add_staticmethod(a=3, b=4) == 7
|
||||
assert DECORATOR_CALLED == [((), {"a": 3, "b": 4})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test via instance
|
||||
obj = DecoratedClass(10)
|
||||
assert obj.add_staticmethod(5, 6) == 11
|
||||
assert DECORATOR_CALLED == [((5, 6), {})]
|
||||
|
||||
|
||||
async def test_staticmethod_async():
|
||||
assert await DecoratedClass.add_staticmethod_async(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((1, 2), {})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test with keyword arguments
|
||||
assert await DecoratedClass.add_staticmethod_async(a=3, b=4) == 7
|
||||
assert DECORATOR_CALLED == [((), {"a": 3, "b": 4})]
|
||||
DECORATOR_CALLED.clear()
|
||||
|
||||
# Test via instance
|
||||
obj = DecoratedClass(10)
|
||||
assert await obj.add_staticmethod_async(5, 6) == 11
|
||||
assert DECORATOR_CALLED == [((5, 6), {})]
|
||||
|
||||
|
||||
def test_staticmethod_wrong_order():
|
||||
assert DecoratedClass.add_staticmethod_reverse_decorator_order(1, 2) == 3
|
||||
assert DECORATOR_CALLED == [((1, 2), {})]
|
||||
|
||||
|
||||
async def test_staticmethod_async_wrong_order():
|
||||
assert (
|
||||
await DecoratedClass.add_staticmethod_async_reverse_decorator_order(1, 2) == 3
|
||||
)
|
||||
assert DECORATOR_CALLED == [((1, 2), {})]
|
||||
Loading…
Add table
Add a link
Reference in a new issue