mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Compare commits
1 commit
main
...
refactor/l
| Author | SHA1 | Date | |
|---|---|---|---|
|
e97190ff1a |
7 changed files with 1289 additions and 1170 deletions
File diff suppressed because it is too large
Load diff
30
src/fastmcp/server/providers/local_provider/__init__.py
Normal file
30
src/fastmcp/server/providers/local_provider/__init__.py
Normal file
|
|
@ -0,0 +1,30 @@
|
|||
"""LocalProvider for locally-defined MCP components.
|
||||
|
||||
This module provides the `LocalProvider` class that manages tools, resources,
|
||||
templates, and prompts registered via decorators or direct methods.
|
||||
|
||||
LocalProvider can be used standalone and attached to multiple servers:
|
||||
|
||||
```python
|
||||
from fastmcp.server.providers import LocalProvider
|
||||
|
||||
# Create a reusable provider with tools
|
||||
provider = LocalProvider()
|
||||
|
||||
@provider.tool
|
||||
def greet(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
|
||||
# Attach to any server
|
||||
from fastmcp import FastMCP
|
||||
server1 = FastMCP("Server1", providers=[provider])
|
||||
server2 = FastMCP("Server2", providers=[provider])
|
||||
```
|
||||
"""
|
||||
|
||||
from fastmcp.server.providers.local_provider.local_provider import (
|
||||
DuplicateBehavior,
|
||||
LocalProvider,
|
||||
)
|
||||
|
||||
__all__ = ["DuplicateBehavior", "LocalProvider"]
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
"""Decorator methods for LocalProvider."""
|
||||
|
||||
from fastmcp.server.providers.local_provider.decorators.prompt import prompt
|
||||
from fastmcp.server.providers.local_provider.decorators.resource import resource
|
||||
from fastmcp.server.providers.local_provider.decorators.tool import tool
|
||||
|
||||
__all__ = ["prompt", "resource", "tool"]
|
||||
206
src/fastmcp/server/providers/local_provider/decorators/prompt.py
Normal file
206
src/fastmcp/server/providers/local_provider/decorators/prompt.py
Normal file
|
|
@ -0,0 +1,206 @@
|
|||
"""Prompt decorator for LocalProvider."""
|
||||
|
||||
import inspect
|
||||
from collections.abc import Callable
|
||||
from functools import partial
|
||||
from typing import Any, overload
|
||||
|
||||
import mcp.types
|
||||
from mcp.types import AnyFunction
|
||||
|
||||
import fastmcp
|
||||
from fastmcp.prompts.function_prompt import FunctionPrompt
|
||||
from fastmcp.prompts.prompt import Prompt
|
||||
from fastmcp.server.providers.local_provider.local_provider import LocalProvider
|
||||
from fastmcp.server.tasks.config import TaskConfig
|
||||
from fastmcp.tools.tool import AuthCheckCallable
|
||||
|
||||
|
||||
@overload
|
||||
def prompt(
|
||||
self: LocalProvider,
|
||||
name_or_fn: AnyFunction,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | None = None,
|
||||
title: str | None = None,
|
||||
description: str | None = None,
|
||||
icons: list[mcp.types.Icon] | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool = True,
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> FunctionPrompt: ...
|
||||
|
||||
|
||||
@overload
|
||||
def prompt(
|
||||
self: LocalProvider,
|
||||
name_or_fn: str | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | None = None,
|
||||
title: str | None = None,
|
||||
description: str | None = None,
|
||||
icons: list[mcp.types.Icon] | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool = True,
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionPrompt]: ...
|
||||
|
||||
|
||||
def prompt(
|
||||
self: LocalProvider,
|
||||
name_or_fn: str | AnyFunction | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | None = None,
|
||||
title: str | None = None,
|
||||
description: str | None = None,
|
||||
icons: list[mcp.types.Icon] | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool = True,
|
||||
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]
|
||||
):
|
||||
"""Decorator to register a prompt.
|
||||
|
||||
This decorator supports multiple calling patterns:
|
||||
- @provider.prompt (without parentheses)
|
||||
- @provider.prompt() (with empty parentheses)
|
||||
- @provider.prompt("custom_name") (with name as first argument)
|
||||
- @provider.prompt(name="custom_name") (with name as keyword argument)
|
||||
- provider.prompt(function, name="custom_name") (direct function call)
|
||||
|
||||
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
|
||||
enabled: Whether the prompt is enabled (default True). If False, adds to blocklist.
|
||||
meta: Optional meta information about the prompt
|
||||
task: Optional task configuration for background execution
|
||||
auth: Optional authorization checks for the prompt
|
||||
|
||||
Returns:
|
||||
The registered FunctionPrompt or a decorator function.
|
||||
|
||||
Example:
|
||||
```python
|
||||
provider = LocalProvider()
|
||||
|
||||
@provider.prompt
|
||||
def analyze(topic: str) -> list:
|
||||
return [{"role": "user", "content": f"Analyze: {topic}"}]
|
||||
|
||||
@provider.prompt("custom_name")
|
||||
def my_prompt(data: str) -> list:
|
||||
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/prompts#using-with-methods"
|
||||
)
|
||||
|
||||
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/prompts#using-with-methods"
|
||||
)
|
||||
|
||||
resolved_task: bool | TaskConfig = task if task is not None else False
|
||||
|
||||
if fastmcp.settings.decorator_mode == "object":
|
||||
prompt_obj = Prompt.from_function(
|
||||
fn,
|
||||
name=prompt_name,
|
||||
version=version,
|
||||
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,
|
||||
version=version,
|
||||
title=title,
|
||||
description=description,
|
||||
icons=icons,
|
||||
tags=tags,
|
||||
meta=meta,
|
||||
task=task,
|
||||
auth=auth,
|
||||
enabled=enabled,
|
||||
)
|
||||
target = fn.__func__ if hasattr(fn, "__func__") else fn
|
||||
target.__fastmcp__ = metadata # type: ignore[attr-defined]
|
||||
self.add_prompt(fn)
|
||||
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, # type: ignore[attr-defined]
|
||||
name=prompt_name,
|
||||
version=version,
|
||||
title=title,
|
||||
description=description,
|
||||
icons=icons,
|
||||
tags=tags,
|
||||
meta=meta,
|
||||
enabled=enabled,
|
||||
task=task,
|
||||
auth=auth,
|
||||
)
|
||||
|
|
@ -0,0 +1,154 @@
|
|||
"""Resource decorator for LocalProvider."""
|
||||
|
||||
import inspect
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import mcp.types
|
||||
from mcp.types import Annotations, AnyFunction
|
||||
|
||||
import fastmcp
|
||||
from fastmcp.resources.function_resource import resource as standalone_resource
|
||||
from fastmcp.resources.resource import Resource
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.server.providers.local_provider.local_provider import LocalProvider
|
||||
from fastmcp.server.tasks.config import TaskConfig
|
||||
from fastmcp.tools.tool import AuthCheckCallable
|
||||
|
||||
|
||||
def resource(
|
||||
self: LocalProvider,
|
||||
uri: str,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | None = None,
|
||||
title: str | None = None,
|
||||
description: str | None = None,
|
||||
icons: list[mcp.types.Icon] | None = None,
|
||||
mime_type: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool = True,
|
||||
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 | AnyFunction]:
|
||||
"""Decorator to register a function as a resource.
|
||||
|
||||
If the URI contains parameters (e.g. "resource://{param}") or the function
|
||||
has parameters, it will be registered as a template resource.
|
||||
|
||||
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
|
||||
enabled: Whether the resource is enabled (default True). If False, adds to blocklist.
|
||||
annotations: Optional annotations about the resource's behavior
|
||||
meta: Optional meta information about the resource
|
||||
task: Optional task configuration for background execution
|
||||
auth: Optional authorization checks for the resource
|
||||
|
||||
Returns:
|
||||
A decorator function.
|
||||
|
||||
Example:
|
||||
```python
|
||||
provider = LocalProvider()
|
||||
|
||||
@provider.resource("data://config")
|
||||
def get_config() -> str:
|
||||
return '{"setting": "value"}'
|
||||
|
||||
@provider.resource("data://{city}/weather")
|
||||
def get_weather(city: str) -> str:
|
||||
return f"Weather for {city}"
|
||||
```
|
||||
"""
|
||||
if isinstance(annotations, dict):
|
||||
annotations = Annotations(**annotations)
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
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/resources#using-with-methods"
|
||||
)
|
||||
|
||||
if fastmcp.settings.decorator_mode == "object":
|
||||
create_resource = standalone_resource(
|
||||
uri,
|
||||
name=name,
|
||||
version=version,
|
||||
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)
|
||||
if not enabled:
|
||||
self.disable(keys={obj.key})
|
||||
else:
|
||||
self.add_resource(obj)
|
||||
if not enabled:
|
||||
self.disable(keys={obj.key})
|
||||
return obj
|
||||
else:
|
||||
from fastmcp.resources.function_resource import ResourceMeta
|
||||
|
||||
metadata = ResourceMeta(
|
||||
uri=uri,
|
||||
name=name,
|
||||
version=version,
|
||||
title=title,
|
||||
description=description,
|
||||
icons=icons,
|
||||
tags=tags,
|
||||
mime_type=mime_type,
|
||||
annotations=annotations,
|
||||
meta=meta,
|
||||
task=task,
|
||||
auth=auth,
|
||||
enabled=enabled,
|
||||
)
|
||||
target = fn.__func__ if hasattr(fn, "__func__") else fn
|
||||
target.__fastmcp__ = metadata # type: ignore[attr-defined]
|
||||
self.add_resource(fn)
|
||||
return fn
|
||||
|
||||
return decorator
|
||||
268
src/fastmcp/server/providers/local_provider/decorators/tool.py
Normal file
268
src/fastmcp/server/providers/local_provider/decorators/tool.py
Normal file
|
|
@ -0,0 +1,268 @@
|
|||
"""Tool decorator for LocalProvider."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import warnings
|
||||
from collections.abc import Callable
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, Any, overload
|
||||
|
||||
import mcp.types
|
||||
from mcp.types import AnyFunction, ToolAnnotations
|
||||
|
||||
import fastmcp
|
||||
from fastmcp.server.tasks.config import TaskConfig
|
||||
from fastmcp.tools.function_tool import FunctionTool
|
||||
from fastmcp.tools.tool import AuthCheckCallable, Tool
|
||||
from fastmcp.utilities.types import NotSet, NotSetT
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.tools.tool import ToolResultSerializerType
|
||||
else:
|
||||
ToolResultSerializerType = Any
|
||||
|
||||
from fastmcp.server.providers.local_provider.local_provider import LocalProvider
|
||||
|
||||
|
||||
@overload
|
||||
def tool(
|
||||
self: LocalProvider,
|
||||
name_or_fn: AnyFunction,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | 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,
|
||||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
enabled: bool = True,
|
||||
task: bool | TaskConfig | None = None,
|
||||
serializer: ToolResultSerializerType | None = None, # Deprecated
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> FunctionTool: ...
|
||||
|
||||
|
||||
@overload
|
||||
def tool(
|
||||
self: LocalProvider,
|
||||
name_or_fn: str | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | 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,
|
||||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
enabled: bool = True,
|
||||
task: bool | TaskConfig | None = None,
|
||||
serializer: ToolResultSerializerType | None = None, # Deprecated
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionTool]: ...
|
||||
|
||||
|
||||
# NOTE: This method mirrors fastmcp.tools.tool() but adds registration,
|
||||
# the `enabled` param, and supports deprecated params (serializer, exclude_args).
|
||||
# When deprecated params are removed, this should delegate to the standalone
|
||||
# decorator to reduce duplication.
|
||||
def tool(
|
||||
self: LocalProvider,
|
||||
name_or_fn: str | AnyFunction | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
version: str | int | 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,
|
||||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
enabled: bool = True,
|
||||
task: bool | TaskConfig | None = None,
|
||||
serializer: ToolResultSerializerType | None = None, # Deprecated
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> (
|
||||
Callable[[AnyFunction], FunctionTool]
|
||||
| FunctionTool
|
||||
| partial[Callable[[AnyFunction], FunctionTool] | FunctionTool]
|
||||
):
|
||||
"""Decorator to register a tool.
|
||||
|
||||
This decorator supports multiple calling patterns:
|
||||
- @provider.tool (without parentheses)
|
||||
- @provider.tool() (with empty parentheses)
|
||||
- @provider.tool("custom_name") (with name as first argument)
|
||||
- @provider.tool(name="custom_name") (with name as keyword argument)
|
||||
- provider.tool(function, name="custom_name") (direct function call)
|
||||
|
||||
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
|
||||
exclude_args: Optional list of argument names to exclude from the tool schema
|
||||
meta: Optional meta information about the tool
|
||||
enabled: Whether the tool is enabled (default True). If False, adds to blocklist.
|
||||
task: Optional task configuration for background execution
|
||||
serializer: Deprecated. Return ToolResult from your tools for full control over serialization.
|
||||
|
||||
Returns:
|
||||
The registered FunctionTool or a decorator function.
|
||||
|
||||
Example:
|
||||
```python
|
||||
provider = LocalProvider()
|
||||
|
||||
@provider.tool
|
||||
def greet(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
|
||||
@provider.tool("custom_name")
|
||||
def my_tool(x: int) -> str:
|
||||
return str(x)
|
||||
```
|
||||
"""
|
||||
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 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 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"
|
||||
)
|
||||
|
||||
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,
|
||||
version=version,
|
||||
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,
|
||||
timeout=timeout,
|
||||
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,
|
||||
version=version,
|
||||
title=title,
|
||||
description=description,
|
||||
icons=icons,
|
||||
tags=tags,
|
||||
output_schema=output_schema,
|
||||
annotations=annotations,
|
||||
meta=meta,
|
||||
task=task,
|
||||
exclude_args=exclude_args,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
enabled=enabled,
|
||||
)
|
||||
target = fn.__func__ if hasattr(fn, "__func__") else fn
|
||||
target.__fastmcp__ = metadata # type: ignore[attr-defined]
|
||||
tool_obj = self.add_tool(fn)
|
||||
return fn
|
||||
|
||||
if inspect.isroutine(name_or_fn):
|
||||
return decorate_and_register(name_or_fn, name)
|
||||
|
||||
elif isinstance(name_or_fn, str):
|
||||
# Case 3: @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 4: @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(
|
||||
self.tool, # type: ignore[attr-defined]
|
||||
name=tool_name,
|
||||
version=version,
|
||||
title=title,
|
||||
description=description,
|
||||
icons=icons,
|
||||
tags=tags,
|
||||
output_schema=output_schema,
|
||||
annotations=annotations,
|
||||
exclude_args=exclude_args,
|
||||
meta=meta,
|
||||
enabled=enabled,
|
||||
task=task,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
624
src/fastmcp/server/providers/local_provider/local_provider.py
Normal file
624
src/fastmcp/server/providers/local_provider/local_provider.py
Normal file
|
|
@ -0,0 +1,624 @@
|
|||
"""LocalProvider for locally-defined MCP components.
|
||||
|
||||
This module provides the `LocalProvider` class that manages tools, resources,
|
||||
templates, and prompts registered via decorators or direct methods.
|
||||
|
||||
LocalProvider can be used standalone and attached to multiple servers:
|
||||
|
||||
```python
|
||||
from fastmcp.server.providers import LocalProvider
|
||||
|
||||
# Create a reusable provider with tools
|
||||
provider = LocalProvider()
|
||||
|
||||
@provider.tool
|
||||
def greet(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
|
||||
# Attach to any server
|
||||
from fastmcp import FastMCP
|
||||
server1 = FastMCP("Server1", providers=[provider])
|
||||
server2 = FastMCP("Server2", providers=[provider])
|
||||
```
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from collections.abc import Callable, Sequence
|
||||
from typing import TYPE_CHECKING, Any, Literal, TypeVar
|
||||
|
||||
from fastmcp.prompts.prompt import Prompt
|
||||
from fastmcp.resources.resource import Resource
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.server.providers.base import Provider
|
||||
from fastmcp.tools.tool import Tool
|
||||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.versions import VersionSpec, version_sort_key
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
DuplicateBehavior = Literal["error", "warn", "replace", "ignore"]
|
||||
|
||||
_C = TypeVar("_C", bound=FastMCPComponent)
|
||||
|
||||
|
||||
class LocalProvider(Provider):
|
||||
"""Provider for locally-defined components.
|
||||
|
||||
Supports decorator-based registration (`@provider.tool`, `@provider.resource`,
|
||||
`@provider.prompt`) and direct object registration methods.
|
||||
|
||||
When used standalone, LocalProvider uses default settings. When attached
|
||||
to a FastMCP server via the server's decorators, server-level settings
|
||||
like `_tool_serializer` and `_support_tasks_by_default` are injected.
|
||||
|
||||
Example:
|
||||
```python
|
||||
from fastmcp.server.providers import LocalProvider
|
||||
|
||||
# Standalone usage
|
||||
provider = LocalProvider()
|
||||
|
||||
@provider.tool
|
||||
def greet(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
|
||||
@provider.resource("data://config")
|
||||
def get_config() -> str:
|
||||
return '{"setting": "value"}'
|
||||
|
||||
@provider.prompt
|
||||
def analyze(topic: str) -> list:
|
||||
return [{"role": "user", "content": f"Analyze: {topic}"}]
|
||||
|
||||
# Attach to server(s)
|
||||
from fastmcp import FastMCP
|
||||
server = FastMCP("MyServer", providers=[provider])
|
||||
```
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
on_duplicate: DuplicateBehavior = "error",
|
||||
) -> None:
|
||||
"""Initialize a LocalProvider with empty storage.
|
||||
|
||||
Args:
|
||||
on_duplicate: Behavior when adding a component that already exists:
|
||||
- "error": Raise ValueError
|
||||
- "warn": Log warning and replace
|
||||
- "replace": Silently replace
|
||||
- "ignore": Keep existing, return it
|
||||
"""
|
||||
super().__init__()
|
||||
self._on_duplicate = on_duplicate
|
||||
# Unified component storage - keyed by prefixed key (e.g., "tool:name", "resource:uri")
|
||||
self._components: dict[str, FastMCPComponent] = {}
|
||||
|
||||
# =========================================================================
|
||||
# Storage methods
|
||||
# =========================================================================
|
||||
|
||||
def _get_component_identity(self, component: FastMCPComponent) -> tuple[type, str]:
|
||||
"""Get the identity (type, name/uri) for a component.
|
||||
|
||||
Returns:
|
||||
A tuple of (component_type, logical_name) where logical_name is
|
||||
the name for tools/prompts or URI for resources/templates.
|
||||
"""
|
||||
if isinstance(component, Tool):
|
||||
return (Tool, component.name)
|
||||
elif isinstance(component, ResourceTemplate):
|
||||
return (ResourceTemplate, component.uri_template)
|
||||
elif isinstance(component, Resource):
|
||||
return (Resource, str(component.uri))
|
||||
elif isinstance(component, Prompt):
|
||||
return (Prompt, component.name)
|
||||
else:
|
||||
# Fall back to key without version suffix
|
||||
key = component.key
|
||||
base_key = key.rsplit("@", 1)[0] if "@" in key else key
|
||||
return (type(component), base_key)
|
||||
|
||||
def _check_version_mixing(self, component: _C) -> None:
|
||||
"""Check that versioned and unversioned components aren't mixed.
|
||||
|
||||
LocalProvider enforces a simple rule: for any given name/URI, all
|
||||
registered components must either be versioned or unversioned, not both.
|
||||
This prevents confusing situations where unversioned components can't
|
||||
be filtered out by version filters.
|
||||
|
||||
Args:
|
||||
component: The component being added.
|
||||
|
||||
Raises:
|
||||
ValueError: If adding would mix versioned and unversioned components.
|
||||
"""
|
||||
comp_type, logical_name = self._get_component_identity(component)
|
||||
is_versioned = component.version is not None
|
||||
|
||||
# Check all existing components of the same type and logical name
|
||||
for existing in self._components.values():
|
||||
if not isinstance(existing, comp_type):
|
||||
continue
|
||||
|
||||
_, existing_name = self._get_component_identity(existing)
|
||||
if existing_name != logical_name:
|
||||
continue
|
||||
|
||||
existing_versioned = existing.version is not None
|
||||
if is_versioned != existing_versioned:
|
||||
type_name = comp_type.__name__.lower()
|
||||
if is_versioned:
|
||||
raise ValueError(
|
||||
f"Cannot add versioned {type_name} {logical_name!r} "
|
||||
f"(version={component.version!r}): an unversioned "
|
||||
f"{type_name} with this name already exists. "
|
||||
f"Either version all components or none."
|
||||
)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Cannot add unversioned {type_name} {logical_name!r}: "
|
||||
f"versioned {type_name}s with this name already exist "
|
||||
f"(e.g., version={existing.version!r}). "
|
||||
f"Either version all components or none."
|
||||
)
|
||||
|
||||
def _add_component(self, component: _C) -> _C:
|
||||
"""Add a component to unified storage.
|
||||
|
||||
Args:
|
||||
component: The component to add.
|
||||
|
||||
Returns:
|
||||
The component that was added (or existing if on_duplicate="ignore").
|
||||
"""
|
||||
existing = self._components.get(component.key)
|
||||
if existing:
|
||||
if self._on_duplicate == "error":
|
||||
raise ValueError(f"Component already exists: {component.key}")
|
||||
elif self._on_duplicate == "warn":
|
||||
logger.warning(f"Component already exists: {component.key}")
|
||||
elif self._on_duplicate == "ignore":
|
||||
return existing # type: ignore[return-value]
|
||||
# "replace" and "warn" fall through to add
|
||||
|
||||
# Check for versioned/unversioned mixing before adding
|
||||
self._check_version_mixing(component)
|
||||
|
||||
self._components[component.key] = component
|
||||
return component
|
||||
|
||||
def _remove_component(self, key: str) -> None:
|
||||
"""Remove a component from unified storage.
|
||||
|
||||
Args:
|
||||
key: The prefixed key of the component.
|
||||
|
||||
Raises:
|
||||
KeyError: If the component is not found.
|
||||
"""
|
||||
component = self._components.get(key)
|
||||
if component is None:
|
||||
raise KeyError(f"Component {key!r} not found")
|
||||
|
||||
del self._components[key]
|
||||
|
||||
def _get_component(self, key: str) -> FastMCPComponent | None:
|
||||
"""Get a component by its prefixed key.
|
||||
|
||||
Args:
|
||||
key: The prefixed key (e.g., "tool:name", "resource:uri").
|
||||
|
||||
Returns:
|
||||
The component, or None if not found.
|
||||
"""
|
||||
return self._components.get(key)
|
||||
|
||||
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 isinstance(tool, Tool):
|
||||
self._add_component(tool)
|
||||
return tool
|
||||
|
||||
# Handle decorated function with metadata
|
||||
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
|
||||
enabled = meta.enabled
|
||||
tool_obj = Tool.from_function(
|
||||
tool,
|
||||
name=meta.name,
|
||||
version=meta.version,
|
||||
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,
|
||||
timeout=meta.timeout,
|
||||
auth=meta.auth,
|
||||
)
|
||||
else:
|
||||
tool_obj = Tool.from_function(tool)
|
||||
enabled = True
|
||||
|
||||
self._add_component(tool_obj)
|
||||
if not enabled:
|
||||
self.disable(keys={tool_obj.key})
|
||||
return tool_obj
|
||||
|
||||
def remove_tool(self, name: str, version: str | None = None) -> None:
|
||||
"""Remove tool(s) from this provider's storage.
|
||||
|
||||
Args:
|
||||
name: The tool name.
|
||||
version: If None, removes ALL versions. If specified, removes only that version.
|
||||
|
||||
Raises:
|
||||
KeyError: If no matching tool is found.
|
||||
"""
|
||||
if version is None:
|
||||
# Remove all versions
|
||||
keys_to_remove = [
|
||||
k
|
||||
for k, c in self._components.items()
|
||||
if isinstance(c, Tool) and c.name == name
|
||||
]
|
||||
if not keys_to_remove:
|
||||
raise KeyError(f"Tool {name!r} not found")
|
||||
for key in keys_to_remove:
|
||||
self._remove_component(key)
|
||||
else:
|
||||
# Remove specific version - key format is "tool:name@version"
|
||||
key = f"{Tool.make_key(name)}@{version}"
|
||||
if key not in self._components:
|
||||
raise KeyError(f"Tool {name!r} version {version!r} not found")
|
||||
self._remove_component(key)
|
||||
|
||||
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 isinstance(resource, (Resource, ResourceTemplate)):
|
||||
self._add_component(resource)
|
||||
return resource
|
||||
|
||||
# Handle decorated function with metadata
|
||||
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
|
||||
enabled = meta.enabled
|
||||
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_obj = ResourceTemplate.from_function(
|
||||
fn=resource,
|
||||
uri_template=meta.uri,
|
||||
name=meta.name,
|
||||
version=meta.version,
|
||||
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_obj = Resource.from_function(
|
||||
fn=resource,
|
||||
uri=meta.uri,
|
||||
name=meta.name,
|
||||
version=meta.version,
|
||||
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."
|
||||
)
|
||||
|
||||
self._add_component(resource_obj)
|
||||
if not enabled:
|
||||
self.disable(keys={resource_obj.key})
|
||||
return resource_obj
|
||||
|
||||
def remove_resource(self, uri: str, version: str | None = None) -> None:
|
||||
"""Remove resource(s) from this provider's storage.
|
||||
|
||||
Args:
|
||||
uri: The resource URI.
|
||||
version: If None, removes ALL versions. If specified, removes only that version.
|
||||
|
||||
Raises:
|
||||
KeyError: If no matching resource is found.
|
||||
"""
|
||||
if version is None:
|
||||
# Remove all versions
|
||||
keys_to_remove = [
|
||||
k
|
||||
for k, c in self._components.items()
|
||||
if isinstance(c, Resource) and str(c.uri) == uri
|
||||
]
|
||||
if not keys_to_remove:
|
||||
raise KeyError(f"Resource {uri!r} not found")
|
||||
for key in keys_to_remove:
|
||||
self._remove_component(key)
|
||||
else:
|
||||
# Remove specific version
|
||||
key = f"{Resource.make_key(uri)}@{version}"
|
||||
if key not in self._components:
|
||||
raise KeyError(f"Resource {uri!r} version {version!r} not found")
|
||||
self._remove_component(key)
|
||||
|
||||
def add_template(self, template: ResourceTemplate) -> ResourceTemplate:
|
||||
"""Add a resource template to this provider's storage."""
|
||||
return self._add_component(template)
|
||||
|
||||
def remove_template(self, uri_template: str, version: str | None = None) -> None:
|
||||
"""Remove resource template(s) from this provider's storage.
|
||||
|
||||
Args:
|
||||
uri_template: The template URI pattern.
|
||||
version: If None, removes ALL versions. If specified, removes only that version.
|
||||
|
||||
Raises:
|
||||
KeyError: If no matching template is found.
|
||||
"""
|
||||
if version is None:
|
||||
# Remove all versions
|
||||
keys_to_remove = [
|
||||
k
|
||||
for k, c in self._components.items()
|
||||
if isinstance(c, ResourceTemplate) and c.uri_template == uri_template
|
||||
]
|
||||
if not keys_to_remove:
|
||||
raise KeyError(f"Template {uri_template!r} not found")
|
||||
for key in keys_to_remove:
|
||||
self._remove_component(key)
|
||||
else:
|
||||
# Remove specific version
|
||||
key = f"{ResourceTemplate.make_key(uri_template)}@{version}"
|
||||
if key not in self._components:
|
||||
raise KeyError(
|
||||
f"Template {uri_template!r} version {version!r} not found"
|
||||
)
|
||||
self._remove_component(key)
|
||||
|
||||
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 isinstance(prompt, Prompt):
|
||||
self._add_component(prompt)
|
||||
return prompt
|
||||
|
||||
# Handle decorated function with metadata
|
||||
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
|
||||
enabled = meta.enabled
|
||||
prompt_obj = Prompt.from_function(
|
||||
prompt,
|
||||
name=meta.name,
|
||||
version=meta.version,
|
||||
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."
|
||||
)
|
||||
|
||||
self._add_component(prompt_obj)
|
||||
if not enabled:
|
||||
self.disable(keys={prompt_obj.key})
|
||||
return prompt_obj
|
||||
|
||||
def remove_prompt(self, name: str, version: str | None = None) -> None:
|
||||
"""Remove prompt(s) from this provider's storage.
|
||||
|
||||
Args:
|
||||
name: The prompt name.
|
||||
version: If None, removes ALL versions. If specified, removes only that version.
|
||||
|
||||
Raises:
|
||||
KeyError: If no matching prompt is found.
|
||||
"""
|
||||
if version is None:
|
||||
# Remove all versions
|
||||
keys_to_remove = [
|
||||
k
|
||||
for k, c in self._components.items()
|
||||
if isinstance(c, Prompt) and c.name == name
|
||||
]
|
||||
if not keys_to_remove:
|
||||
raise KeyError(f"Prompt {name!r} not found")
|
||||
for key in keys_to_remove:
|
||||
self._remove_component(key)
|
||||
else:
|
||||
# Remove specific version
|
||||
key = f"{Prompt.make_key(name)}@{version}"
|
||||
if key not in self._components:
|
||||
raise KeyError(f"Prompt {name!r} version {version!r} not found")
|
||||
self._remove_component(key)
|
||||
|
||||
# =========================================================================
|
||||
# Provider interface implementation
|
||||
# =========================================================================
|
||||
|
||||
async def _list_tools(self) -> Sequence[Tool]:
|
||||
"""Return all tools."""
|
||||
return [v for v in self._components.values() if isinstance(v, Tool)]
|
||||
|
||||
async def _get_tool(
|
||||
self, name: str, version: VersionSpec | None = None
|
||||
) -> Tool | None:
|
||||
"""Get a tool by name.
|
||||
|
||||
Args:
|
||||
name: The tool name.
|
||||
version: Optional version filter. If None, returns highest version.
|
||||
"""
|
||||
matching = [
|
||||
v
|
||||
for v in self._components.values()
|
||||
if isinstance(v, Tool) and v.name == name
|
||||
]
|
||||
if version:
|
||||
matching = [t for t in matching if version.matches(t.version)]
|
||||
if not matching:
|
||||
return None
|
||||
return max(matching, key=version_sort_key) # type: ignore[type-var]
|
||||
|
||||
async def _list_resources(self) -> Sequence[Resource]:
|
||||
"""Return all resources."""
|
||||
return [v for v in self._components.values() if isinstance(v, Resource)]
|
||||
|
||||
async def _get_resource(
|
||||
self, uri: str, version: VersionSpec | None = None
|
||||
) -> Resource | None:
|
||||
"""Get a resource by URI.
|
||||
|
||||
Args:
|
||||
uri: The resource URI.
|
||||
version: Optional version filter. If None, returns highest version.
|
||||
"""
|
||||
matching = [
|
||||
v
|
||||
for v in self._components.values()
|
||||
if isinstance(v, Resource) and str(v.uri) == uri
|
||||
]
|
||||
if version:
|
||||
matching = [r for r in matching if version.matches(r.version)]
|
||||
if not matching:
|
||||
return None
|
||||
return max(matching, key=version_sort_key) # type: ignore[type-var]
|
||||
|
||||
async def _list_resource_templates(self) -> Sequence[ResourceTemplate]:
|
||||
"""Return all resource templates."""
|
||||
return [v for v in self._components.values() if isinstance(v, ResourceTemplate)]
|
||||
|
||||
async def _get_resource_template(
|
||||
self, uri: str, version: VersionSpec | None = None
|
||||
) -> ResourceTemplate | None:
|
||||
"""Get a resource template that matches the given URI.
|
||||
|
||||
Args:
|
||||
uri: The URI to match against templates.
|
||||
version: Optional version filter. If None, returns highest version.
|
||||
"""
|
||||
# Find all templates that match the URI
|
||||
matching = [
|
||||
component
|
||||
for component in self._components.values()
|
||||
if isinstance(component, ResourceTemplate)
|
||||
and component.matches(uri) is not None
|
||||
]
|
||||
if version:
|
||||
matching = [t for t in matching if version.matches(t.version)]
|
||||
if not matching:
|
||||
return None
|
||||
return max(matching, key=version_sort_key) # type: ignore[type-var]
|
||||
|
||||
async def _list_prompts(self) -> Sequence[Prompt]:
|
||||
"""Return all prompts."""
|
||||
return [v for v in self._components.values() if isinstance(v, Prompt)]
|
||||
|
||||
async def _get_prompt(
|
||||
self, name: str, version: VersionSpec | None = None
|
||||
) -> Prompt | None:
|
||||
"""Get a prompt by name.
|
||||
|
||||
Args:
|
||||
name: The prompt name.
|
||||
version: Optional version filter. If None, returns highest version.
|
||||
"""
|
||||
matching = [
|
||||
v
|
||||
for v in self._components.values()
|
||||
if isinstance(v, Prompt) and v.name == name
|
||||
]
|
||||
if version:
|
||||
matching = [p for p in matching if version.matches(p.version)]
|
||||
if not matching:
|
||||
return None
|
||||
return max(matching, key=version_sort_key) # type: ignore[type-var]
|
||||
|
||||
# =========================================================================
|
||||
# Task registration
|
||||
# =========================================================================
|
||||
|
||||
async def get_tasks(self) -> Sequence[FastMCPComponent]:
|
||||
"""Return components eligible for background task execution.
|
||||
|
||||
Returns components that have task_config.mode != 'forbidden'.
|
||||
This includes both FunctionTool/Resource/Prompt instances created via
|
||||
decorators and custom Tool/Resource/Prompt subclasses.
|
||||
"""
|
||||
return [c for c in self._components.values() if c.task_config.supports_tasks()]
|
||||
|
||||
# Decorator methods - implementations in decorators/ subpackage
|
||||
# Imported here to avoid circular imports and keep type checker happy
|
||||
def tool(self, *args: Any, **kwargs: Any) -> Any:
|
||||
"""Tool decorator - see decorators/tool.py for implementation."""
|
||||
from fastmcp.server.providers.local_provider.decorators.tool import tool
|
||||
|
||||
return tool(self, *args, **kwargs)
|
||||
|
||||
def resource(self, *args: Any, **kwargs: Any) -> Any:
|
||||
"""Resource decorator - see decorators/resource.py for implementation."""
|
||||
from fastmcp.server.providers.local_provider.decorators.resource import resource
|
||||
|
||||
return resource(self, *args, **kwargs)
|
||||
|
||||
def prompt(self, *args: Any, **kwargs: Any) -> Any:
|
||||
"""Prompt decorator - see decorators/prompt.py for implementation."""
|
||||
from fastmcp.server.providers.local_provider.decorators.prompt import prompt
|
||||
|
||||
return prompt(self, *args, **kwargs)
|
||||
Loading…
Add table
Add a link
Reference in a new issue