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

Decorators return functions instead of component objects (#2856)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-12 21:58:07 -05:00 • committed by GitHub
commit 1b723f302d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
38 changed files with 2406 additions and 1606 deletions
Download patch file Download diff file Expand all files Collapse all files

39
docs/development/upgrade-guide.mdx
View file

@ -14,6 +14,45 @@ This guide provides migration instructions for breaking changes and major update
The deprecated `WSTransport` client transport has been removed. Use `StreamableHttpTransport` instead.
### Decorators Return Functions
<Warning>
**Breaking Change:** Decorators now return your original function instead of a component object. Code that treats the decorated function as a `FunctionTool`, `FunctionResource`, or `FunctionPrompt` will break.
</Warning>
Decorators (`@tool`, `@resource`, `@prompt`) now return the original function instead of transforming it into a component object:
<CodeGroup>
```python Before
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
isinstance(greet, FunctionTool) # True
greet.name # "greet"
```
```python After
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
isinstance(greet, FunctionTool) # False - it's your function now
greet("World") # "Hello, World!" - still callable
```
</CodeGroup>
**Why this changed:** Functions staying as functions means they're directly callable for testing, work naturally with instance methods, and match how Flask/FastAPI decorators behave.
**For v2 compatibility:**
```python
import fastmcp
fastmcp.settings.decorator_mode = "object"
```
Or set the environment variable `FASTMCP_DECORATOR_MODE=object`.
### Provider Architecture
FastMCP v3 introduces a unified provider architecture for sourcing components. All tools, resources, and prompts now flow through providers:

56
docs/development/v3-notes/v3-features.mdx
View file

@ -238,6 +238,38 @@ Requires Docket server for task scheduling and result polling.
---
## Decorators Return Functions
v3.0 changes what decorators (`@tool`, `@resource`, `@prompt`) return. Decorators now return the original function unchanged, rather than transforming it into a component object.
**v3 behavior (default):**
```python
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
# greet is still your function - call it directly
greet("World") # "Hello, World!"
```
**Why this matters:**
- Functions stay callable - useful for testing and reuse
- Instance methods just work: `mcp.add_tool(obj.method)`
- Matches how Flask, FastAPI, and Typer decorators behave
**For v2 compatibility:**
```python
import fastmcp
# v2 behavior: decorators return FunctionTool/FunctionResource/FunctionPrompt objects
fastmcp.settings.decorator_mode = "object"
```
Environment variable: `FASTMCP_DECORATOR_MODE=object`
---
## CLI Auto-Reload
The `--reload` flag enables file watching with automatic server restarts for development.
@ -361,6 +393,30 @@ The `tool_serializer` parameter on `FastMCP` is deprecated. Return `ToolResult`
The deprecated `WSTransport` client transport has been removed. Use `StreamableHttpTransport` instead.
### Decorators Return Functions
Decorators (`@tool`, `@resource`, `@prompt`) now return the original function instead of component objects. Code that treats the decorated function as a `FunctionTool`, `FunctionResource`, or `FunctionPrompt` will break.
```python
# v2.x
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
isinstance(greet, FunctionTool) # True
# v3.0
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
isinstance(greet, FunctionTool) # False
callable(greet) # True - it's still your function
greet("World") # "Hello, World!"
```
Set `FASTMCP_DECORATOR_MODE=object` or `fastmcp.settings.decorator_mode = "object"` for v2 behavior.
### Component Enable/Disable Moved to Server/Provider
The `enabled` field and `enable()`/`disable()` methods removed from component objects:

1
docs/docs.json
View file

@ -264,7 +264,6 @@
"group": "Patterns",
"pages": [
"patterns/tool-transformation",
"patterns/decorating-methods",
"patterns/cli",
"patterns/contrib",
"patterns/testing"

225
docs/patterns/decorating-methods.mdx
View file

@ -1,225 +0,0 @@
---
title: Decorating Methods
sidebarTitle: Decorating Methods
description: Properly use instance methods, class methods, and static methods with FastMCP decorators.
icon: at
---
FastMCP's decorator system is designed to work with functions, but you may see unexpected behavior if you try to decorate an instance or class method. This guide explains the correct approach for using methods with all FastMCP decorators (`@tool`, `@resource`, and `@prompt`).
## Why Are Methods Hard?
When you apply a FastMCP decorator like `@tool`, `@resource`, or `@prompt` to a method, the decorator captures the function at decoration time. For instance methods and class methods, this poses a challenge because:
1. For instance methods: The decorator gets the unbound method before any instance exists
2. For class methods: The decorator gets the function before it's bound to the class
This means directly decorating these methods doesn't work as expected. In practice, the LLM would see parameters like `self` or `cls` that it cannot provide values for.
Additionally, **FastMCP decorators return objects (Tool, Resource, or Prompt instances) rather than the original function**. This means that when you decorate a method directly, the method becomes the returned object and is no longer callable by your code:
<Warning>
**Don't do this!**
The method will no longer be callable from Python, and the tool won't be callable by LLMs.
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@mcp.tool
def my_method(self, x: int) -> int:
return x * 2
obj = MyClass()
obj.my_method(5) # Fails - my_method is a Tool, not a function
```
</Warning>
This is another important reason to register methods functionally after defining the class.
## Recommended Patterns
### Instance Methods
<Warning>
**Don't do this!**
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@mcp.tool # This won't work correctly
def add(self, x, y):
return x + y
```
</Warning>
When the decorator is applied this way, it captures the unbound method. When the LLM later tries to use this component, it will see `self` as a required parameter, but it won't know what to provide for it, causing errors or unexpected behavior.
<Check>
**Do this instead**:
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
def add(self, x, y):
return x + y
# Create an instance first, then register the bound methods
obj = MyClass()
mcp.tool(obj.add)
# Now you can call it without 'self' showing up as a parameter
await mcp._mcp_call_tool('add', {'x': 1, 'y': 2}) # Returns 3
```
</Check>
This approach works because:
1. You first create an instance of the class (`obj`)
2. When you access the method through the instance (`obj.add`), Python creates a bound method where `self` is already set to that instance
3. When you register this bound method, the system sees a callable that only expects the appropriate parameters, not `self`
### Class Methods
The behavior of decorating class methods depends on the order of decorators:
<Warning>
**Don't do this** (decorator order matters):
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@classmethod
@mcp.tool # This won't work but won't raise an error
def from_string_v1(cls, s):
return cls(s)
@mcp.tool
@classmethod # This will raise a helpful ValueError
def from_string_v2(cls, s):
return cls(s)
```
</Warning>
- If `@classmethod` comes first, then `@mcp.tool`: No error is raised, but it won't work correctly
- If `@mcp.tool` comes first, then `@classmethod`: FastMCP will detect this and raise a helpful `ValueError` with guidance
<Check>
**Do this instead**:
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@classmethod
def from_string(cls, s):
return cls(s)
# Register the class method after the class is defined
mcp.tool(MyClass.from_string)
```
</Check>
This works because:
1. The `@classmethod` decorator is applied properly during class definition
2. When you access `MyClass.from_string`, Python provides a special method object that automatically binds the class to the `cls` parameter
3. When registered, only the appropriate parameters are exposed to the LLM, hiding the implementation detail of the `cls` parameter
### Static Methods
Static methods "work" with FastMCP decorators, but this is not recommended because the FastMCP decorator will not return a callable method. Therefore, you should register static methods the same way as other methods.
<Warning>
**This is not recommended, though it will work.**
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@mcp.tool
@staticmethod
def utility(x, y):
return x + y
```
</Warning>
This works because `@staticmethod` converts the method to a regular function, which the FastMCP decorator can then properly process. However, this is not recommended because the FastMCP decorator will not return a callable staticmethod. Therefore, you should register static methods the same way as other methods.
<Check>
**Prefer this pattern:**
```python
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@staticmethod
def utility(x, y):
return x + y
# This also works
mcp.tool(MyClass.utility)
```
</Check>
## Additional Patterns
### Creating Components at Class Initialization
You can automatically register instance methods when creating an object:
```python
from fastmcp import FastMCP
mcp = FastMCP()
class ComponentProvider:
def __init__(self, mcp_instance):
# Register methods
mcp_instance.tool(self.tool_method)
mcp_instance.resource("resource://data")(self.resource_method)
def tool_method(self, x):
return x * 2
def resource_method(self):
return "Resource data"
# The methods are automatically registered when creating the instance
provider = ComponentProvider(mcp)
```
This pattern is useful when:
- You want to encapsulate registration logic within the class itself
- You have multiple related components that should be registered together
- You want to ensure that methods are always properly registered when creating an instance
The class automatically registers its methods during initialization, ensuring they're properly bound to the instance before registration.
## Summary
The current behavior of FastMCP decorators with methods is:
- **Static methods**: Can be decorated directly and work perfectly with all FastMCP decorators
- **Class methods**: Cannot be decorated directly and will raise a helpful `ValueError` with guidance
- **Instance methods**: Should be registered after creating an instance using the decorator calls
For class and instance methods, you should register them after creating the instance or class to ensure proper method binding. This ensures that the methods are properly bound before being registered.
Understanding these patterns allows you to effectively organize your components into classes while maintaining proper method binding, giving you the benefits of object-oriented design without sacrificing the simplicity of FastMCP's decorator system.

4
docs/servers/prompts.mdx
View file

@ -113,6 +113,10 @@ def data_analysis_prompt(
</ParamField>
</Card>
#### Using with Methods
For decorating instance or class methods, use the standalone `@prompt` decorator and register the bound method. See [Tools: Using with Methods](/servers/tools#using-with-methods) for the pattern.
### Argument Types
<VersionBadge version="2.9.0" />

4
docs/servers/resources.mdx
View file

@ -132,6 +132,10 @@ def get_application_status() -> str:
</ParamField>
</Card>
#### Using with Methods
For decorating instance or class methods, use the standalone `@resource` decorator and register the bound method. See [Tools: Using with Methods](/servers/tools#using-with-methods) for the pattern.
### Return Values
Resource functions must return one of three types:

21
docs/servers/tools.mdx
View file

@ -116,6 +116,27 @@ def search_products_implementation(query: str, category: str | None = None) -> l
</ParamField>
</Card>
### Using with Methods
The `@mcp.tool` decorator registers tools immediately, which doesn't work with instance or class methods (you'd see `self` or `cls` as required parameters). For methods, use the standalone `@tool` decorator to attach metadata, then register the bound method:
```python
from fastmcp import FastMCP
from fastmcp.tools import tool
class Calculator:
def __init__(self, multiplier: int):
self.multiplier = multiplier
@tool()
def multiply(self, x: int) -> int:
"""Multiply x by the instance multiplier."""
return x * self.multiplier
calc = Calculator(multiplier=3)
mcp = FastMCP()
mcp.add_tool(calc.multiply) # Registers with correct schema (only 'x', not 'self')
```
### Async Support

41
src/fastmcp/decorators.py Normal file
View file

@ -0,0 +1,41 @@
"""Shared decorator utilities for FastMCP."""
from __future__ import annotations
import inspect
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
if TYPE_CHECKING:
from fastmcp.prompts.function_prompt import PromptMeta
from fastmcp.resources.function_resource import ResourceMeta
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.function_tool import ToolMeta
FastMCPMeta = ToolMeta | ResourceMeta | PromptMeta
def resolve_task_config(task: bool | TaskConfig | None) -> bool | TaskConfig:
"""Resolve task config, defaulting None to False."""
return task if task is not None else False
@runtime_checkable
class HasFastMCPMeta(Protocol):
"""Protocol for callables decorated with FastMCP metadata."""
__fastmcp__: Any
def get_fastmcp_meta(fn: Any) -> Any | None:
"""Extract FastMCP metadata from a function, handling bound methods and wrappers."""
if hasattr(fn, "__fastmcp__"):
return fn.__fastmcp__
if hasattr(fn, "__func__") and hasattr(fn.__func__, "__fastmcp__"):
return fn.__func__.__fastmcp__
try:
unwrapped = inspect.unwrap(fn)
if unwrapped is not fn and hasattr(unwrapped, "__fastmcp__"):
return unwrapped.__fastmcp__
except ValueError:
pass
return None

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

@ -1,9 +1,11 @@
from .prompt import FunctionPrompt, Message, Prompt, PromptMessage, PromptResult, prompt
from .function_prompt import FunctionPrompt, prompt
from .prompt import Message, Prompt, PromptArgument, PromptMessage, PromptResult
__all__ = [
"FunctionPrompt",
"Message",
"Prompt",
"PromptArgument",
"PromptMessage",
"PromptResult",
"prompt",

449
src/fastmcp/prompts/function_prompt.py Normal file
View file

@ -0,0 +1,449 @@
"""Standalone @prompt decorator for FastMCP."""
from __future__ import annotations
import inspect
import json
import warnings
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Any,
Literal,
Protocol,
TypeVar,
overload,
runtime_checkable,
)
import pydantic_core
from mcp.types import Icon
import fastmcp
from fastmcp.decorators import resolve_task_config
from fastmcp.exceptions import PromptError
from fastmcp.prompts.prompt import Prompt, PromptArgument, PromptResult
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.tool import AuthCheckCallable
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import get_cached_typeadapter
if TYPE_CHECKING:
from docket import Docket
from docket.execution import Execution
F = TypeVar("F", bound=Callable[..., Any])
logger = get_logger(__name__)
@runtime_checkable
class DecoratedPrompt(Protocol):
"""Protocol for functions decorated with @prompt."""
__fastmcp__: PromptMeta
def __call__(self, *args: Any, **kwargs: Any) -> Any: ...
@dataclass(frozen=True, kw_only=True)
class PromptMeta:
"""Metadata attached to functions by the @prompt decorator."""
type: Literal["prompt"] = field(default="prompt", init=False)
name: str | None = None
title: str | None = None
description: str | None = None
icons: list[Icon] | None = None
tags: set[str] | None = None
meta: dict[str, Any] | None = None
task: bool | TaskConfig | None = None
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None
class FunctionPrompt(Prompt):
"""A prompt that is a function."""
fn: Callable[..., Any]
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
*,
metadata: PromptMeta | None = None,
# Keep individual params for backwards compat
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionPrompt:
"""Create a Prompt from a function.
Args:
fn: The function to wrap
metadata: PromptMeta object with all configuration. If provided,
individual parameters must not be passed.
name, title, etc.: Individual parameters for backwards compatibility.
Cannot be used together with metadata parameter.
The function can return:
- str: wrapped as single user Message
- list[Message | str]: converted to list[Message]
- PromptResult: used directly
"""
# Check mutual exclusion
individual_params_provided = any(
x is not None
for x in [name, title, description, icons, tags, meta, task, auth]
)
if metadata is not None and individual_params_provided:
raise TypeError(
"Cannot pass both 'metadata' and individual parameters to from_function(). "
"Use metadata alone or individual parameters alone."
)
# Build metadata from kwargs if not provided
if metadata is None:
metadata = PromptMeta(
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=task,
auth=auth,
)
func_name = (
metadata.name or getattr(fn, "__name__", None) or fn.__class__.__name__
)
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
# Reject functions with *args or **kwargs
sig = inspect.signature(fn)
for param in sig.parameters.values():
if param.kind == inspect.Parameter.VAR_POSITIONAL:
raise ValueError("Functions with *args are not supported as prompts")
if param.kind == inspect.Parameter.VAR_KEYWORD:
raise ValueError("Functions with **kwargs are not supported as prompts")
description = metadata.description or inspect.getdoc(fn)
# Normalize task to TaskConfig and validate
task_value = metadata.task
if task_value is None:
task_config = TaskConfig(mode="forbidden")
elif isinstance(task_value, bool):
task_config = TaskConfig.from_bool(task_value)
else:
task_config = task_value
task_config.validate_function(fn, func_name)
# if the fn is a callable class, we need to get the __call__ method from here out
if not inspect.isroutine(fn):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__ # type: ignore[assignment]
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Wrap fn to handle dependency resolution internally
wrapped_fn = without_injected_parameters(fn)
type_adapter = get_cached_typeadapter(wrapped_fn)
parameters = type_adapter.json_schema()
parameters = compress_schema(parameters, prune_titles=True)
# Convert parameters to PromptArguments
arguments: list[PromptArgument] = []
if "properties" in parameters:
for param_name, param in parameters["properties"].items():
arg_description = param.get("description")
# For non-string parameters, append JSON schema info to help users
# understand the expected format when passing as strings (MCP requirement)
if param_name in sig.parameters:
sig_param = sig.parameters[param_name]
if (
sig_param.annotation != inspect.Parameter.empty
and sig_param.annotation is not str
):
# Get the JSON schema for this specific parameter type
try:
param_adapter = get_cached_typeadapter(sig_param.annotation)
param_schema = param_adapter.json_schema()
# Create compact schema representation
schema_str = json.dumps(param_schema, separators=(",", ":"))
# Append schema info to description
schema_note = f"Provide as a JSON string matching the following schema: {schema_str}"
if arg_description:
arg_description = f"{arg_description}\n\n{schema_note}"
else:
arg_description = schema_note
except Exception as e:
# If schema generation fails, skip enhancement
logger.debug(
"Failed to generate schema for prompt argument %s: %s",
param_name,
e,
)
arguments.append(
PromptArgument(
name=param_name,
description=arg_description,
required=param_name in parameters.get("required", []),
)
)
return cls(
name=func_name,
title=metadata.title,
description=description,
icons=metadata.icons,
arguments=arguments,
tags=metadata.tags or set(),
fn=wrapped_fn,
meta=metadata.meta,
task_config=task_config,
auth=metadata.auth,
)
def _convert_string_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
"""Convert string arguments to expected types based on function signature."""
from fastmcp.server.dependencies import without_injected_parameters
wrapper_fn = without_injected_parameters(self.fn)
sig = inspect.signature(wrapper_fn)
converted_kwargs = {}
for param_name, param_value in kwargs.items():
if param_name in sig.parameters:
param = sig.parameters[param_name]
# If parameter has no annotation or annotation is str, pass as-is
if (
param.annotation == inspect.Parameter.empty
or param.annotation is str
) or not isinstance(param_value, str):
converted_kwargs[param_name] = param_value
else:
# Try to convert string argument using type adapter
try:
adapter = get_cached_typeadapter(param.annotation)
# Try JSON parsing first for complex types
try:
converted_kwargs[param_name] = adapter.validate_json(
param_value
)
except (ValueError, TypeError, pydantic_core.ValidationError):
# Fallback to direct validation
converted_kwargs[param_name] = adapter.validate_python(
param_value
)
except (ValueError, TypeError, pydantic_core.ValidationError) as e:
# If conversion fails, provide informative error
raise PromptError(
f"Could not convert argument '{param_name}' with value '{param_value}' "
f"to expected type {param.annotation}. Error: {e}"
) from e
else:
# Parameter not in function signature, pass as-is
converted_kwargs[param_name] = param_value
return converted_kwargs
async def render(
self,
arguments: dict[str, Any] | None = None,
) -> PromptResult:
"""Render the prompt with arguments."""
# Validate required arguments
if self.arguments:
required = {arg.name for arg in self.arguments if arg.required}
provided = set(arguments or {})
missing = required - provided
if missing:
raise ValueError(f"Missing required arguments: {missing}")
try:
# Prepare arguments
kwargs = arguments.copy() if arguments else {}
# Convert string arguments to expected types BEFORE validation
kwargs = self._convert_string_arguments(kwargs)
# self.fn is wrapped by without_injected_parameters which handles
# dependency resolution internally
result = self.fn(**kwargs)
if inspect.isawaitable(result):
result = await result
return self.convert_result(result)
except Exception as e:
logger.exception(f"Error rendering prompt {self.name}")
raise PromptError(f"Error rendering prompt {self.name}.") from e
def register_with_docket(self, docket: Docket) -> None:
"""Register this prompt with docket for background execution.
FunctionPrompt registers the underlying function, which has the user's
Depends parameters for docket to resolve.
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key]) # type: ignore[arg-type]
async def add_to_docket( # type: ignore[override]
self,
docket: Docket,
arguments: dict[str, Any] | None,
*,
fn_key: str | None = None,
task_key: str | None = None,
**kwargs: Any,
) -> Execution:
"""Schedule this prompt for background execution via docket.
FunctionPrompt splats the arguments dict since .fn expects **kwargs.
Args:
docket: The Docket instance
arguments: Prompt arguments
fn_key: Function lookup key in Docket registry (defaults to self.key)
task_key: Redis storage key for the result
**kwargs: Additional kwargs passed to docket.add()
"""
lookup_key = fn_key or self.key
if task_key:
kwargs["key"] = task_key
return await docket.add(lookup_key, **kwargs)(**(arguments or {}))
@overload
def prompt(fn: F) -> F: ...
@overload
def prompt(
name_or_fn: str,
*,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[F], F]: ...
@overload
def prompt(
name_or_fn: None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[F], F]: ...
def prompt(
name_or_fn: str | Callable[..., Any] | None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Any:
"""Standalone decorator to mark a function as an MCP prompt.
Returns the original function with metadata attached. Register with a server
using mcp.add_prompt().
"""
if isinstance(name_or_fn, classmethod):
raise TypeError(
"To decorate a classmethod, use @classmethod above @prompt. "
"See https://gofastmcp.com/servers/tools#using-with-methods"
)
def create_prompt(
fn: Callable[..., Any], prompt_name: str | None
) -> FunctionPrompt:
# Create metadata first, then pass it
prompt_meta = PromptMeta(
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=resolve_task_config(task),
auth=auth,
)
return FunctionPrompt.from_function(fn, metadata=prompt_meta)
def attach_metadata(fn: F, prompt_name: str | None) -> F:
metadata = PromptMeta(
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=task,
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
return fn
def decorator(fn: F, prompt_name: str | None) -> F:
if fastmcp.settings.decorator_mode == "object":
warnings.warn(
"decorator_mode='object' is deprecated and will be removed in a future version. "
"Decorators now return the original function with metadata attached.",
DeprecationWarning,
stacklevel=4,
)
return create_prompt(fn, prompt_name) # type: ignore[return-value]
return attach_metadata(fn, prompt_name)
if inspect.isroutine(name_or_fn):
return decorator(name_or_fn, name)
elif isinstance(name_or_fn, str):
if name is not None:
raise TypeError("Cannot specify name both as first argument and keyword")
prompt_name = name_or_fn
elif name_or_fn is None:
prompt_name = name
else:
raise TypeError(f"Invalid first argument: {type(name_or_fn)}")
def wrapper(fn: F) -> F:
return decorator(fn, prompt_name)
return wrapper

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

@ -2,10 +2,8 @@
from __future__ import annotations as _annotations
import inspect
import json
import warnings
from collections.abc import Callable
from functools import partial
from typing import TYPE_CHECKING, Any, ClassVar, Literal, overload
import pydantic
@ -14,6 +12,8 @@ import pydantic_core
if TYPE_CHECKING:
from docket import Docket
from docket.execution import Execution
from fastmcp.prompts.function_prompt import FunctionPrompt
import mcp.types
from mcp import GetPromptResult
from mcp.types import (
@ -26,19 +26,12 @@ from mcp.types import Prompt as SDKPrompt
from mcp.types import PromptArgument as SDKPromptArgument
from pydantic import Field
from fastmcp.exceptions import PromptError
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
from fastmcp.tools.tool import AuthCheckCallable
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
FastMCPBaseModel,
get_cached_typeadapter,
)
logger = get_logger(__name__)
@ -232,9 +225,11 @@ class Prompt(FastMCPComponent):
),
)
@staticmethod
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
@ -251,6 +246,8 @@ class Prompt(FastMCPComponent):
- list[Message | str]: converted to list[Message]
- PromptResult: used directly
"""
from fastmcp.prompts.function_prompt import FunctionPrompt
return FunctionPrompt.from_function(
fn=fn,
name=name,
@ -396,398 +393,33 @@ class Prompt(FastMCPComponent):
return await docket.add(lookup_key, **kwargs)(arguments)
class FunctionPrompt(Prompt):
"""A prompt that is a function."""
fn: Callable[..., Any]
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionPrompt:
"""Create a Prompt from a function.
The function can return:
- str: wrapped as single user Message
- list[Message | str]: converted to list[Message]
- PromptResult: used directly
"""
func_name = name or getattr(fn, "__name__", None) or fn.__class__.__name__
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
# Reject functions with *args or **kwargs
sig = inspect.signature(fn)
for param in sig.parameters.values():
if param.kind == inspect.Parameter.VAR_POSITIONAL:
raise ValueError("Functions with *args are not supported as prompts")
if param.kind == inspect.Parameter.VAR_KEYWORD:
raise ValueError("Functions with **kwargs are not supported as prompts")
description = description or inspect.getdoc(fn)
# Normalize task to TaskConfig and validate
if task is None:
task_config = TaskConfig(mode="forbidden")
elif isinstance(task, bool):
task_config = TaskConfig.from_bool(task)
else:
task_config = task
task_config.validate_function(fn, func_name)
# if the fn is a callable class, we need to get the __call__ method from here out
if not inspect.isroutine(fn):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__ # type: ignore[assignment]
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Wrap fn to handle dependency resolution internally
wrapped_fn = without_injected_parameters(fn)
type_adapter = get_cached_typeadapter(wrapped_fn)
parameters = type_adapter.json_schema()
parameters = compress_schema(parameters, prune_titles=True)
# Convert parameters to PromptArguments
arguments: list[PromptArgument] = []
if "properties" in parameters:
for param_name, param in parameters["properties"].items():
arg_description = param.get("description")
# For non-string parameters, append JSON schema info to help users
# understand the expected format when passing as strings (MCP requirement)
if param_name in sig.parameters:
sig_param = sig.parameters[param_name]
if (
sig_param.annotation != inspect.Parameter.empty
and sig_param.annotation is not str
):
# Get the JSON schema for this specific parameter type
try:
param_adapter = get_cached_typeadapter(sig_param.annotation)
param_schema = param_adapter.json_schema()
# Create compact schema representation
schema_str = json.dumps(param_schema, separators=(",", ":"))
# Append schema info to description
schema_note = f"Provide as a JSON string matching the following schema: {schema_str}"
if arg_description:
arg_description = f"{arg_description}\n\n{schema_note}"
else:
arg_description = schema_note
except Exception:
# If schema generation fails, skip enhancement
pass
arguments.append(
PromptArgument(
name=param_name,
description=arg_description,
required=param_name in parameters.get("required", []),
)
)
return cls(
name=func_name,
title=title,
description=description,
icons=icons,
arguments=arguments,
tags=tags or set(),
fn=wrapped_fn,
meta=meta,
task_config=task_config,
auth=auth,
)
def _convert_string_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
"""Convert string arguments to expected types based on function signature."""
from fastmcp.server.dependencies import without_injected_parameters
wrapper_fn = without_injected_parameters(self.fn)
sig = inspect.signature(wrapper_fn)
converted_kwargs = {}
for param_name, param_value in kwargs.items():
if param_name in sig.parameters:
param = sig.parameters[param_name]
# If parameter has no annotation or annotation is str, pass as-is
if (
param.annotation == inspect.Parameter.empty
or param.annotation is str
) or not isinstance(param_value, str):
converted_kwargs[param_name] = param_value
else:
# Try to convert string argument using type adapter
try:
adapter = get_cached_typeadapter(param.annotation)
# Try JSON parsing first for complex types
try:
converted_kwargs[param_name] = adapter.validate_json(
param_value
)
except (ValueError, TypeError, pydantic_core.ValidationError):
# Fallback to direct validation
converted_kwargs[param_name] = adapter.validate_python(
param_value
)
except (ValueError, TypeError, pydantic_core.ValidationError) as e:
# If conversion fails, provide informative error
raise PromptError(
f"Could not convert argument '{param_name}' with value '{param_value}' "
f"to expected type {param.annotation}. Error: {e}"
) from e
else:
# Parameter not in function signature, pass as-is
converted_kwargs[param_name] = param_value
return converted_kwargs
async def render(
self,
arguments: dict[str, Any] | None = None,
) -> PromptResult:
"""Render the prompt with arguments."""
# Validate required arguments
if self.arguments:
required = {arg.name for arg in self.arguments if arg.required}
provided = set(arguments or {})
missing = required - provided
if missing:
raise ValueError(f"Missing required arguments: {missing}")
try:
# Prepare arguments
kwargs = arguments.copy() if arguments else {}
# Convert string arguments to expected types BEFORE validation
kwargs = self._convert_string_arguments(kwargs)
# self.fn is wrapped by without_injected_parameters which handles
# dependency resolution internally
result = self.fn(**kwargs)
if inspect.isawaitable(result):
result = await result
return self.convert_result(result)
except Exception as e:
logger.exception(f"Error rendering prompt {self.name}")
raise PromptError(f"Error rendering prompt {self.name}.") from e
def register_with_docket(self, docket: Docket) -> None:
"""Register this prompt with docket for background execution.
FunctionPrompt registers the underlying function, which has the user's
Depends parameters for docket to resolve.
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key]) # type: ignore[arg-type]
async def add_to_docket( # type: ignore[override]
self,
docket: Docket,
arguments: dict[str, Any] | None,
*,
fn_key: str | None = None,
task_key: str | None = None,
**kwargs: Any,
) -> Execution:
"""Schedule this prompt for background execution via docket.
FunctionPrompt splats the arguments dict since .fn expects **kwargs.
Args:
docket: The Docket instance
arguments: Prompt arguments
fn_key: Function lookup key in Docket registry (defaults to self.key)
task_key: Redis storage key for the result
**kwargs: Additional kwargs passed to docket.add()
"""
lookup_key = fn_key or self.key
if task_key:
kwargs["key"] = task_key
return await docket.add(lookup_key, **kwargs)(**(arguments or {}))
__all__ = [
"Message",
"Prompt",
"PromptArgument",
"PromptResult",
]
# Type alias for any function that can be decorated
AnyFunction = Callable[..., Any]
def __getattr__(name: str) -> Any:
"""Deprecated re-exports for backwards compatibility."""
deprecated_exports = {
"FunctionPrompt": "FunctionPrompt",
"prompt": "prompt",
}
if name in deprecated_exports:
import fastmcp
@overload
def prompt(fn: AnyFunction) -> FunctionPrompt: ...
@overload
def prompt(
name_or_fn: str,
*,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], FunctionPrompt]: ...
@overload
def prompt(
name_or_fn: None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], FunctionPrompt]: ...
def prompt(
name_or_fn: str | AnyFunction | None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> (
Callable[[AnyFunction], FunctionPrompt]
| FunctionPrompt
| partial[Callable[[AnyFunction], FunctionPrompt] | FunctionPrompt]
):
"""Standalone decorator to create a prompt without registering it to a server.
This decorator creates a FunctionPrompt object from a function. Unlike
@server.prompt(), this does NOT register the prompt with any server - you must
explicitly add it using server.add_prompt().
This is useful for:
- Creating prompts that will be modified before registration
- Defining prompts in modules that are discovered by FileSystemProvider
- Creating reusable prompt definitions
This decorator supports multiple calling patterns:
- @prompt (without parentheses)
- @prompt() (with empty parentheses)
- @prompt("custom_name") (with name as first argument)
- @prompt(name="custom_name") (with name as keyword argument)
Args:
name_or_fn: Either a function (when used as @prompt), a string name, or None
name: Optional name for the prompt (keyword-only, alternative to name_or_fn)
title: Optional title for the prompt
description: Optional description of what the prompt does
icons: Optional icons for the prompt
tags: Optional set of tags for categorizing the prompt
meta: Optional meta information about the prompt
task: Optional task configuration for background execution (default False)
auth: Optional authorization checks for the prompt
Returns:
A FunctionPrompt when decorating a function, or a decorator function when
called with parameters.
Example:
```python
from fastmcp.prompts import prompt
from fastmcp import FastMCP
@prompt
def analyze(topic: str) -> str:
return f"Please analyze: {topic}"
@prompt("custom_prompt")
def my_prompt(data: str) -> str:
return f"Process this data: {data}"
# Prompts are not registered yet - add them explicitly
mcp = FastMCP()
mcp.add_prompt(analyze)
mcp.add_prompt(my_prompt)
```
"""
if isinstance(name_or_fn, classmethod):
raise TypeError(
inspect.cleandoc(
"""
To decorate a classmethod, first define the method and then call
prompt() directly on the method instead of using it as a
decorator. See https://gofastmcp.com/patterns/decorating-methods
for examples and more information.
"""
if fastmcp.settings.deprecation_warnings:
warnings.warn(
f"Importing {name} from fastmcp.prompts.prompt is deprecated. "
f"Import from fastmcp.prompts.function_prompt instead.",
DeprecationWarning,
stacklevel=2,
)
)
from fastmcp.prompts import function_prompt
# Determine the actual name and function based on the calling pattern
if inspect.isroutine(name_or_fn):
# Case 1: @prompt (without parens) - function passed directly
fn = name_or_fn
prompt_name = name # Use keyword name if provided, otherwise None
return getattr(function_prompt, name)
# Default to False for standalone usage (no server to inherit from)
supports_task: bool | TaskConfig = task if task is not None else False
# Create the prompt object without registration
return Prompt.from_function(
fn=fn,
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=supports_task,
auth=auth,
)
elif isinstance(name_or_fn, str):
# Case 2: @prompt("custom_name") - name passed as first argument
if name is not None:
raise TypeError(
"Cannot specify both a name as first argument and as keyword argument. "
f"Use either @prompt('{name_or_fn}') or @prompt(name='{name}'), not both."
)
prompt_name = name_or_fn
elif name_or_fn is None:
# Case 3: @prompt() or @prompt(name="something") - use keyword name
prompt_name = name
else:
raise TypeError(
f"First argument to @prompt must be a function, string, or None, got {type(name_or_fn)}"
)
# Return partial for cases where we need to wait for the function
return partial(
prompt,
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=task,
auth=auth,
)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

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

@ -1,10 +1,5 @@
from .resource import (
FunctionResource,
Resource,
ResourceContent,
ResourceResult,
resource,
)
from .function_resource import FunctionResource, resource
from .resource import Resource, ResourceContent, ResourceResult
from .template import ResourceTemplate
from .types import (
BinaryResource,

311
src/fastmcp/resources/function_resource.py Normal file
View file

@ -0,0 +1,311 @@
"""Standalone @resource decorator for FastMCP."""
from __future__ import annotations
import inspect
import warnings
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeVar, runtime_checkable
from mcp.types import Annotations, Icon
from pydantic import AnyUrl
import fastmcp
from fastmcp.decorators import resolve_task_config
from fastmcp.resources.resource import Resource, ResourceResult
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.tool import AuthCheckCallable
from fastmcp.utilities.types import get_fn_name
if TYPE_CHECKING:
from docket import Docket
from fastmcp.resources.template import ResourceTemplate
F = TypeVar("F", bound=Callable[..., Any])
@runtime_checkable
class DecoratedResource(Protocol):
"""Protocol for functions decorated with @resource."""
__fastmcp__: ResourceMeta
def __call__(self, *args: Any, **kwargs: Any) -> Any: ...
@dataclass(frozen=True, kw_only=True)
class ResourceMeta:
"""Metadata attached to functions by the @resource decorator."""
type: Literal["resource"] = field(default="resource", init=False)
uri: str
name: str | None = None
title: str | None = None
description: str | None = None
icons: list[Icon] | None = None
tags: set[str] | None = None
mime_type: str | None = None
annotations: Annotations | None = None
meta: dict[str, Any] | None = None
task: bool | TaskConfig | None = None
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None
class FunctionResource(Resource):
"""A resource that defers data loading by wrapping a function.
The function is only called when the resource is read, allowing for lazy loading
of potentially expensive data. This is particularly useful when listing resources,
as the function won't be called until the resource is actually accessed.
The function can return:
- str for text content (default)
- bytes for binary content
- other types will be converted to JSON
"""
fn: Callable[..., Any]
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
uri: str | AnyUrl | None = None,
*,
metadata: ResourceMeta | None = None,
# Keep individual params for backwards compat
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
annotations: Annotations | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionResource:
"""Create a FunctionResource from a function.
Args:
fn: The function to wrap
uri: The URI for the resource (required if metadata not provided)
metadata: ResourceMeta object with all configuration. If provided,
individual parameters must not be passed.
name, title, etc.: Individual parameters for backwards compatibility.
Cannot be used together with metadata parameter.
"""
# Check mutual exclusion
individual_params_provided = (
any(
x is not None
for x in [
name,
title,
description,
icons,
mime_type,
tags,
annotations,
meta,
task,
auth,
]
)
or uri is not None
)
if metadata is not None and individual_params_provided:
raise TypeError(
"Cannot pass both 'metadata' and individual parameters to from_function(). "
"Use metadata alone or individual parameters alone."
)
# Build metadata from kwargs if not provided
if metadata is None:
if uri is None:
raise TypeError("uri is required when metadata is not provided")
metadata = ResourceMeta(
uri=str(uri),
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
mime_type=mime_type,
annotations=annotations,
meta=meta,
task=task,
auth=auth,
)
uri_obj = AnyUrl(metadata.uri)
# Get function name before any transformations
func_name = metadata.name or get_fn_name(fn)
# Normalize task to TaskConfig and validate
task_value = metadata.task
if task_value is None:
task_config = TaskConfig(mode="forbidden")
elif isinstance(task_value, bool):
task_config = TaskConfig.from_bool(task_value)
else:
task_config = task_value
task_config.validate_function(fn, func_name)
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Wrap fn to handle dependency resolution internally
wrapped_fn = without_injected_parameters(fn)
return cls(
fn=wrapped_fn,
uri=uri_obj,
name=func_name,
title=metadata.title,
description=metadata.description or inspect.getdoc(fn),
icons=metadata.icons,
mime_type=metadata.mime_type or "text/plain",
tags=metadata.tags or set(),
annotations=metadata.annotations,
meta=metadata.meta,
task_config=task_config,
auth=metadata.auth,
)
async def read(
self,
) -> str | bytes | ResourceResult:
"""Read the resource by calling the wrapped function."""
# self.fn is wrapped by without_injected_parameters which handles
# dependency resolution internally
result = self.fn()
if inspect.isawaitable(result):
result = await result
# If user returned another Resource, read it recursively
if isinstance(result, Resource):
return await result.read()
return result
def register_with_docket(self, docket: Docket) -> None:
"""Register this resource with docket for background execution.
FunctionResource registers the underlying function, which has the user's
Depends parameters for docket to resolve.
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key])
def resource(
uri: str,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
annotations: Annotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[F], F]:
"""Standalone decorator to mark a function as an MCP resource.
Returns the original function with metadata attached. Register with a server
using mcp.add_resource().
"""
if isinstance(annotations, dict):
annotations = Annotations(**annotations)
if inspect.isroutine(uri):
raise TypeError(
"The @resource decorator requires a URI. "
"Use @resource('uri') instead of @resource"
)
def create_resource(fn: Callable[..., Any]) -> FunctionResource | ResourceTemplate:
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.dependencies import without_injected_parameters
resolved = resolve_task_config(task)
has_uri_params = "{" in uri and "}" in uri
wrapper_fn = without_injected_parameters(fn)
has_func_params = bool(inspect.signature(wrapper_fn).parameters)
# Create metadata first
resource_meta = ResourceMeta(
uri=uri,
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
mime_type=mime_type,
annotations=annotations,
meta=meta,
task=resolved,
auth=auth,
)
if has_uri_params or has_func_params:
# ResourceTemplate doesn't have metadata support yet, so pass individual params
return ResourceTemplate.from_function(
fn=fn,
uri_template=uri,
name=name,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
task=resolved,
auth=auth,
)
else:
return FunctionResource.from_function(fn, metadata=resource_meta)
def attach_metadata(fn: F) -> F:
metadata = ResourceMeta(
uri=uri,
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
mime_type=mime_type,
annotations=annotations,
meta=meta,
task=task,
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
return fn
def decorator(fn: F) -> F:
if fastmcp.settings.decorator_mode == "object":
warnings.warn(
"decorator_mode='object' is deprecated and will be removed in a future version. "
"Decorators now return the original function with metadata attached.",
DeprecationWarning,
stacklevel=3,
)
return create_resource(fn) # type: ignore[return-value]
return attach_metadata(fn)
return decorator

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

@ -3,7 +3,6 @@
from __future__ import annotations
import base64
import inspect
from collections.abc import Callable
from typing import TYPE_CHECKING, Annotated, Any, ClassVar, overload
@ -13,7 +12,8 @@ if TYPE_CHECKING:
from docket import Docket
from docket.execution import Execution
from fastmcp.resources.template import ResourceTemplate
from fastmcp.resources.function_resource import FunctionResource
import pydantic
import pydantic_core
from mcp.types import Annotations, Icon
@ -28,14 +28,9 @@ from pydantic import (
)
from typing_extensions import Self
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
from fastmcp.tools.tool import AuthCheckCallable
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.types import get_fn_name
class ResourceContent(pydantic.BaseModel):
@ -235,10 +230,12 @@ class Resource(FastMCPComponent):
Field(description="Authorization checks for this resource", exclude=True),
] = None
@staticmethod
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
uri: str | AnyUrl,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
@ -250,6 +247,10 @@ class Resource(FastMCPComponent):
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionResource:
from fastmcp.resources.function_resource import (
FunctionResource,
)
return FunctionResource.from_function(
fn=fn,
uri=uri,
@ -409,232 +410,34 @@ class Resource(FastMCPComponent):
return await docket.add(lookup_key, **kwargs)()
class FunctionResource(Resource):
"""A resource that defers data loading by wrapping a function.
The function is only called when the resource is read, allowing for lazy loading
of potentially expensive data. This is particularly useful when listing resources,
as the function won't be called until the resource is actually accessed.
The function can return:
- str for text content (default)
- bytes for binary content
- other types will be converted to JSON
"""
fn: Callable[..., Any]
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
uri: str | AnyUrl,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
annotations: Annotations | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionResource:
"""Create a FunctionResource from a function."""
if isinstance(uri, str):
uri = AnyUrl(uri)
func_name = name or get_fn_name(fn)
# Normalize task to TaskConfig and validate
if task is None:
task_config = TaskConfig(mode="forbidden")
elif isinstance(task, bool):
task_config = TaskConfig.from_bool(task)
else:
task_config = task
task_config.validate_function(fn, func_name)
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Wrap fn to handle dependency resolution internally
wrapped_fn = without_injected_parameters(fn)
return cls(
fn=wrapped_fn,
uri=uri,
name=name or get_fn_name(fn),
title=title,
description=description or inspect.getdoc(fn),
icons=icons,
mime_type=mime_type or "text/plain",
tags=tags or set(),
annotations=annotations,
meta=meta,
task_config=task_config,
auth=auth,
)
async def read(
self,
) -> str | bytes | ResourceResult:
"""Read the resource by calling the wrapped function."""
# self.fn is wrapped by without_injected_parameters which handles
# dependency resolution internally
result = self.fn()
if inspect.isawaitable(result):
result = await result
# If user returned another Resource, read it recursively
if isinstance(result, Resource):
return await result.read()
return result
def register_with_docket(self, docket: Docket) -> None:
"""Register this resource with docket for background execution.
FunctionResource registers the underlying function, which has the user's
Depends parameters for docket to resolve.
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key])
__all__ = [
"Resource",
"ResourceContent",
"ResourceResult",
]
# Type alias for any function that can be decorated
AnyFunction = Callable[..., Any]
def __getattr__(name: str) -> Any:
"""Deprecated re-exports for backwards compatibility."""
deprecated_exports = {
"FunctionResource": "FunctionResource",
"resource": "resource",
}
if name in deprecated_exports:
import warnings
def resource(
uri: str,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
annotations: Annotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], Resource | ResourceTemplate]:
"""Standalone decorator to create a resource without registering it to a server.
import fastmcp
This decorator creates a Resource or ResourceTemplate object from a function.
Unlike @server.resource(), this does NOT register the resource with any server -
you must explicitly add it using server.add_resource() or server.add_template().
If the URI contains parameters (e.g. "resource://{param}") or the function
has parameters, it will create a ResourceTemplate instead of a Resource.
This is useful for:
- Creating resources that will be modified before registration
- Defining resources in modules that are discovered by FileSystemProvider
- Creating reusable resource definitions
Args:
uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
name: Optional name for the resource
title: Optional title for the resource
description: Optional description of the resource
icons: Optional icons for the resource
mime_type: Optional MIME type for the resource
tags: Optional set of tags for categorizing the resource
annotations: Optional annotations about the resource's behavior
meta: Optional meta information about the resource
task: Optional task configuration for background execution (default False)
auth: Optional authorization checks for the resource
Returns:
A decorator function that returns a Resource or ResourceTemplate.
Example:
```python
from fastmcp.resources import resource
from fastmcp import FastMCP
@resource("data://config")
def get_config() -> str:
return '{"setting": "value"}'
@resource("data://{city}/weather")
def get_weather(city: str) -> str:
return f"Weather for {city}"
# Resources are not registered yet - add them explicitly
mcp = FastMCP()
mcp.add_resource(get_config)
mcp.add_template(get_weather)
```
"""
if isinstance(annotations, dict):
annotations = Annotations(**annotations)
# Check if user passed function directly instead of calling decorator
if inspect.isroutine(uri):
raise TypeError(
"The @resource decorator was used incorrectly. "
"Did you forget to call it? Use @resource('uri') instead of @resource"
)
def decorator(fn: AnyFunction) -> Resource | ResourceTemplate:
if isinstance(fn, classmethod):
raise TypeError(
inspect.cleandoc(
"""
To decorate a classmethod, first define the method and then call
resource() directly on the method instead of using it as a
decorator. See https://gofastmcp.com/patterns/decorating-methods
for examples and more information.
"""
)
if fastmcp.settings.deprecation_warnings:
warnings.warn(
f"Importing {name} from fastmcp.resources.resource is deprecated. "
f"Import from fastmcp.resources.function_resource instead.",
DeprecationWarning,
stacklevel=2,
)
from fastmcp.resources import function_resource
# Default to False for standalone usage (no server to inherit from)
supports_task: bool | TaskConfig = task if task is not None else False
return getattr(function_resource, name)
# Check if this should be a template
has_uri_params = "{" in uri and "}" in uri
# Use wrapper to check for user-facing parameters
from fastmcp.server.dependencies import without_injected_parameters
wrapper_fn = without_injected_parameters(fn)
has_func_params = bool(inspect.signature(wrapper_fn).parameters)
if has_uri_params or has_func_params:
from fastmcp.resources.template import ResourceTemplate
return ResourceTemplate.from_function(
fn=fn,
uri_template=uri,
name=name,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
task=supports_task,
auth=auth,
)
else:
return Resource.from_function(
fn=fn,
uri=uri,
name=name,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
task=supports_task,
auth=auth,
)
return decorator
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

84
src/fastmcp/server/providers/filesystem_discovery.py
View file

@ -176,7 +176,8 @@ def extract_components(module: ModuleType) -> list[FastMCPComponent]:
"""Extract all MCP components from a module.
Scans all module attributes for instances of Tool, Resource,
ResourceTemplate, or Prompt objects created by standalone decorators.
ResourceTemplate, or Prompt objects created by standalone decorators,
or functions decorated with @tool/@resource/@prompt that have __fastmcp__ metadata.
Args:
module: The imported module to scan.
@ -185,9 +186,16 @@ def extract_components(module: ModuleType) -> list[FastMCPComponent]:
List of component objects (Tool, Resource, ResourceTemplate, Prompt).
"""
# Import here to avoid circular imports
import inspect
from fastmcp.decorators import get_fastmcp_meta
from fastmcp.prompts.function_prompt import PromptMeta
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.function_resource import ResourceMeta
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.dependencies import without_injected_parameters
from fastmcp.tools.function_tool import ToolMeta
from fastmcp.tools.tool import Tool
component_types = (Tool, Resource, ResourceTemplate, Prompt)
@ -206,6 +214,80 @@ def extract_components(module: ModuleType) -> list[FastMCPComponent]:
# Check if this object is a component type
if isinstance(obj, component_types):
components.append(obj)
continue
# Check for functions with __fastmcp__ metadata
meta = get_fastmcp_meta(obj)
if meta is not None:
if isinstance(meta, ToolMeta):
resolved_task = meta.task if meta.task is not None else False
tool = Tool.from_function(
obj,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
tags=meta.tags,
output_schema=meta.output_schema,
annotations=meta.annotations,
meta=meta.meta,
task=resolved_task,
exclude_args=meta.exclude_args,
serializer=meta.serializer,
auth=meta.auth,
)
components.append(tool)
elif isinstance(meta, ResourceMeta):
resolved_task = meta.task if meta.task is not None else False
has_uri_params = "{" in meta.uri and "}" in meta.uri
wrapper_fn = without_injected_parameters(obj)
has_func_params = bool(inspect.signature(wrapper_fn).parameters)
if has_uri_params or has_func_params:
resource = ResourceTemplate.from_function(
fn=obj,
uri_template=meta.uri,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
mime_type=meta.mime_type,
tags=meta.tags,
annotations=meta.annotations,
meta=meta.meta,
task=resolved_task,
auth=meta.auth,
)
else:
resource = Resource.from_function(
fn=obj,
uri=meta.uri,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
mime_type=meta.mime_type,
tags=meta.tags,
annotations=meta.annotations,
meta=meta.meta,
task=resolved_task,
auth=meta.auth,
)
components.append(resource)
elif isinstance(meta, PromptMeta):
resolved_task = meta.task if meta.task is not None else False
prompt = Prompt.from_function(
obj,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
tags=meta.tags,
meta=meta.meta,
task=resolved_task,
auth=meta.auth,
)
components.append(prompt)
return components

452
src/fastmcp/server/providers/local_provider.py
View file

@ -34,14 +34,15 @@ import mcp.types
from mcp.types import Annotations, AnyFunction, ToolAnnotations
import fastmcp
from fastmcp.prompts.prompt import FunctionPrompt, Prompt
from fastmcp.prompts.prompt import prompt as standalone_prompt
from fastmcp.prompts.function_prompt import FunctionPrompt
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.function_resource import resource as standalone_resource
from fastmcp.resources.resource import Resource
from fastmcp.resources.resource import resource as standalone_resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.providers.base import Provider
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.tool import AuthCheckCallable, FunctionTool, Tool
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import AuthCheckCallable, Tool
from fastmcp.tools.tool_transform import (
ToolTransformConfig,
apply_transformations_to_tools,
@ -183,16 +184,95 @@ class LocalProvider(Provider):
"""
return self._components.get(key)
def add_tool(self, tool: Tool) -> Tool:
"""Add a tool to this provider's storage."""
def add_tool(self, tool: Tool | Callable[..., Any]) -> Tool:
"""Add a tool to this provider's storage.
Accepts either a Tool object or a decorated function with __fastmcp__ metadata.
"""
if not isinstance(tool, Tool):
from fastmcp.decorators import get_fastmcp_meta
from fastmcp.tools.function_tool import ToolMeta
meta = get_fastmcp_meta(tool)
if meta is not None and isinstance(meta, ToolMeta):
resolved_task = meta.task if meta.task is not None else False
tool = Tool.from_function(
tool,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
tags=meta.tags,
output_schema=meta.output_schema,
annotations=meta.annotations,
meta=meta.meta,
task=resolved_task,
exclude_args=meta.exclude_args,
serializer=meta.serializer,
auth=meta.auth,
)
else:
tool = Tool.from_function(tool)
return self._add_component(tool)
def remove_tool(self, name: str) -> None:
"""Remove a tool from this provider's storage."""
self._remove_component(Tool.make_key(name))
def add_resource(self, resource: Resource) -> Resource:
"""Add a resource to this provider's storage."""
def add_resource(
self, resource: Resource | ResourceTemplate | Callable[..., Any]
) -> Resource | ResourceTemplate:
"""Add a resource to this provider's storage.
Accepts either a Resource/ResourceTemplate object or a decorated function with __fastmcp__ metadata.
"""
if not isinstance(resource, (Resource, ResourceTemplate)):
from fastmcp.decorators import get_fastmcp_meta
from fastmcp.resources.function_resource import ResourceMeta
from fastmcp.server.dependencies import without_injected_parameters
meta = get_fastmcp_meta(resource)
if meta is not None and isinstance(meta, ResourceMeta):
resolved_task = meta.task if meta.task is not None else False
has_uri_params = "{" in meta.uri and "}" in meta.uri
wrapper_fn = without_injected_parameters(resource)
has_func_params = bool(inspect.signature(wrapper_fn).parameters)
if has_uri_params or has_func_params:
resource = ResourceTemplate.from_function(
fn=resource,
uri_template=meta.uri,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
mime_type=meta.mime_type,
tags=meta.tags,
annotations=meta.annotations,
meta=meta.meta,
task=resolved_task,
auth=meta.auth,
)
else:
resource = Resource.from_function(
fn=resource,
uri=meta.uri,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
mime_type=meta.mime_type,
tags=meta.tags,
annotations=meta.annotations,
meta=meta.meta,
task=resolved_task,
auth=meta.auth,
)
else:
raise TypeError(
f"Expected Resource, ResourceTemplate, or @resource-decorated function, got {type(resource).__name__}. "
"Use @resource('uri') decorator or pass a Resource/ResourceTemplate instance."
)
return self._add_component(resource)
def remove_resource(self, uri: str) -> None:
@ -207,8 +287,34 @@ class LocalProvider(Provider):
"""Remove a resource template from this provider's storage."""
self._remove_component(ResourceTemplate.make_key(uri_template))
def add_prompt(self, prompt: Prompt) -> Prompt:
"""Add a prompt to this provider's storage."""
def add_prompt(self, prompt: Prompt | Callable[..., Any]) -> Prompt:
"""Add a prompt to this provider's storage.
Accepts either a Prompt object or a decorated function with __fastmcp__ metadata.
"""
if not isinstance(prompt, Prompt):
from fastmcp.decorators import get_fastmcp_meta
from fastmcp.prompts.function_prompt import PromptMeta
meta = get_fastmcp_meta(prompt)
if meta is not None and isinstance(meta, PromptMeta):
resolved_task = meta.task if meta.task is not None else False
prompt = Prompt.from_function(
prompt,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
tags=meta.tags,
meta=meta.meta,
task=resolved_task,
auth=meta.auth,
)
else:
raise TypeError(
f"Expected Prompt or @prompt-decorated function, got {type(prompt).__name__}. "
"Use @prompt decorator or pass a Prompt instance."
)
return self._add_component(prompt)
def remove_prompt(self, name: str) -> None:
@ -465,47 +571,81 @@ class LocalProvider(Provider):
if isinstance(name_or_fn, classmethod):
raise TypeError(
inspect.cleandoc(
"""
To decorate a classmethod, first define the method and then call
tool() directly on the method instead of using it as a
decorator. See https://gofastmcp.com/patterns/decorating-methods
for examples and more information.
"""
"To decorate a classmethod, use @classmethod above @tool. "
"See https://gofastmcp.com/servers/tools#using-with-methods"
)
def decorate_and_register(
fn: AnyFunction, tool_name: str | None
) -> FunctionTool | AnyFunction:
# Check for unbound method
try:
params = list(inspect.signature(fn).parameters.keys())
except (ValueError, TypeError):
params = []
if params and params[0] in ("self", "cls"):
fn_name = getattr(fn, "__name__", "function")
raise TypeError(
f"The function '{fn_name}' has '{params[0]}' as its first parameter. "
f"Use the standalone @tool decorator and register the bound method:\n\n"
f" from fastmcp.tools import tool\n\n"
f" class MyClass:\n"
f" @tool\n"
f" def {fn_name}(...):\n"
f" ...\n\n"
f" obj = MyClass()\n"
f" mcp.add_tool(obj.{fn_name})\n\n"
f"See https://gofastmcp.com/servers/tools#using-with-methods"
)
)
# Determine the actual name and function based on the calling pattern
resolved_task: bool | TaskConfig = task if task is not None else False
if fastmcp.settings.decorator_mode == "object":
tool_obj = Tool.from_function(
fn,
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
exclude_args=exclude_args,
meta=meta,
serializer=serializer,
task=resolved_task,
auth=auth,
)
self._add_component(tool_obj)
if not enabled:
self.disable(keys=[tool_obj.key])
return tool_obj
else:
from fastmcp.tools.function_tool import ToolMeta
metadata = ToolMeta(
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=task,
exclude_args=exclude_args,
serializer=serializer,
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
tool_obj = self.add_tool(fn)
if not enabled:
self.disable(keys=[tool_obj.key])
return fn
if inspect.isroutine(name_or_fn):
# Case 1: @tool (without parens) - function passed directly
# Case 2: direct call like tool(fn, name="something")
fn = name_or_fn
tool_name = name # Use keyword name if provided, otherwise None
# Resolve task parameter - default to False for standalone usage
supports_task: bool | TaskConfig = task if task is not None else False
# Register the tool immediately and return the tool object
tool_obj = Tool.from_function(
fn,
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
exclude_args=exclude_args,
meta=meta,
serializer=serializer,
task=supports_task,
auth=auth,
)
self.add_tool(tool_obj)
# If disabled, add to blocklist
if not enabled:
self.disable(keys=[tool_obj.key])
return tool_obj
return decorate_and_register(name_or_fn, name)
elif isinstance(name_or_fn, str):
# Case 3: @tool("custom_name") - name passed as first argument
@ -556,7 +696,7 @@ class LocalProvider(Provider):
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], Resource | ResourceTemplate]:
) -> Callable[[AnyFunction], Resource | ResourceTemplate | AnyFunction]:
"""Decorator to register a function as a resource.
If the URI contains parameters (e.g. "resource://{param}") or the function
@ -592,36 +732,85 @@ class LocalProvider(Provider):
return f"Weather for {city}"
```
"""
# Resolve task parameter - default to False for standalone usage
supports_task: bool | TaskConfig = task if task is not None else False
if isinstance(annotations, dict):
annotations = Annotations(**annotations)
# Get the standalone decorator
create_resource = standalone_resource(
uri,
name=name,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
task=supports_task,
auth=auth,
)
if inspect.isroutine(uri):
raise TypeError(
"The @resource decorator was used incorrectly. "
"It requires a URI as the first argument. "
"Use @resource('uri') instead of @resource"
)
def decorator(fn: AnyFunction) -> Resource | ResourceTemplate:
# Delegate to standalone decorator for object creation
obj = create_resource(fn)
# Register with this provider
if isinstance(obj, ResourceTemplate):
self.add_template(obj)
resolved_task: bool | TaskConfig = task if task is not None else False
def decorator(fn: AnyFunction) -> Resource | ResourceTemplate | AnyFunction:
# Check for unbound method
try:
params = list(inspect.signature(fn).parameters.keys())
except (ValueError, TypeError):
params = []
if params and params[0] in ("self", "cls"):
fn_name = getattr(fn, "__name__", "function")
raise TypeError(
f"The function '{fn_name}' has '{params[0]}' as its first parameter. "
f"Use the standalone @resource decorator and register the bound method:\n\n"
f" from fastmcp.resources import resource\n\n"
f" class MyClass:\n"
f" @resource('{uri}')\n"
f" def {fn_name}(...):\n"
f" ...\n\n"
f" obj = MyClass()\n"
f" mcp.add_resource(obj.{fn_name})\n\n"
f"See https://gofastmcp.com/servers/tools#using-with-methods"
)
if fastmcp.settings.decorator_mode == "object":
create_resource = standalone_resource(
uri,
name=name,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
task=resolved_task,
auth=auth,
)
obj = create_resource(fn)
# In legacy mode, standalone_resource always returns a component
assert isinstance(obj, (Resource, ResourceTemplate))
if isinstance(obj, ResourceTemplate):
self.add_template(obj)
else:
self.add_resource(obj)
if not enabled:
self.disable(keys=[obj.key])
return obj
else:
self.add_resource(obj)
# Handle enabled flag
if not enabled:
self.disable(keys=[obj.key])
return obj
from fastmcp.resources.function_resource import ResourceMeta
metadata = ResourceMeta(
uri=uri,
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
mime_type=mime_type,
annotations=annotations,
meta=meta,
task=task,
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
obj = self.add_resource(fn)
if not enabled:
self.disable(keys=[obj.key])
return fn
return decorator
@ -712,40 +901,97 @@ class LocalProvider(Provider):
return [{"role": "user", "content": data}]
```
"""
if isinstance(name_or_fn, classmethod):
raise TypeError(
"To decorate a classmethod, use @classmethod above @prompt. "
"See https://gofastmcp.com/servers/tools#using-with-methods"
)
def register(prompt_obj: FunctionPrompt) -> FunctionPrompt:
"""Register the prompt and handle enabled flag."""
self.add_prompt(prompt_obj)
if not enabled:
self.disable(keys=[prompt_obj.key])
return prompt_obj
def decorate_and_register(
fn: AnyFunction, prompt_name: str | None
) -> FunctionPrompt | AnyFunction:
# Check for unbound method
try:
params = list(inspect.signature(fn).parameters.keys())
except (ValueError, TypeError):
params = []
if params and params[0] in ("self", "cls"):
fn_name = getattr(fn, "__name__", "function")
raise TypeError(
f"The function '{fn_name}' has '{params[0]}' as its first parameter. "
f"Use the standalone @prompt decorator and register the bound method:\n\n"
f" from fastmcp.prompts import prompt\n\n"
f" class MyClass:\n"
f" @prompt\n"
f" def {fn_name}(...):\n"
f" ...\n\n"
f" obj = MyClass()\n"
f" mcp.add_prompt(obj.{fn_name})\n\n"
f"See https://gofastmcp.com/servers/tools#using-with-methods"
)
# Resolve task parameter - default to False for standalone usage
supports_task: bool | TaskConfig = task if task is not None else False
resolved_task: bool | TaskConfig = task if task is not None else False
# Delegate to standalone decorator for object creation
# Type ignore: standalone_prompt has overloads for specific types, but we pass
# through the union type. Runtime behavior is correct.
result = standalone_prompt(
name_or_fn, # type: ignore[arg-type]
name=name,
if fastmcp.settings.decorator_mode == "object":
prompt_obj = Prompt.from_function(
fn,
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=resolved_task,
auth=auth,
)
self._add_component(prompt_obj)
if not enabled:
self.disable(keys=[prompt_obj.key])
return prompt_obj
else:
from fastmcp.prompts.function_prompt import PromptMeta
metadata = PromptMeta(
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=task,
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
prompt_obj = self.add_prompt(fn)
if not enabled:
self.disable(keys=[prompt_obj.key])
return fn
if inspect.isroutine(name_or_fn):
return decorate_and_register(name_or_fn, name)
elif isinstance(name_or_fn, str):
if name is not None:
raise TypeError(
f"Cannot specify both a name as first argument and as keyword argument. "
f"Use either @prompt('{name_or_fn}') or @prompt(name='{name}'), not both."
)
prompt_name = name_or_fn
elif name_or_fn is None:
prompt_name = name
else:
raise TypeError(f"Invalid first argument: {type(name_or_fn)}")
return partial(
self.prompt,
name=prompt_name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=supports_task,
enabled=enabled,
task=task,
auth=auth,
)
# If standalone returned a FunctionPrompt directly (@prompt without parens),
# register it and return
if isinstance(result, FunctionPrompt):
return register(result)
# Otherwise, standalone returned a decorator/partial - wrap it to register after creation
def decorator(fn: AnyFunction) -> FunctionPrompt:
prompt_obj = result(fn)
return register(prompt_obj)
return decorator

2
src/fastmcp/server/sampling/sampling_tool.py
View file

@ -9,7 +9,7 @@ from typing import Any
from mcp.types import Tool as SDKTool
from pydantic import ConfigDict
from fastmcp.tools.tool import ParsedFunction
from fastmcp.tools.function_tool import ParsedFunction
from fastmcp.utilities.types import FastMCPBaseModel

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

@ -66,7 +66,8 @@ from fastmcp.exceptions import (
)
from fastmcp.mcp_config import MCPConfig
from fastmcp.prompts import Prompt
from fastmcp.prompts.prompt import FunctionPrompt, PromptResult
from fastmcp.prompts.function_prompt import FunctionPrompt
from fastmcp.prompts.prompt import PromptResult
from fastmcp.resources.resource import Resource, ResourceResult
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.auth import AuthContext, AuthProvider, run_auth_checks
@ -85,7 +86,8 @@ from fastmcp.server.tasks.capabilities import get_task_capabilities
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
from fastmcp.settings import DuplicateBehavior as DuplicateBehaviorSetting
from fastmcp.settings import Settings
from fastmcp.tools.tool import AuthCheckCallable, FunctionTool, Tool, ToolResult
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import AuthCheckCallable, Tool, ToolResult
from fastmcp.tools.tool_transform import ToolTransformConfig
from fastmcp.utilities.async_utils import gather
from fastmcp.utilities.cli import log_server_banner
@ -1806,14 +1808,14 @@ class FastMCP(Generic[LifespanResultT]):
except NotFoundError:
raise
def add_tool(self, tool: Tool) -> Tool:
def add_tool(self, tool: Tool | Callable[..., Any]) -> Tool:
"""Add a tool to the server.
The tool function can optionally request a Context object by adding a parameter
with the Context type annotation. See the @tool decorator for examples.
Args:
tool: The Tool instance to register
tool: The Tool instance or @tool-decorated function to register
Returns:
The tool instance that was added to the server.
@ -1967,11 +1969,13 @@ class FastMCP(Generic[LifespanResultT]):
return result
def add_resource(self, resource: Resource) -> Resource:
def add_resource(
self, resource: Resource | Callable[..., Any]
) -> Resource | ResourceTemplate:
"""Add a resource to the server.
Args:
resource: A Resource instance to add
resource: A Resource instance or @resource-decorated function to add
Returns:
The resource instance that was added to the server.
@ -2003,7 +2007,7 @@ class FastMCP(Generic[LifespanResultT]):
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], Resource | ResourceTemplate]:
) -> Callable[[AnyFunction], Resource | ResourceTemplate | AnyFunction]:
"""Decorator to register a function as a resource.
The function will be called when the resource is read to generate its content.
@ -2070,16 +2074,16 @@ class FastMCP(Generic[LifespanResultT]):
auth=auth,
)
def decorator(fn: AnyFunction) -> Resource | ResourceTemplate:
def decorator(fn: AnyFunction) -> Resource | ResourceTemplate | AnyFunction:
return inner_decorator(fn)
return decorator
def add_prompt(self, prompt: Prompt) -> Prompt:
def add_prompt(self, prompt: Prompt | Callable[..., Any]) -> Prompt:
"""Add a prompt to the server.
Args:
prompt: A Prompt instance to add
prompt: A Prompt instance or @prompt-decorated function to add
Returns:
The prompt instance that was added to the server.

17
src/fastmcp/settings.py
View file

@ -348,3 +348,20 @@ class Settings(BaseSettings):
),
),
] = "stable"
decorator_mode: Annotated[
Literal["function", "object"],
Field(
description=inspect.cleandoc(
"""
Controls what decorators (@tool, @resource, @prompt) return.
- "function" (default): Decorators return the original function unchanged.
The function remains callable and is registered with the server normally.
- "object" (deprecated): Decorators return component objects (FunctionTool,
FunctionResource, FunctionPrompt). This was the default behavior in v2 and
will be removed in a future version.
"""
),
),
] = "function"

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

@ -1,4 +1,12 @@
from .tool import FunctionTool, Tool, tool
from .function_tool import FunctionTool, tool
from .tool import Tool, ToolResult
from .tool_transform import forward, forward_raw
__all__ = ["FunctionTool", "Tool", "forward", "forward_raw", "tool"]
__all__ = [
"FunctionTool",
"Tool",
"ToolResult",
"forward",
"forward_raw",
"tool",
]

594
src/fastmcp/tools/function_tool.py Normal file
View file

@ -0,0 +1,594 @@
"""Standalone @tool decorator for FastMCP."""
from __future__ import annotations
import inspect
import warnings
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Any,
Generic,
Literal,
Protocol,
TypeVar,
get_type_hints,
overload,
runtime_checkable,
)
import mcp.types
from mcp.types import Icon, ToolAnnotations, ToolExecution
from pydantic import PydanticSchemaGenerationError
from typing_extensions import TypeVar as TypeVarExt
import fastmcp
from fastmcp.decorators import resolve_task_config
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.tool import (
AuthCheckCallable,
Tool,
ToolResult,
ToolResultSerializerType,
)
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
Audio,
File,
Image,
NotSet,
NotSetT,
create_function_without_params,
get_cached_typeadapter,
replace_type,
)
if TYPE_CHECKING:
from docket import Docket
from docket.execution import Execution
F = TypeVar("F", bound=Callable[..., Any])
T = TypeVarExt("T", default=Any)
logger = get_logger(__name__)
@runtime_checkable
class DecoratedTool(Protocol):
"""Protocol for functions decorated with @tool."""
__fastmcp__: ToolMeta
def __call__(self, *args: Any, **kwargs: Any) -> Any: ...
@dataclass(frozen=True, kw_only=True)
class ToolMeta:
"""Metadata attached to functions by the @tool decorator."""
type: Literal["tool"] = field(default="tool", init=False)
name: str | None = None
title: str | None = None
description: str | None = None
icons: list[Icon] | None = None
tags: set[str] | None = None
output_schema: dict[str, Any] | NotSetT | None = NotSet
annotations: ToolAnnotations | None = None
meta: dict[str, Any] | None = None
task: bool | TaskConfig | None = None
exclude_args: list[str] | None = None
serializer: Any | None = None
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None
@dataclass
class _WrappedResult(Generic[T]):
"""Generic wrapper for non-object return types."""
result: T
class _UnserializableType:
pass
def _is_object_schema(schema: dict[str, Any]) -> bool:
"""Check if a JSON schema represents an object type."""
# Direct object type
if schema.get("type") == "object":
return True
# Schema with properties but no explicit type is treated as object
if "properties" in schema:
return True
# Self-referencing types use $ref pointing to $defs
# The referenced type is always an object in our use case
return "$ref" in schema and "$defs" in schema
@dataclass
class ParsedFunction:
fn: Callable[..., Any]
name: str
description: str | None
input_schema: dict[str, Any]
output_schema: dict[str, Any] | None
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
exclude_args: list[str] | None = None,
validate: bool = True,
wrap_non_object_output_schema: bool = True,
) -> ParsedFunction:
if validate:
sig = inspect.signature(fn)
# Reject functions with *args or **kwargs
for param in sig.parameters.values():
if param.kind == inspect.Parameter.VAR_POSITIONAL:
raise ValueError("Functions with *args are not supported as tools")
if param.kind == inspect.Parameter.VAR_KEYWORD:
raise ValueError(
"Functions with **kwargs are not supported as tools"
)
# Reject exclude_args that don't exist in the function or don't have a default value
if exclude_args:
for arg_name in exclude_args:
if arg_name not in sig.parameters:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args does not exist in function."
)
param = sig.parameters[arg_name]
if param.default == inspect.Parameter.empty:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args must have a default value."
)
# collect name and doc before we potentially modify the function
fn_name = getattr(fn, "__name__", None) or fn.__class__.__name__
fn_doc = inspect.getdoc(fn)
# if the fn is a callable class, we need to get the __call__ method from here out
if not inspect.isroutine(fn):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Handle injected parameters (Context, Docket dependencies)
wrapper_fn = without_injected_parameters(fn)
# Also handle exclude_args with non-serializable types (issue #2431)
# This must happen before Pydantic tries to serialize the parameters
if exclude_args:
wrapper_fn = create_function_without_params(wrapper_fn, list(exclude_args))
input_type_adapter = get_cached_typeadapter(wrapper_fn)
input_schema = input_type_adapter.json_schema()
# Compress and handle exclude_args
prune_params = list(exclude_args) if exclude_args else None
input_schema = compress_schema(
input_schema, prune_params=prune_params, prune_titles=True
)
output_schema = None
# Get the return annotation from the signature
sig = inspect.signature(fn)
output_type = sig.return_annotation
# If the annotation is a string (from __future__ annotations), resolve it
if isinstance(output_type, str):
try:
# Use get_type_hints to resolve the return type
# include_extras=True preserves Annotated metadata
type_hints = get_type_hints(fn, include_extras=True)
output_type = type_hints.get("return", output_type)
except Exception as e:
# If resolution fails, keep the string annotation
logger.debug("Failed to resolve type hint for return annotation: %s", e)
if output_type not in (inspect._empty, None, Any, ...):
# there are a variety of types that we don't want to attempt to
# serialize because they are either used by FastMCP internally,
# or are MCP content types that explicitly don't form structured
# content. By replacing them with an explicitly unserializable type,
# we ensure that no output schema is automatically generated.
clean_output_type = replace_type(
output_type,
dict.fromkeys( # type: ignore[arg-type]
(
Image,
Audio,
File,
ToolResult,
mcp.types.TextContent,
mcp.types.ImageContent,
mcp.types.AudioContent,
mcp.types.ResourceLink,
mcp.types.EmbeddedResource,
),
_UnserializableType,
),
)
try:
type_adapter = get_cached_typeadapter(clean_output_type)
base_schema = type_adapter.json_schema(mode="serialization")
# Generate schema for wrapped type if it's non-object
# because MCP requires that output schemas are objects
# Check if schema is an object type, resolving $ref references
# (self-referencing types use $ref at root level)
if wrap_non_object_output_schema and not _is_object_schema(base_schema):
# Use the wrapped result schema directly
wrapped_type = _WrappedResult[clean_output_type]
wrapped_adapter = get_cached_typeadapter(wrapped_type)
output_schema = wrapped_adapter.json_schema(mode="serialization")
output_schema["x-fastmcp-wrap-result"] = True
else:
output_schema = base_schema
output_schema = compress_schema(output_schema, prune_titles=True)
except PydanticSchemaGenerationError as e:
if "_UnserializableType" not in str(e):
logger.debug(f"Unable to generate schema for type {output_type!r}")
return cls(
fn=fn,
name=fn_name,
description=fn_doc,
input_schema=input_schema,
output_schema=output_schema or None,
)
class FunctionTool(Tool):
fn: Callable[..., Any]
def to_mcp_tool(
self,
*,
include_fastmcp_meta: bool | None = None,
**overrides: Any,
) -> mcp.types.Tool:
"""Convert the FastMCP tool to an MCP tool.
Extends the base implementation to add task execution mode if enabled.
"""
# Get base MCP tool from parent
mcp_tool = super().to_mcp_tool(
include_fastmcp_meta=include_fastmcp_meta, **overrides
)
# Add task execution mode per SEP-1686
# Only set execution if not overridden and task execution is supported
if self.task_config.supports_tasks() and "execution" not in overrides:
mcp_tool.execution = ToolExecution(taskSupport=self.task_config.mode)
return mcp_tool
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
*,
metadata: ToolMeta | None = None,
# Keep individual params for backwards compat
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
serializer: ToolResultSerializerType | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionTool:
"""Create a FunctionTool from a function.
Args:
fn: The function to wrap
metadata: ToolMeta object with all configuration. If provided,
individual parameters must not be passed.
name, title, etc.: Individual parameters for backwards compatibility.
Cannot be used together with metadata parameter.
"""
# Check mutual exclusion
individual_params_provided = (
any(
x is not None and x is not NotSet
for x in [
name,
title,
description,
icons,
tags,
annotations,
meta,
task,
serializer,
auth,
]
)
or output_schema is not NotSet
or exclude_args is not None
)
if metadata is not None and individual_params_provided:
raise TypeError(
"Cannot pass both 'metadata' and individual parameters to from_function(). "
"Use metadata alone or individual parameters alone."
)
# Build metadata from kwargs if not provided
if metadata is None:
metadata = ToolMeta(
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=task,
exclude_args=exclude_args,
serializer=serializer,
auth=auth,
)
if metadata.serializer is not None and fastmcp.settings.deprecation_warnings:
warnings.warn(
"The `serializer` parameter is deprecated. "
"Return ToolResult from your tools for full control over serialization. "
"See https://gofastmcp.com/servers/tools#custom-serialization for migration examples.",
DeprecationWarning,
stacklevel=2,
)
if metadata.exclude_args and fastmcp.settings.deprecation_warnings:
warnings.warn(
"The `exclude_args` parameter is deprecated as of FastMCP 2.14. "
"Use dependency injection with `Depends()` instead for better lifecycle management. "
"See https://gofastmcp.com/servers/dependencies for examples.",
DeprecationWarning,
stacklevel=2,
)
parsed_fn = ParsedFunction.from_function(fn, exclude_args=metadata.exclude_args)
func_name = metadata.name or parsed_fn.name
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
# Normalize task to TaskConfig
task_value = metadata.task
if task_value is None:
task_config = TaskConfig(mode="forbidden")
elif isinstance(task_value, bool):
task_config = TaskConfig.from_bool(task_value)
else:
task_config = task_value
task_config.validate_function(fn, func_name)
# Handle output_schema
if isinstance(metadata.output_schema, NotSetT):
final_output_schema = parsed_fn.output_schema
else:
final_output_schema = metadata.output_schema
if final_output_schema is not None and isinstance(final_output_schema, dict):
if not _is_object_schema(final_output_schema):
raise ValueError(
f"Output schemas must represent object types due to MCP spec limitations. "
f"Received: {final_output_schema!r}"
)
return cls(
fn=parsed_fn.fn,
name=metadata.name or parsed_fn.name,
title=metadata.title,
description=metadata.description or parsed_fn.description,
icons=metadata.icons,
parameters=parsed_fn.input_schema,
output_schema=final_output_schema,
annotations=metadata.annotations,
tags=metadata.tags or set(),
serializer=metadata.serializer,
meta=metadata.meta,
task_config=task_config,
auth=metadata.auth,
)
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Run the tool with arguments."""
wrapper_fn = without_injected_parameters(self.fn)
type_adapter = get_cached_typeadapter(wrapper_fn)
result = type_adapter.validate_python(arguments)
if inspect.isawaitable(result):
result = await result
return self.convert_result(result)
def register_with_docket(self, docket: Docket) -> None:
"""Register this tool with docket for background execution.
FunctionTool registers the underlying function, which has the user's
Depends parameters for docket to resolve.
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key])
async def add_to_docket( # type: ignore[override]
self,
docket: Docket,
arguments: dict[str, Any],
*,
fn_key: str | None = None,
task_key: str | None = None,
**kwargs: Any,
) -> Execution:
"""Schedule this tool for background execution via docket.
FunctionTool splats the arguments dict since .fn expects **kwargs.
Args:
docket: The Docket instance
arguments: Tool arguments
fn_key: Function lookup key in Docket registry (defaults to self.key)
task_key: Redis storage key for the result
**kwargs: Additional kwargs passed to docket.add()
"""
lookup_key = fn_key or self.key
if task_key:
kwargs["key"] = task_key
return await docket.add(lookup_key, **kwargs)(**arguments)
@overload
def tool(fn: F) -> F: ...
@overload
def tool(
name_or_fn: str,
*,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
exclude_args: list[str] | None = None,
serializer: Any | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[F], F]: ...
@overload
def tool(
name_or_fn: None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
exclude_args: list[str] | None = None,
serializer: Any | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[F], F]: ...
def tool(
name_or_fn: str | Callable[..., Any] | None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
exclude_args: list[str] | None = None,
serializer: Any | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Any:
"""Standalone decorator to mark a function as an MCP tool.
Returns the original function with metadata attached. Register with a server
using mcp.add_tool().
"""
if isinstance(annotations, dict):
annotations = ToolAnnotations(**annotations)
if isinstance(name_or_fn, classmethod):
raise TypeError(
"To decorate a classmethod, use @classmethod above @tool. "
"See https://gofastmcp.com/servers/tools#using-with-methods"
)
def create_tool(fn: Callable[..., Any], tool_name: str | None) -> FunctionTool:
# Create metadata first, then pass it
tool_meta = ToolMeta(
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=resolve_task_config(task),
exclude_args=exclude_args,
serializer=serializer,
auth=auth,
)
return FunctionTool.from_function(fn, metadata=tool_meta)
def attach_metadata(fn: F, tool_name: str | None) -> F:
metadata = ToolMeta(
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=task,
exclude_args=exclude_args,
serializer=serializer,
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
return fn
def decorator(fn: F, tool_name: str | None) -> F:
if fastmcp.settings.decorator_mode == "object":
warnings.warn(
"decorator_mode='object' is deprecated and will be removed in a future version. "
"Decorators now return the original function with metadata attached.",
DeprecationWarning,
stacklevel=4,
)
return create_tool(fn, tool_name) # type: ignore[return-value]
return attach_metadata(fn, tool_name)
if inspect.isroutine(name_or_fn):
return decorator(name_or_fn, name)
elif isinstance(name_or_fn, str):
if name is not None:
raise TypeError("Cannot specify name both as first argument and keyword")
tool_name = name_or_fn
elif name_or_fn is None:
tool_name = name
else:
raise TypeError(f"Invalid first argument: {type(name_or_fn)}")
def wrapper(fn: F) -> F:
return decorator(fn, tool_name)
return wrapper

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

@ -1,18 +1,13 @@
from __future__ import annotations
import inspect
import warnings
from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
from typing import (
TYPE_CHECKING,
Annotated,
Any,
ClassVar,
Generic,
TypeAlias,
get_type_hints,
overload,
)
@ -28,17 +23,10 @@ from mcp.types import (
ToolExecution,
)
from mcp.types import Tool as MCPTool
from pydantic import BaseModel, Field, PydanticSchemaGenerationError, model_validator
from typing_extensions import TypeVar
from pydantic import BaseModel, Field, model_validator
import fastmcp
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
Audio,
@ -46,9 +34,6 @@ from fastmcp.utilities.types import (
Image,
NotSet,
NotSetT,
create_function_without_params,
get_cached_typeadapter,
replace_type,
)
# Runtime type alias for auth checks to avoid circular imports with authorization.py
@ -59,23 +44,13 @@ if TYPE_CHECKING:
from docket import Docket
from docket.execution import Execution
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool_transform import ArgTransform, TransformedTool
# Re-export from function_tool module
logger = get_logger(__name__)
T = TypeVar("T", default=Any)
@dataclass
class _WrappedResult(Generic[T]):
"""Generic wrapper for non-object return types."""
result: T
class _UnserializableType:
pass
ToolResultSerializerType: TypeAlias = Callable[[Any], str]
@ -209,9 +184,11 @@ class Tool(FastMCPComponent):
),
)
@staticmethod
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
@ -226,6 +203,8 @@ class Tool(FastMCPComponent):
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionTool:
"""Create a Tool from a function."""
from fastmcp.tools.function_tool import FunctionTool
return FunctionTool.from_function(
fn=fn,
name=name,
@ -408,316 +387,6 @@ class Tool(FastMCPComponent):
)
class FunctionTool(Tool):
fn: Callable[..., Any]
def to_mcp_tool(
self,
*,
include_fastmcp_meta: bool | None = None,
**overrides: Any,
) -> MCPTool:
"""Convert the FastMCP tool to an MCP tool.
Extends the base implementation to add task execution mode if enabled.
"""
# Get base MCP tool from parent
mcp_tool = super().to_mcp_tool(
include_fastmcp_meta=include_fastmcp_meta, **overrides
)
# Add task execution mode per SEP-1686
# Only set execution if not overridden and task execution is supported
if self.task_config.supports_tasks() and "execution" not in overrides:
mcp_tool.execution = ToolExecution(taskSupport=self.task_config.mode)
return mcp_tool
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[Icon] | None = None,
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
serializer: ToolResultSerializerType | None = None, # Deprecated
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> FunctionTool:
"""Create a Tool from a function."""
if serializer is not None and fastmcp.settings.deprecation_warnings:
warnings.warn(
"The `serializer` parameter is deprecated. "
"Return ToolResult from your tools for full control over serialization. "
"See https://gofastmcp.com/servers/tools#custom-serialization for migration examples.",
DeprecationWarning,
stacklevel=2,
)
if exclude_args and fastmcp.settings.deprecation_warnings:
warnings.warn(
"The `exclude_args` parameter is deprecated as of FastMCP 2.14. "
"Use dependency injection with `Depends()` instead for better lifecycle management. "
"See https://gofastmcp.com/servers/dependencies for examples.",
DeprecationWarning,
stacklevel=2,
)
parsed_fn = ParsedFunction.from_function(fn, exclude_args=exclude_args)
func_name = name or parsed_fn.name
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
# Normalize task to TaskConfig and validate
if task is None:
task_config = TaskConfig(mode="forbidden")
elif isinstance(task, bool):
task_config = TaskConfig.from_bool(task)
else:
task_config = task
task_config.validate_function(fn, func_name)
if isinstance(output_schema, NotSetT):
final_output_schema = parsed_fn.output_schema
else:
# At this point output_schema is not NotSetT, so it must be dict | None
final_output_schema = output_schema
# Note: explicit schemas (dict) are used as-is without auto-wrapping
# Validate that explicit schemas are object type for structured content
# (resolving $ref references for self-referencing types)
if final_output_schema is not None and isinstance(final_output_schema, dict):
if not _is_object_schema(final_output_schema):
raise ValueError(
f"Output schemas must represent object types due to MCP spec limitations. Received: {final_output_schema!r}"
)
return cls(
fn=parsed_fn.fn,
name=name or parsed_fn.name,
title=title,
description=description or parsed_fn.description,
icons=icons,
parameters=parsed_fn.input_schema,
output_schema=final_output_schema,
annotations=annotations,
tags=tags or set(),
serializer=serializer,
meta=meta,
task_config=task_config,
auth=auth,
)
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Run the tool with arguments."""
wrapper_fn = without_injected_parameters(self.fn)
type_adapter = get_cached_typeadapter(wrapper_fn)
result = type_adapter.validate_python(arguments)
if inspect.isawaitable(result):
result = await result
return self.convert_result(result)
def register_with_docket(self, docket: Docket) -> None:
"""Register this tool with docket for background execution.
FunctionTool registers the underlying function, which has the user's
Depends parameters for docket to resolve.
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key])
async def add_to_docket( # type: ignore[override]
self,
docket: Docket,
arguments: dict[str, Any],
*,
fn_key: str | None = None,
task_key: str | None = None,
**kwargs: Any,
) -> Execution:
"""Schedule this tool for background execution via docket.
FunctionTool splats the arguments dict since .fn expects **kwargs.
Args:
docket: The Docket instance
arguments: Tool arguments
fn_key: Function lookup key in Docket registry (defaults to self.key)
task_key: Redis storage key for the result
**kwargs: Additional kwargs passed to docket.add()
"""
lookup_key = fn_key or self.key
if task_key:
kwargs["key"] = task_key
return await docket.add(lookup_key, **kwargs)(**arguments)
def _is_object_schema(schema: dict[str, Any]) -> bool:
"""Check if a JSON schema represents an object type."""
# Direct object type
if schema.get("type") == "object":
return True
# Schema with properties but no explicit type is treated as object
if "properties" in schema:
return True
# Self-referencing types use $ref pointing to $defs
# The referenced type is always an object in our use case
return "$ref" in schema and "$defs" in schema
@dataclass
class ParsedFunction:
fn: Callable[..., Any]
name: str
description: str | None
input_schema: dict[str, Any]
output_schema: dict[str, Any] | None
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
exclude_args: list[str] | None = None,
validate: bool = True,
wrap_non_object_output_schema: bool = True,
) -> ParsedFunction:
if validate:
sig = inspect.signature(fn)
# Reject functions with *args or **kwargs
for param in sig.parameters.values():
if param.kind == inspect.Parameter.VAR_POSITIONAL:
raise ValueError("Functions with *args are not supported as tools")
if param.kind == inspect.Parameter.VAR_KEYWORD:
raise ValueError(
"Functions with **kwargs are not supported as tools"
)
# Reject exclude_args that don't exist in the function or don't have a default value
if exclude_args:
for arg_name in exclude_args:
if arg_name not in sig.parameters:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args does not exist in function."
)
param = sig.parameters[arg_name]
if param.default == inspect.Parameter.empty:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args must have a default value."
)
# collect name and doc before we potentially modify the function
fn_name = getattr(fn, "__name__", None) or fn.__class__.__name__
fn_doc = inspect.getdoc(fn)
# if the fn is a callable class, we need to get the __call__ method from here out
if not inspect.isroutine(fn):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Handle injected parameters (Context, Docket dependencies)
wrapper_fn = without_injected_parameters(fn)
# Also handle exclude_args with non-serializable types (issue #2431)
# This must happen before Pydantic tries to serialize the parameters
if exclude_args:
wrapper_fn = create_function_without_params(wrapper_fn, list(exclude_args))
input_type_adapter = get_cached_typeadapter(wrapper_fn)
input_schema = input_type_adapter.json_schema()
# Compress and handle exclude_args
prune_params = list(exclude_args) if exclude_args else None
input_schema = compress_schema(
input_schema, prune_params=prune_params, prune_titles=True
)
output_schema = None
# Get the return annotation from the signature
sig = inspect.signature(fn)
output_type = sig.return_annotation
# If the annotation is a string (from __future__ annotations), resolve it
if isinstance(output_type, str):
try:
# Use get_type_hints to resolve the return type
# include_extras=True preserves Annotated metadata
type_hints = get_type_hints(fn, include_extras=True)
output_type = type_hints.get("return", output_type)
except Exception:
# If resolution fails, keep the string annotation
pass
if output_type not in (inspect._empty, None, Any, ...):
# there are a variety of types that we don't want to attempt to
# serialize because they are either used by FastMCP internally,
# or are MCP content types that explicitly don't form structured
# content. By replacing them with an explicitly unserializable type,
# we ensure that no output schema is automatically generated.
clean_output_type = replace_type(
output_type,
dict.fromkeys( # type: ignore[arg-type]
(
Image,
Audio,
File,
ToolResult,
mcp.types.TextContent,
mcp.types.ImageContent,
mcp.types.AudioContent,
mcp.types.ResourceLink,
mcp.types.EmbeddedResource,
),
_UnserializableType,
),
)
try:
type_adapter = get_cached_typeadapter(clean_output_type)
base_schema = type_adapter.json_schema(mode="serialization")
# Generate schema for wrapped type if it's non-object
# because MCP requires that output schemas are objects
# Check if schema is an object type, resolving $ref references
# (self-referencing types use $ref at root level)
if wrap_non_object_output_schema and not _is_object_schema(base_schema):
# Use the wrapped result schema directly
wrapped_type = _WrappedResult[clean_output_type]
wrapped_adapter = get_cached_typeadapter(wrapped_type)
output_schema = wrapped_adapter.json_schema(mode="serialization")
output_schema["x-fastmcp-wrap-result"] = True
else:
output_schema = base_schema
output_schema = compress_schema(output_schema, prune_titles=True)
except PydanticSchemaGenerationError as e:
if "_UnserializableType" not in str(e):
logger.debug(f"Unable to generate schema for type {output_type!r}")
return cls(
fn=fn,
name=fn_name,
description=fn_doc,
input_schema=input_schema,
output_schema=output_schema or None,
)
def _serialize_with_fallback(
result: Any, serializer: ToolResultSerializerType | None = None
) -> str:
@ -785,183 +454,29 @@ def _convert_to_content(
return [TextContent(type="text", text=_serialize_with_fallback(result, serializer))]
# Type alias for any function that can be decorated
AnyFunction = Callable[..., Any]
__all__ = ["Tool", "ToolResult"]
@overload
def tool(fn: AnyFunction) -> FunctionTool: ...
def __getattr__(name: str) -> Any:
"""Deprecated re-exports for backwards compatibility."""
deprecated_exports = {
"FunctionTool": "FunctionTool",
"ParsedFunction": "ParsedFunction",
"tool": "tool",
}
if name in deprecated_exports:
import fastmcp
@overload
def tool(
name_or_fn: str,
*,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], FunctionTool]: ...
@overload
def tool(
name_or_fn: None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> Callable[[AnyFunction], FunctionTool]: ...
def tool(
name_or_fn: str | AnyFunction | None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | NotSetT | None = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
) -> (
Callable[[AnyFunction], FunctionTool]
| FunctionTool
| partial[Callable[[AnyFunction], FunctionTool] | FunctionTool]
):
"""Standalone decorator to create a tool without registering it to a server.
This decorator creates a FunctionTool object from a function. Unlike
@server.tool(), this does NOT register the tool with any server - you must
explicitly add it using server.add_tool().
This is useful for:
- Creating tools that will be transformed before registration
- Defining tools in modules that are discovered by FileSystemProvider
- Creating reusable tool definitions
This decorator supports multiple calling patterns:
- @tool (without parentheses)
- @tool() (with empty parentheses)
- @tool("custom_name") (with name as first argument)
- @tool(name="custom_name") (with name as keyword argument)
Args:
name_or_fn: Either a function (when used as @tool), a string name, or None
name: Optional name for the tool (keyword-only, alternative to name_or_fn)
title: Optional title for the tool
description: Optional description of what the tool does
icons: Optional icons for the tool
tags: Optional set of tags for categorizing the tool
output_schema: Optional JSON schema for the tool's output
annotations: Optional annotations about the tool's behavior
meta: Optional meta information about the tool
task: Optional task configuration for background execution (default False)
Returns:
A FunctionTool when decorating a function, or a decorator function when
called with parameters.
Example:
```python
from fastmcp.tools import tool
from fastmcp import FastMCP
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
@tool("search_products")
def search(query: str) -> list[dict]:
return database.search(query)
# Tools are not registered yet - add them explicitly
mcp = FastMCP()
mcp.add_tool(greet)
mcp.add_tool(search)
```
"""
if isinstance(annotations, dict):
annotations = ToolAnnotations(**annotations)
if isinstance(name_or_fn, classmethod):
raise TypeError(
inspect.cleandoc(
"""
To decorate a classmethod, first define the method and then call
tool() directly on the method instead of using it as a
decorator. See https://gofastmcp.com/patterns/decorating-methods
for examples and more information.
"""
if fastmcp.settings.deprecation_warnings:
warnings.warn(
f"Importing {name} from fastmcp.tools.tool is deprecated. "
f"Import from fastmcp.tools.function_tool instead.",
DeprecationWarning,
stacklevel=2,
)
)
from fastmcp.tools import function_tool
# Determine the actual name and function based on the calling pattern
if inspect.isroutine(name_or_fn):
# Case 1: @tool (without parens) - function passed directly
fn = name_or_fn
tool_name = name # Use keyword name if provided, otherwise None
return getattr(function_tool, name)
# Default to False for standalone usage (no server to inherit from)
supports_task: bool | TaskConfig = task if task is not None else False
# Create the tool object without registration
return Tool.from_function(
fn,
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=supports_task,
auth=auth,
)
elif isinstance(name_or_fn, str):
# Case 2: @tool("custom_name") - name passed as first argument
if name is not None:
raise TypeError(
"Cannot specify both a name as first argument and as keyword argument. "
f"Use either @tool('{name_or_fn}') or @tool(name='{name}'), not both."
)
tool_name = name_or_fn
elif name_or_fn is None:
# Case 3: @tool() or @tool(name="something") - use keyword name
tool_name = name
else:
raise TypeError(
f"First argument to @tool must be a function, string, or None, got {type(name_or_fn)}"
)
# Return partial for cases where we need to wait for the function
return partial(
tool,
name=tool_name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=task,
auth=auth,
)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

3
src/fastmcp/tools/tool_transform.py
View file

@ -15,7 +15,8 @@ from pydantic.fields import Field
from pydantic.functional_validators import BeforeValidator
import fastmcp
from fastmcp.tools.tool import ParsedFunction, Tool, ToolResult, _convert_to_content
from fastmcp.tools.function_tool import ParsedFunction
from fastmcp.tools.tool import Tool, ToolResult, _convert_to_content
from fastmcp.utilities.components import _convert_set_default_none
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger

121
tests/deprecated/test_function_component_imports.py Normal file
View file

@ -0,0 +1,121 @@
"""Test that deprecated import paths for function components still work."""
import warnings
import pytest
from fastmcp.utilities.tests import temporary_settings
class TestDeprecatedFunctionToolImports:
def test_function_tool_from_tool_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning, match="Import from fastmcp.tools.function_tool"
):
from fastmcp.tools.tool import FunctionTool
# Verify it's the real class
from fastmcp.tools.function_tool import (
FunctionTool as CanonicalFunctionTool,
)
assert FunctionTool is CanonicalFunctionTool
def test_parsed_function_from_tool_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning, match="Import from fastmcp.tools.function_tool"
):
from fastmcp.tools.tool import ParsedFunction
from fastmcp.tools.function_tool import (
ParsedFunction as CanonicalParsedFunction,
)
assert ParsedFunction is CanonicalParsedFunction
def test_tool_decorator_from_tool_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning, match="Import from fastmcp.tools.function_tool"
):
from fastmcp.tools.tool import tool
from fastmcp.tools.function_tool import tool as canonical_tool
assert tool is canonical_tool
def test_no_warning_when_disabled(self):
with temporary_settings(deprecation_warnings=False):
with warnings.catch_warnings():
warnings.simplefilter("error")
from fastmcp.tools.tool import FunctionTool # noqa: F401
class TestDeprecatedFunctionResourceImports:
def test_function_resource_from_resource_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning,
match="Import from fastmcp.resources.function_resource",
):
from fastmcp.resources.resource import FunctionResource
from fastmcp.resources.function_resource import (
FunctionResource as CanonicalFunctionResource,
)
assert FunctionResource is CanonicalFunctionResource
def test_resource_decorator_from_resource_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning,
match="Import from fastmcp.resources.function_resource",
):
from fastmcp.resources.resource import resource
from fastmcp.resources.function_resource import (
resource as canonical_resource,
)
assert resource is canonical_resource
def test_no_warning_when_disabled(self):
with temporary_settings(deprecation_warnings=False):
with warnings.catch_warnings():
warnings.simplefilter("error")
from fastmcp.resources.resource import FunctionResource # noqa: F401
class TestDeprecatedFunctionPromptImports:
def test_function_prompt_from_prompt_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning, match="Import from fastmcp.prompts.function_prompt"
):
from fastmcp.prompts.prompt import FunctionPrompt
from fastmcp.prompts.function_prompt import (
FunctionPrompt as CanonicalFunctionPrompt,
)
assert FunctionPrompt is CanonicalFunctionPrompt
def test_prompt_decorator_from_prompt_module(self):
with temporary_settings(deprecation_warnings=True):
with pytest.warns(
DeprecationWarning, match="Import from fastmcp.prompts.function_prompt"
):
from fastmcp.prompts.prompt import prompt
from fastmcp.prompts.function_prompt import prompt as canonical_prompt
assert prompt is canonical_prompt
def test_no_warning_when_disabled(self):
with temporary_settings(deprecation_warnings=False):
with warnings.catch_warnings():
warnings.simplefilter("error")
from fastmcp.prompts.prompt import FunctionPrompt # noqa: F401

3
tests/deprecated/test_import_server.py
View file

@ -5,7 +5,8 @@ from mcp.types import TextContent, TextResourceContents
from fastmcp.client.client import Client
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import FunctionTool, Tool
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import Tool
from tests.conftest import get_fn_name

75
tests/prompts/test_standalone_decorator.py
View file

@ -1,39 +1,47 @@
"""Tests for the standalone @prompt decorator.
The @prompt decorator creates FunctionPrompt objects without registering them
to a server. Objects can be added explicitly via server.add_prompt() or
The @prompt decorator attaches metadata to functions without registering them
to a server. Functions can be added explicitly via server.add_prompt() or
discovered by FileSystemProvider.
"""
from typing import cast
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.prompts import FunctionPrompt, prompt
from fastmcp.prompts import prompt
from fastmcp.prompts.function_prompt import DecoratedPrompt, PromptMeta
class TestPromptDecorator:
"""Tests for the @prompt decorator."""
def test_prompt_without_parens(self):
"""@prompt without parentheses should create a FunctionPrompt."""
"""@prompt without parentheses should attach metadata."""
@prompt
def analyze(topic: str) -> str:
return f"Analyze: {topic}"
assert isinstance(analyze, FunctionPrompt)
assert analyze.name == "analyze"
decorated = cast(DecoratedPrompt, analyze)
assert callable(analyze)
assert hasattr(analyze, "__fastmcp__")
assert isinstance(decorated.__fastmcp__, PromptMeta)
assert decorated.__fastmcp__.name is None # Uses function name by default
def test_prompt_with_empty_parens(self):
"""@prompt() with empty parentheses should create a FunctionPrompt."""
"""@prompt() with empty parentheses should attach metadata."""
@prompt()
def analyze(topic: str) -> str:
return f"Analyze: {topic}"
assert isinstance(analyze, FunctionPrompt)
assert analyze.name == "analyze"
decorated = cast(DecoratedPrompt, analyze)
assert callable(analyze)
assert hasattr(analyze, "__fastmcp__")
assert isinstance(decorated.__fastmcp__, PromptMeta)
def test_prompt_with_name_arg(self):
"""@prompt("name") with name as first arg should work."""
@ -42,8 +50,10 @@ class TestPromptDecorator:
def analyze(topic: str) -> str:
return f"Analyze: {topic}"
assert isinstance(analyze, FunctionPrompt)
assert analyze.name == "custom-analyze"
decorated = cast(DecoratedPrompt, analyze)
assert callable(analyze)
assert hasattr(analyze, "__fastmcp__")
assert decorated.__fastmcp__.name == "custom-analyze"
def test_prompt_with_name_kwarg(self):
"""@prompt(name="name") with keyword arg should work."""
@ -52,8 +62,10 @@ class TestPromptDecorator:
def analyze(topic: str) -> str:
return f"Analyze: {topic}"
assert isinstance(analyze, FunctionPrompt)
assert analyze.name == "custom-analyze"
decorated = cast(DecoratedPrompt, analyze)
assert callable(analyze)
assert hasattr(analyze, "__fastmcp__")
assert decorated.__fastmcp__.name == "custom-analyze"
def test_prompt_with_all_metadata(self):
"""@prompt with all metadata should store it all."""
@ -62,29 +74,32 @@ class TestPromptDecorator:
name="custom-analyze",
title="Analysis Prompt",
description="Analyzes topics",
tags={"analysis"},
tags={"analysis", "demo"},
meta={"custom": "value"},
)
def analyze(topic: str) -> str:
return f"Analyze: {topic}"
assert isinstance(analyze, FunctionPrompt)
assert analyze.name == "custom-analyze"
assert analyze.title == "Analysis Prompt"
assert analyze.description == "Analyzes topics"
assert analyze.tags == {"analysis"}
assert analyze.meta == {"custom": "value"}
decorated = cast(DecoratedPrompt, analyze)
assert callable(analyze)
assert hasattr(analyze, "__fastmcp__")
assert decorated.__fastmcp__.name == "custom-analyze"
assert decorated.__fastmcp__.title == "Analysis Prompt"
assert decorated.__fastmcp__.description == "Analyzes topics"
assert decorated.__fastmcp__.tags == {"analysis", "demo"}
assert decorated.__fastmcp__.meta == {"custom": "value"}
async def test_prompt_can_be_rendered(self):
"""Prompt created by @prompt should be renderable."""
async def test_prompt_function_still_callable(self):
"""Decorated function should still be directly callable."""
@prompt
def analyze(topic: str) -> str:
"""Analyze a topic."""
return f"Analyze: {topic}"
return f"Please analyze: {topic}"
result = await analyze.render({"topic": "Python"})
assert result.messages[0].content.text == "Analyze: Python" # type: ignore[union-attr]
# The function is still callable even though it has metadata
result = cast(DecoratedPrompt, analyze)("Python")
assert result == "Please analyze: Python"
def test_prompt_rejects_classmethod_decorator(self):
"""@prompt should reject classmethod-decorated functions."""
@ -93,12 +108,12 @@ class TestPromptDecorator:
class MyClass:
@prompt # type: ignore[arg-type]
@classmethod
def my_prompt(cls) -> str:
return "hello"
def my_prompt(cls, topic: str) -> str:
return f"Analyze: {topic}"
def test_prompt_with_both_name_args_raises(self):
"""@prompt should raise if both positional and keyword name are given."""
with pytest.raises(TypeError, match="Cannot specify both"):
with pytest.raises(TypeError, match="Cannot specify.*both.*argument.*keyword"):
@prompt("name1", name="name2") # type: ignore[call-overload]
def my_prompt() -> str:
@ -119,5 +134,5 @@ class TestPromptDecorator:
prompts = await client.list_prompts()
assert any(p.name == "analyze" for p in prompts)
result = await client.get_prompt("analyze", {"topic": "Python"})
assert "Python" in str(result)
result = await client.get_prompt("analyze", {"topic": "FastMCP"})
assert "FastMCP" in str(result)

3
tests/resources/test_function_resources.py
View file

@ -1,7 +1,8 @@
import pytest
from pydantic import AnyUrl, BaseModel
from fastmcp.resources.resource import FunctionResource, ResourceContent
from fastmcp.resources.function_resource import FunctionResource
from fastmcp.resources.resource import ResourceContent
class TestFunctionResource:

2
tests/resources/test_resource_template.py
View file

@ -6,7 +6,7 @@ from pydantic import BaseModel
from fastmcp import Context, FastMCP
from fastmcp.resources import ResourceTemplate
from fastmcp.resources.resource import FunctionResource
from fastmcp.resources.function_resource import FunctionResource
from fastmcp.resources.template import match_uri_template

2
tests/resources/test_resources.py
View file

@ -4,7 +4,7 @@ from pydantic import AnyUrl, BaseModel
from fastmcp import Client, FastMCP
from fastmcp.resources import Resource, ResourceContent, ResourceResult
from fastmcp.resources.resource import FunctionResource
from fastmcp.resources.function_resource import FunctionResource
class TestResourceValidation:

83
tests/resources/test_standalone_decorator.py
View file

@ -1,16 +1,18 @@
"""Tests for the standalone @resource decorator.
The @resource decorator creates Resource or ResourceTemplate objects without
registering them to a server. Objects can be added explicitly via
server.add_resource() / server.add_template() or discovered by FileSystemProvider.
The @resource decorator attaches metadata to functions without registering them
to a server. Functions can be added explicitly via server.add_resource() /
server.add_template() or discovered by FileSystemProvider.
"""
from typing import cast
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.resources import FunctionResource, resource
from fastmcp.resources.template import FunctionResourceTemplate
from fastmcp.resources import resource
from fastmcp.resources.function_resource import DecoratedResource, ResourceMeta
class TestResourceDecorator:
@ -25,34 +27,41 @@ class TestResourceDecorator:
return "{}"
def test_resource_with_uri(self):
"""@resource("uri") should create a FunctionResource."""
"""@resource("uri") should attach metadata."""
@resource("config://app")
def get_config() -> dict:
return {"setting": "value"}
assert isinstance(get_config, FunctionResource)
assert str(get_config.uri) == "config://app"
decorated = cast(DecoratedResource, get_config)
assert callable(get_config)
assert hasattr(get_config, "__fastmcp__")
assert isinstance(decorated.__fastmcp__, ResourceMeta)
assert decorated.__fastmcp__.uri == "config://app"
def test_resource_with_template_uri(self):
"""@resource with template URI should create a FunctionResourceTemplate."""
"""@resource with template URI should attach metadata."""
@resource("users://{user_id}/profile")
def get_profile(user_id: str) -> dict:
return {"id": user_id}
assert isinstance(get_profile, FunctionResourceTemplate)
assert get_profile.uri_template == "users://{user_id}/profile"
decorated = cast(DecoratedResource, get_profile)
assert callable(get_profile)
assert hasattr(get_profile, "__fastmcp__")
assert decorated.__fastmcp__.uri == "users://{user_id}/profile"
def test_resource_with_function_params_becomes_template(self):
"""@resource with function params and URI params should create a template."""
"""@resource with function params should attach metadata."""
@resource("data://items/{category}")
def get_items(category: str, limit: int = 10) -> list:
return list(range(limit))
assert isinstance(get_items, FunctionResourceTemplate)
assert get_items.uri_template == "data://items/{category}"
decorated = cast(DecoratedResource, get_items)
assert callable(get_items)
assert hasattr(get_items, "__fastmcp__")
assert decorated.__fastmcp__.uri == "data://items/{category}"
def test_resource_with_all_metadata(self):
"""@resource with all metadata should store it all."""
@ -69,36 +78,39 @@ class TestResourceDecorator:
def get_config() -> dict:
return {"setting": "value"}
assert isinstance(get_config, FunctionResource)
assert str(get_config.uri) == "config://app"
assert get_config.name == "app-config"
assert get_config.title == "Application Config"
assert get_config.description == "Gets app configuration"
assert get_config.mime_type == "application/json"
assert get_config.tags == {"config"}
assert get_config.meta == {"custom": "value"}
decorated = cast(DecoratedResource, get_config)
assert callable(get_config)
assert hasattr(get_config, "__fastmcp__")
assert decorated.__fastmcp__.uri == "config://app"
assert decorated.__fastmcp__.name == "app-config"
assert decorated.__fastmcp__.title == "Application Config"
assert decorated.__fastmcp__.description == "Gets app configuration"
assert decorated.__fastmcp__.mime_type == "application/json"
assert decorated.__fastmcp__.tags == {"config"}
assert decorated.__fastmcp__.meta == {"custom": "value"}
async def test_resource_can_be_read(self):
"""Resource created by @resource should be readable."""
async def test_resource_function_still_callable(self):
"""Decorated function should still be directly callable."""
@resource("config://app")
def get_config() -> dict:
"""Get config."""
return {"setting": "value"}
assert isinstance(get_config, FunctionResource)
result = await get_config.read()
# The function is still callable even though it has metadata
result = cast(DecoratedResource, get_config)()
assert result == {"setting": "value"}
def test_resource_rejects_classmethod_decorator(self):
"""@resource should reject classmethod-decorated functions."""
with pytest.raises(TypeError, match="classmethod"):
class MyClass:
@resource("config://app") # type: ignore[arg-type]
@classmethod
def get_config(cls) -> str:
return "{}"
# Note: This now happens when added to server, not at decoration time
@resource("config://app")
def standalone() -> str:
return "{}"
# Should not raise at decoration
assert callable(standalone)
async def test_resource_added_to_server(self):
"""Resource created by @resource should work when added to a server."""
@ -108,7 +120,7 @@ class TestResourceDecorator:
"""Get config."""
return '{"version": "1.0"}'
assert isinstance(get_config, FunctionResource)
assert callable(get_config)
mcp = FastMCP("Test")
mcp.add_resource(get_config)
@ -128,10 +140,11 @@ class TestResourceDecorator:
"""Get user profile."""
return f'{{"id": "{user_id}"}}'
assert isinstance(get_profile, FunctionResourceTemplate)
assert callable(get_profile)
mcp = FastMCP("Test")
mcp.add_template(get_profile)
# add_resource handles both resources and templates based on metadata
mcp.add_resource(get_profile)
async with Client(mcp) as client:
templates = await client.list_resource_templates()

3
tests/server/middleware/test_caching.py
View file

@ -21,7 +21,8 @@ from pydantic import AnyUrl, BaseModel
from fastmcp import Context, FastMCP
from fastmcp.client.client import CallToolResult, Client
from fastmcp.client.transports import FastMCPTransport
from fastmcp.prompts.prompt import FunctionPrompt, Message, Prompt
from fastmcp.prompts.function_prompt import FunctionPrompt
from fastmcp.prompts.prompt import Message, Prompt
from fastmcp.resources.resource import Resource
from fastmcp.server.middleware.caching import (
CachableToolResult,

3
tests/server/middleware/test_tool_injection.py
View file

@ -16,7 +16,8 @@ from fastmcp.server.middleware.tool_injection import (
ResourceToolMiddleware,
ToolInjectionMiddleware,
)
from fastmcp.tools.tool import FunctionTool, Tool
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import Tool
def multiply_fn(a: int, b: int) -> int:

15
tests/server/providers/test_local_provider_prompts.py
View file

@ -10,7 +10,7 @@ from mcp.types import TextContent
from fastmcp import Client, Context, FastMCP
from fastmcp.exceptions import NotFoundError
from fastmcp.prompts.prompt import FunctionPrompt, Prompt
from fastmcp.prompts.prompt import Prompt
class TestPromptContext:
@ -247,6 +247,10 @@ class TestPromptDecorator:
async def test_prompt_direct_function_call(self):
"""Test that prompts can be registered via direct function call."""
from typing import cast
from fastmcp.prompts.function_prompt import DecoratedPrompt
mcp = FastMCP()
def standalone_function() -> str:
@ -255,11 +259,16 @@ class TestPromptDecorator:
result_fn = mcp.prompt(standalone_function, name="direct_call_prompt")
assert isinstance(result_fn, FunctionPrompt)
# In new decorator mode, returns the function with metadata
decorated = cast(DecoratedPrompt, result_fn)
assert hasattr(result_fn, "__fastmcp__")
assert decorated.__fastmcp__.name == "direct_call_prompt"
assert result_fn is standalone_function
prompts = await mcp.get_prompts()
prompt = next(p for p in prompts if p.name == "direct_call_prompt")
assert prompt is result_fn
# Prompt is registered separately, not same object as decorated function
assert prompt.name == "direct_call_prompt"
result = await mcp.render_prompt("direct_call_prompt")
assert len(result.messages) == 1

24
tests/server/providers/test_local_provider_tools.py
View file

@ -793,6 +793,10 @@ class TestToolOutputSchema:
assert result.structured_content is None
async def test_manual_structured_content(self):
from typing import cast
from fastmcp.tools.function_tool import DecoratedTool
mcp = FastMCP()
@mcp.tool
@ -801,7 +805,12 @@ class TestToolOutputSchema:
content="Hello, world!", structured_content={"message": "Hello, world!"}
)
assert f.output_schema is None
# In new decorator mode, check metadata instead of attributes
from fastmcp.utilities.types import NotSet
decorated = cast(DecoratedTool, f)
assert hasattr(f, "__fastmcp__")
assert decorated.__fastmcp__.output_schema is NotSet
result = await mcp.call_tool("f", {})
assert isinstance(result.content, list)
@ -1312,7 +1321,9 @@ class TestToolDecorator:
async def test_tool_direct_function_call(self):
"""Test that tools can be registered via direct function call."""
from fastmcp.tools import FunctionTool
from typing import cast
from fastmcp.tools.function_tool import DecoratedTool
mcp = FastMCP()
@ -1322,11 +1333,16 @@ class TestToolDecorator:
result_fn = mcp.tool(standalone_function, name="direct_call_tool")
assert isinstance(result_fn, FunctionTool)
# In new decorator mode, returns the function with metadata
decorated = cast(DecoratedTool, result_fn)
assert hasattr(result_fn, "__fastmcp__")
assert decorated.__fastmcp__.name == "direct_call_tool"
assert result_fn is standalone_function
tools = await mcp.get_tools()
tool = next(t for t in tools if t.name == "direct_call_tool")
assert tool is result_fn
# Tool is registered separately, not same object as decorated function
assert tool.name == "direct_call_tool"
result = await mcp.call_tool("direct_call_tool", {"x": 5, "y": 3})
assert result.structured_content == {"result": 8}

6
tests/server/tasks/test_sync_function_task_disabled.py
View file

@ -8,9 +8,9 @@ ValueError when task=True is used with a sync function.
import pytest
from fastmcp import FastMCP
from fastmcp.prompts.prompt import FunctionPrompt
from fastmcp.resources.resource import FunctionResource
from fastmcp.tools.tool import FunctionTool
from fastmcp.prompts.function_prompt import FunctionPrompt
from fastmcp.resources.function_resource import FunctionResource
from fastmcp.tools.function_tool import FunctionTool
async def test_sync_tool_with_explicit_task_true_raises():

6
tests/server/test_providers.py
View file

@ -7,8 +7,10 @@ import pytest
from mcp.types import AnyUrl, TextContent
from fastmcp import FastMCP
from fastmcp.prompts.prompt import FunctionPrompt, Prompt
from fastmcp.resources.resource import FunctionResource, Resource
from fastmcp.prompts.function_prompt import FunctionPrompt
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.function_resource import FunctionResource
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import FunctionResourceTemplate, ResourceTemplate
from fastmcp.server.providers import Provider
from fastmcp.tools.tool import Tool, ToolResult

63
tests/tools/test_standalone_decorator.py
View file

@ -1,39 +1,47 @@
"""Tests for the standalone @tool decorator.
The @tool decorator creates FunctionTool objects without registering them
to a server. Objects can be added explicitly via server.add_tool() or
The @tool decorator attaches metadata to functions without registering them
to a server. Functions can be added explicitly via server.add_tool() or
discovered by FileSystemProvider.
"""
from typing import cast
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.tools import FunctionTool, tool
from fastmcp.tools import tool
from fastmcp.tools.function_tool import DecoratedTool, ToolMeta
class TestToolDecorator:
"""Tests for the @tool decorator."""
def test_tool_without_parens(self):
"""@tool without parentheses should create a FunctionTool."""
"""@tool without parentheses should attach metadata."""
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
assert isinstance(greet, FunctionTool)
assert greet.name == "greet"
decorated = cast(DecoratedTool, greet)
assert callable(greet)
assert hasattr(greet, "__fastmcp__")
assert isinstance(decorated.__fastmcp__, ToolMeta)
assert decorated.__fastmcp__.name is None # Uses function name by default
def test_tool_with_empty_parens(self):
"""@tool() with empty parentheses should create a FunctionTool."""
"""@tool() with empty parentheses should attach metadata."""
@tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
assert isinstance(greet, FunctionTool)
assert greet.name == "greet"
decorated = cast(DecoratedTool, greet)
assert callable(greet)
assert hasattr(greet, "__fastmcp__")
assert isinstance(decorated.__fastmcp__, ToolMeta)
def test_tool_with_name_arg(self):
"""@tool("name") with name as first arg should work."""
@ -42,8 +50,10 @@ class TestToolDecorator:
def greet(name: str) -> str:
return f"Hello, {name}!"
assert isinstance(greet, FunctionTool)
assert greet.name == "custom-greet"
decorated = cast(DecoratedTool, greet)
assert callable(greet)
assert hasattr(greet, "__fastmcp__")
assert decorated.__fastmcp__.name == "custom-greet"
def test_tool_with_name_kwarg(self):
"""@tool(name="name") with keyword arg should work."""
@ -52,8 +62,10 @@ class TestToolDecorator:
def greet(name: str) -> str:
return f"Hello, {name}!"
assert isinstance(greet, FunctionTool)
assert greet.name == "custom-greet"
decorated = cast(DecoratedTool, greet)
assert callable(greet)
assert hasattr(greet, "__fastmcp__")
assert decorated.__fastmcp__.name == "custom-greet"
def test_tool_with_all_metadata(self):
"""@tool with all metadata should store it all."""
@ -68,23 +80,26 @@ class TestToolDecorator:
def greet(name: str) -> str:
return f"Hello, {name}!"
assert isinstance(greet, FunctionTool)
assert greet.name == "custom-greet"
assert greet.title == "Greeting Tool"
assert greet.description == "Greets people"
assert greet.tags == {"greeting", "demo"}
assert greet.meta == {"custom": "value"}
decorated = cast(DecoratedTool, greet)
assert callable(greet)
assert hasattr(greet, "__fastmcp__")
assert decorated.__fastmcp__.name == "custom-greet"
assert decorated.__fastmcp__.title == "Greeting Tool"
assert decorated.__fastmcp__.description == "Greets people"
assert decorated.__fastmcp__.tags == {"greeting", "demo"}
assert decorated.__fastmcp__.meta == {"custom": "value"}
async def test_tool_can_be_run(self):
"""Tool created by @tool should be runnable."""
async def test_tool_function_still_callable(self):
"""Decorated function should still be directly callable."""
@tool
def greet(name: str) -> str:
"""Greet someone."""
return f"Hello, {name}!"
result = await greet.run({"name": "World"})
assert result.content[0].text == "Hello, World!" # type: ignore[union-attr]
# The function is still callable even though it has metadata
result = cast(DecoratedTool, greet)("World")
assert result == "Hello, World!"
def test_tool_rejects_classmethod_decorator(self):
"""@tool should reject classmethod-decorated functions."""
@ -98,7 +113,7 @@ class TestToolDecorator:
def test_tool_with_both_name_args_raises(self):
"""@tool should raise if both positional and keyword name are given."""
with pytest.raises(TypeError, match="Cannot specify both"):
with pytest.raises(TypeError, match="Cannot specify.*both.*argument.*keyword"):
@tool("name1", name="name2") # type: ignore[call-overload]
def my_tool() -> str:

13
tests/tools/test_tool_transform.py
View file

@ -13,7 +13,8 @@ from fastmcp import FastMCP
from fastmcp.client.client import Client
from fastmcp.exceptions import ToolError
from fastmcp.tools import Tool, forward, forward_raw
from fastmcp.tools.tool import FunctionTool, ToolResult
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import ToolResult
from fastmcp.tools.tool_transform import (
ArgTransform,
ToolTransformConfig,
@ -1052,7 +1053,10 @@ class TestEnableDisable:
def add(x: int, y: int = 10) -> int:
return x + y
new_add = Tool.from_tool(add, name="new_add")
# Get the registered Tool object from the server
add_tool = await mcp._local_provider.get_component("tool:add")
assert isinstance(add_tool, Tool)
new_add = Tool.from_tool(add_tool, name="new_add")
mcp.add_tool(new_add)
# Disable original tool, but new_add should still work
@ -1076,7 +1080,10 @@ class TestEnableDisable:
def add(x: int, y: int = 10) -> int:
return x + y
new_add = Tool.from_tool(add, name="new_add")
# Get the registered Tool object from the server
add_tool = await mcp._local_provider.get_component("tool:add")
assert isinstance(add_tool, Tool)
new_add = Tool.from_tool(add_tool, name="new_add")
mcp.add_tool(new_add)
# Disable both tools via server