ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-31 03:23:19 +02:00
Code Issues Projects Releases Packages Wiki Activity
fastmcp/fastmcp_slim/fastmcp/server/server.py
Jeremiah Lowin 8ba5b89918
 Add server-level cache hints (SEP-2549) (#4464)
2026-07-08 09:10:41 -04:00

2316 lines
91 KiB
Python
Raw Permalink Blame History

"""FastMCP - A more ergonomic interface for MCP servers."""
from __future__ import annotations
import asyncio
import logging
import secrets
from collections.abc import (
AsyncIterator,
Callable,
Sequence,
)
from contextlib import (
AbstractAsyncContextManager,
asynccontextmanager,
)
from dataclasses import replace
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, cast, overload
import httpx
import mcp_types
from key_value.aio.adapters.pydantic import PydanticAdapter
from key_value.aio.protocols import AsyncKeyValue
from key_value.aio.stores.memory import MemoryStore
from mcp.server.lowlevel.server import LifespanResultT
from mcp.shared.exceptions import MCPError
from mcp_types import (
Annotations,
CallToolRequestParams,
ToolAnnotations,
)
from pydantic import AnyUrl
from pydantic import ValidationError as PydanticValidationError
from starlette.routing import BaseRoute
from typing_extensions import Self
import fastmcp
import fastmcp.server
from fastmcp.apps.config import (
AppConfig,
app_config_to_meta_dict,
resolve_ui_mime_type,
)
from fastmcp.exceptions import (
AuthorizationError,
FastMCPError,
NotFoundError,
PromptError,
ResourceError,
ToolError,
ValidationError,
)
from fastmcp.mcp_config import MCPConfig
from fastmcp.prompts import Prompt
from fastmcp.prompts.base import PromptResult
from fastmcp.prompts.function_prompt import FunctionPrompt
from fastmcp.resources.base import Resource, ResourceResult
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.auth import AuthCheck, AuthContext, AuthProvider, run_auth_checks
from fastmcp.server.caching import build_cache_hints
from fastmcp.server.lifespan import Lifespan
from fastmcp.server.low_level import LowLevelServer
from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
from fastmcp.server.mixins import LifespanMixin, MCPOperationsMixin, TransportMixin
from fastmcp.server.providers import LocalProvider, Provider
from fastmcp.server.providers.aggregate import AggregateProvider
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
from fastmcp.server.telemetry import server_span
from fastmcp.server.transforms import (
ToolTransform,
Transform,
)
from fastmcp.server.transforms.visibility import apply_session_transforms, is_enabled
from fastmcp.settings import DuplicateBehavior as DuplicateBehaviorSetting
from fastmcp.tools.base import Tool, ToolResult
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool_transform import ToolTransformConfig
from fastmcp.utilities.components import FastMCPComponent, _coerce_version
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import AnyFunction, FastMCPBaseModel, NotSet, NotSetT
from fastmcp.utilities.versions import (
VersionSpec,
version_sort_key,
)
if TYPE_CHECKING:
from fastmcp.client import Client
from fastmcp.client.client import SDKServer
from fastmcp.client.sampling import SamplingHandler
from fastmcp.client.transports import ClientTransport, ClientTransportT
from fastmcp.server.providers.openapi import ComponentFn as OpenAPIComponentFn
from fastmcp.server.providers.openapi import RouteMap
from fastmcp.server.providers.openapi import RouteMapFn as OpenAPIRouteMapFn
from fastmcp.server.providers.proxy import FastMCPProxy
logger = get_logger(__name__)
def _version_request_meta(
version: VersionSpec | None,
) -> dict[str, Any] | None:
if version is None:
return None
if version.eq is not None and version.gte is None and version.lt is None:
version_value: str | dict[str, str] = version.eq
else:
version_value = {
key: value
for key, value in {
"gte": version.gte,
"lt": version.lt,
"eq": version.eq,
}.items()
if value is not None
}
if not version_value:
return None
# SDK v2: request `_meta` is a plain dict (the `Meta` type alias), not the
# old `RequestParams.Meta` nested model.
return {"fastmcp": {"version": version_value}}
# The MCP SDK warns "Tool X not listed, no validation will be performed"
# for every call to app-only tools (hidden from list_tools by design).
# This fires even when validate_input=False. Suppress it.
class _SuppressUnlistedToolWarning(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
return "not listed, no validation" not in record.getMessage()
logging.getLogger("mcp.server.lowlevel.server").addFilter(
_SuppressUnlistedToolWarning()
)
F = TypeVar("F", bound=Callable[..., Any])
DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]
_REMOVED_KWARGS: dict[str, str] = {
"host": "Pass `host` to `run_http_async()`, or set FASTMCP_HOST.",
"port": "Pass `port` to `run_http_async()`, or set FASTMCP_PORT.",
"sse_path": "Pass `path` to `run_http_async()` or `http_app()`, or set FASTMCP_SSE_PATH.",
"message_path": "Set FASTMCP_MESSAGE_PATH.",
"streamable_http_path": "Pass `path` to `run_http_async()` or `http_app()`, or set FASTMCP_STREAMABLE_HTTP_PATH.",
"json_response": "Pass `json_response` to `run_http_async()` or `http_app()`, or set FASTMCP_JSON_RESPONSE.",
"stateless_http": "Pass `stateless_http` to `run_http_async()` or `http_app()`, or set FASTMCP_STATELESS_HTTP.",
"debug": "Set FASTMCP_DEBUG.",
"log_level": "Pass `log_level` to `run_http_async()`, or set FASTMCP_LOG_LEVEL.",
"on_duplicate_tools": "Use `on_duplicate=` instead.",
"on_duplicate_resources": "Use `on_duplicate=` instead.",
"on_duplicate_prompts": "Use `on_duplicate=` instead.",
"tool_serializer": "Return ToolResult from your tools instead. See https://gofastmcp.com/servers/tools#custom-serialization",
"include_tags": "Use `server.enable(tags=..., only=True)` after creating the server.",
"exclude_tags": "Use `server.disable(tags=...)` after creating the server.",
"tool_transformations": "Use `server.add_transform(ToolTransform(...))` after creating the server.",
}
def _check_removed_kwargs(kwargs: dict[str, Any]) -> None:
"""Raise helpful TypeErrors for kwargs removed in v3."""
for key in kwargs:
if key in _REMOVED_KWARGS:
raise TypeError(
f"FastMCP() no longer accepts `{key}`. {_REMOVED_KWARGS[key]}"
)
if kwargs:
raise TypeError(
f"FastMCP() got unexpected keyword argument(s): {', '.join(repr(k) for k in kwargs)}"
)
Transport = Literal["stdio", "http", "sse", "streamable-http"]
LifespanCallable = Callable[
["FastMCP[LifespanResultT]"], AbstractAsyncContextManager[LifespanResultT]
]
def _get_auth_context() -> tuple[bool, Any]:
"""Get auth context for the current request.
Returns a tuple of (skip_auth, token) where:
- skip_auth=True means auth checks should be skipped (STDIO transport)
- token is the access token for HTTP transports (may be None if unauthenticated)
Uses late import to avoid circular import with context.py.
"""
from fastmcp.server.context import _current_transport
is_stdio = _current_transport.get() == "stdio"
if is_stdio:
return (True, None)
from fastmcp.server.dependencies import get_access_token
return (False, get_access_token())
def _is_backend_tool(tool: Tool) -> bool:
"""Check whether a tool is handled specially as backend tool
Tools registered via ``@app.tool()`` (without ``model=True``) have
``meta["ui"]["visibility"] == ["app"]`` — they are callable by app UIs
but should not appear in tool list the client passes to the model.
They are handled specially for in various ways - e.g. they are looked
up via get_app_tool(), and don't appear in the tools/list output.
(FIXME: the latter isn't correct behavior according to the mcp-apps spec.)
Returns True (a backend tool) when:
- The tool has ``meta.fastmcp.app``.
- The tool has ``meta.ui.visibility``.
- The visibility is precisely ``["app"]``.
Returns False otherwise.
"""
meta = tool.meta
if not meta:
return False
fastmcp = meta.get("fastmcp")
if not isinstance(fastmcp, dict):
return False
if fastmcp.get("app") is None:
return False
ui = meta.get("ui")
if not isinstance(ui, dict):
return False
visibility = ui.get("visibility")
if not isinstance(visibility, list):
return False
return len(visibility) == 1 and visibility[0] == "app"
def _is_app_visible(tool: Tool) -> bool:
"""Check whether a tool has explicitly opted into app-callable visibility.
Gates the dispatcher's hashed-name routing path: only tools whose
``meta.ui.visibility`` list contains ``"app"`` can be reached via
``<hash>_<local_name>`` calls. Tools without an explicit visibility
declaration are NOT app-callable — they must be reached by their
display name through the normal transform-aware resolution path.
This is the inverse of the "everything is dot-callable" trap: the
hashed-name path is an opt-in mechanism for FastMCPApp backend tools,
not a general bypass for arbitrary tools.
"""
meta = tool.meta
if not meta:
return False
ui = meta.get("ui")
if not isinstance(ui, dict):
return False
visibility = ui.get("visibility")
if not isinstance(visibility, list):
return False
return "app" in visibility
@asynccontextmanager
async def default_lifespan(server: FastMCP[LifespanResultT]) -> AsyncIterator[Any]:
"""Default lifespan context manager that does nothing.
Args:
server: The server instance this lifespan is managing
Returns:
An empty dictionary as the lifespan result.
"""
yield {}
def _lifespan_proxy(
fastmcp_server: FastMCP[LifespanResultT],
) -> Callable[
[LowLevelServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
]:
@asynccontextmanager
async def wrap(
low_level_server: LowLevelServer[LifespanResultT],
) -> AsyncIterator[LifespanResultT]:
# Drive the FastMCP lifespan rather than merely reading it back. The
# SDK enters this proxy exactly once per manager/server run (via
# ``StreamableHTTPSessionManager.run`` → ``app.lifespan(app)`` or
# ``Server.run`` → ``self.lifespan(self)``) and reuses the yielded
# state for every session. ``_lifespan_manager`` is ref-counted, so
# when an outer caller (``run_http_async``/``run_stdio_async``) has
# already entered it, this nested entry reuses the existing result
# instead of re-running setup.
async with fastmcp_server._lifespan_manager():
yield fastmcp_server._lifespan_result # ty:ignore[invalid-yield]
return wrap
class StateValue(FastMCPBaseModel):
"""Wrapper for stored context state values."""
value: Any
class FastMCP(
AggregateProvider,
LifespanMixin,
MCPOperationsMixin,
TransportMixin,
Generic[LifespanResultT],
):
def __init__(
self,
name: str | None = None,
instructions: str | None = None,
*,
version: str | int | float | None = None,
website_url: str | None = None,
icons: list[mcp_types.Icon] | None = None,
auth: AuthProvider | None = None,
middleware: Sequence[Middleware] | None = None,
providers: Sequence[Provider] | None = None,
transforms: Sequence[Transform] | None = None,
lifespan: LifespanCallable | Lifespan | None = None,
tools: Sequence[Tool | Callable[..., Any]] | None = None,
on_duplicate: DuplicateBehavior | None = None,
mask_error_details: bool | None = None,
dereference_schemas: bool = True,
strict_input_validation: bool | None = None,
list_page_size: int | None = None,
cache_ttl: int | None = None,
cache_scope: Literal["public", "private"] | None = None,
tasks: bool | None = None,
session_state_store: AsyncKeyValue | None = None,
sampling_handler: SamplingHandler | None = None,
sampling_handler_behavior: Literal["always", "fallback"] | None = None,
client_log_level: mcp_types.LoggingLevel | None = None,
experimental_capabilities: dict[str, dict[str, Any]] | None = None,
**kwargs: Any,
):
_check_removed_kwargs(kwargs)
# Initialize Provider (sets up _transforms)
super().__init__()
self._on_duplicate: DuplicateBehaviorSetting = on_duplicate or "warn"
# Resolve server default for background task support
self._support_tasks_by_default: bool = tasks if tasks is not None else False
# Docket and Worker instances (set during lifespan for cross-task access)
self._docket = None
self._worker = None
self._additional_http_routes: list[BaseRoute] = []
# Session-scoped state store (shared across all requests)
self._state_storage: AsyncKeyValue = session_state_store or MemoryStore()
self._state_store: PydanticAdapter[StateValue] = PydanticAdapter[StateValue](
key_value=self._state_storage,
pydantic_model=StateValue,
default_collection="fastmcp_state",
)
# Create LocalProvider for local components
self._local_provider: LocalProvider = LocalProvider(
on_duplicate=self._on_duplicate
)
# Add providers using AggregateProvider's add_provider
# LocalProvider is always first (no namespace)
self.add_provider(self._local_provider)
for p in providers or []:
self.add_provider(p)
for t in transforms or []:
self.add_transform(t)
# Store mask_error_details for execution error handling
self._mask_error_details: bool = (
mask_error_details
if mask_error_details is not None
else fastmcp.settings.mask_error_details
)
# Store list_page_size for pagination of list operations
if list_page_size is not None and list_page_size <= 0:
raise ValueError("list_page_size must be a positive integer")
self._list_page_size: int | None = list_page_size
# Server-level SEP-2549 cache hints, applied uniformly to every
# SDK-cacheable result by the low-level server's runner (raises on
# invalid ttl/scope).
cache_hints = build_cache_hints(cache_ttl, cache_scope)
# Handle Lifespan instances (they're callable) or regular lifespan functions
if lifespan is not None:
self._lifespan: LifespanCallable[LifespanResultT] = cast(
LifespanCallable[LifespanResultT], lifespan
)
else:
self._lifespan = cast(LifespanCallable[LifespanResultT], default_lifespan)
self._lifespan_result: LifespanResultT | None = None
self._lifespan_result_set: bool = False
# Snapshot of SharedContext ContextVar values captured during the
# lifespan, re-applied per request by FastMCPServerMiddleware because
# the SDK v2 dispatcher runs handlers in the sender's context.
self._shared_context_snapshot: dict[Any, Any] | None = None
self._lifespan_ref_count: int = 0
self._lifespan_lock: asyncio.Lock = asyncio.Lock()
self._started: asyncio.Event = asyncio.Event()
# Generate random ID if no name provided
self._mcp_server: LowLevelServer[LifespanResultT] = LowLevelServer[
LifespanResultT
](
fastmcp=self,
name=name or self.generate_name(),
version=_coerce_version(version) or fastmcp.__version__,
instructions=instructions,
website_url=website_url,
icons=icons,
lifespan=_lifespan_proxy(fastmcp_server=self),
cache_hints=cache_hints,
)
self.auth: AuthProvider | None = auth
if tools:
for tool in tools:
if not isinstance(tool, Tool):
tool = Tool.from_function(tool)
self.add_tool(tool)
self.strict_input_validation: bool = (
strict_input_validation
if strict_input_validation is not None
else fastmcp.settings.strict_input_validation
)
self.client_log_level: mcp_types.LoggingLevel | None = (
client_log_level
if client_log_level is not None
else fastmcp.settings.client_log_level
)
# Per-session minimum log level requested by clients via logging/setLevel.
# Keyed by session id (a sentinel for stdio where session_id is None).
# v2 sessions are per-request so this state lives on the server, not the
# session object.
self._client_log_levels: dict[str, mcp_types.LoggingLevel] = {}
self.experimental_capabilities: dict[str, dict[str, Any]] = (
experimental_capabilities or {}
)
self.middleware: list[Middleware] = list(middleware or [])
if dereference_schemas:
from fastmcp.server.middleware.dereference import (
DereferenceRefsMiddleware,
)
self.middleware.append(DereferenceRefsMiddleware())
# Set up MCP protocol handlers
self._setup_handlers()
self.sampling_handler: SamplingHandler | None = sampling_handler
self.sampling_handler_behavior: Literal["always", "fallback"] = (
sampling_handler_behavior or "fallback"
)
def __repr__(self) -> str:
return f"{type(self).__name__}({self.name!r})"
@property
def name(self) -> str:
return self._mcp_server.name
@property
def instructions(self) -> str | None:
return self._mcp_server.instructions
@instructions.setter
def instructions(self, value: str | None) -> None:
self._mcp_server.instructions = value
@property
def version(self) -> str | None:
return self._mcp_server.version
@property
def website_url(self) -> str | None:
return self._mcp_server.website_url
@property
def icons(self) -> list[mcp_types.Icon]:
if self._mcp_server.icons is None:
return []
else:
return list(self._mcp_server.icons)
@property
def local_provider(self) -> LocalProvider:
"""The server's local provider, which stores directly-registered components.
Use this to remove components:
mcp.local_provider.remove_tool("my_tool")
mcp.local_provider.remove_resource("data://info")
mcp.local_provider.remove_prompt("my_prompt")
"""
return self._local_provider
async def _run_middleware(
self,
context: MiddlewareContext[Any],
call_next: CallNext[Any, Any],
) -> Any:
"""Builds and executes the middleware chain."""
chain = call_next
for mw in reversed(self.middleware):
next_chain: CallNext[Any, Any] = chain
async def wrapped(
context: MiddlewareContext[Any],
mw: Middleware = mw,
call_next: CallNext[Any, Any] = next_chain,
) -> Any:
return await mw(context, call_next)
chain = cast(CallNext[Any, Any], wrapped)
return await chain(context)
def add_middleware(self, middleware: Middleware) -> None:
self.middleware.append(middleware)
def add_provider(self, provider: Provider, *, namespace: str = "") -> None:
"""Add a provider for dynamic tools, resources, and prompts.
Providers are queried in registration order. The first provider to return
a non-None result wins. Static components (registered via decorators)
always take precedence over providers.
Args:
provider: A Provider instance that will provide components dynamically.
namespace: Optional namespace prefix. When set:
- Tools become "namespace_toolname"
- Resources become "protocol://namespace/path"
- Prompts become "namespace_promptname"
"""
super().add_provider(provider, namespace=namespace)
def _rewrite_prefab_uris(self, tools: list[Tool]) -> list[Tool]:
"""Replace placeholder Prefab URIs with per-tool hashed ones.
For each tool whose ``meta.ui.resourceUri`` is the placeholder,
reads the tool's stored hash from ``meta.fastmcp._tool_hash``
and rewrites the URI to the per-tool form. Also strips CSP from
tool meta (it belongs on the resource). Produces ``model_copy``
views — originals are untouched.
"""
from fastmcp.server.providers.prefab_synthesis import (
_is_prefab_tool,
rewrite_tool_meta_for_wire,
)
return [
rewrite_tool_meta_for_wire(t) if _is_prefab_tool(t) else t for t in tools
]
# -------------------------------------------------------------------------
# Provider interface overrides - inherited from AggregateProvider
# -------------------------------------------------------------------------
# _list_tools, _list_resources, _list_resource_templates, _list_prompts
# are inherited from AggregateProvider which handles aggregation and namespacing
async def get_tasks(self) -> Sequence[FastMCPComponent]:
"""Get task-eligible components with all transforms applied.
Overrides AggregateProvider.get_tasks() to apply server-level transforms
after aggregation. AggregateProvider handles provider-level namespacing.
"""
# Get tasks from AggregateProvider (handles aggregation and namespacing)
components = list(await super().get_tasks())
# Separate by component type for server-level transform application
tools = [c for c in components if isinstance(c, Tool)]
resources = [c for c in components if isinstance(c, Resource)]
templates = [c for c in components if isinstance(c, ResourceTemplate)]
prompts = [c for c in components if isinstance(c, Prompt)]
# Apply server-level transforms sequentially
for transform in self.transforms:
tools = await transform.list_tools(tools)
resources = await transform.list_resources(resources)
templates = await transform.list_resource_templates(templates)
prompts = await transform.list_prompts(prompts)
return [
*tools,
*resources,
*templates,
*prompts,
]
def add_transform(self, transform: Transform) -> None:
"""Add a server-level transform.
Server-level transforms are applied after all providers are aggregated.
They transform tools, resources, and prompts from ALL providers.
Args:
transform: The transform to add.
Example:
```python
from fastmcp.server.transforms import Namespace
server = FastMCP("Server")
server.add_transform(Namespace("api"))
# All tools from all providers become "api_toolname"
```
"""
self._transforms.append(transform)
async def list_tools(self, *, run_middleware: bool = True) -> Sequence[Tool]:
"""List all enabled tools from providers.
Overrides Provider.list_tools() to add visibility filtering, auth filtering,
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message=mcp_types.ListToolsRequest(method="tools/list"),
source="client",
type="request",
method="tools/list",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.list_tools(run_middleware=False),
)
# Core logic: list tools
with server_span("tools/list", "tools/list", self.name, "tool", ""):
# Get all tools, apply session transforms, then filter enabled
# and model-visible (app-only tools are hidden from the model).
tools = list(await super().list_tools())
tools = await apply_session_transforms(tools)
tools = [t for t in tools if is_enabled(t) and not _is_backend_tool(t)]
# Rewrite per-tool Prefab renderer URIs based on the tool's
# mount-point address. The walk pairs each tool with the
# provider that yielded it, computes the hashed URI, and
# produces a model_copy with the URI in place. Original
# Tool objects are not mutated.
tools = self._rewrite_prefab_uris(tools)
skip_auth, token = _get_auth_context()
authorized: list[Tool] = []
for tool in tools:
if not skip_auth and tool.auth is not None:
ctx = AuthContext(token=token, component=tool)
try:
if not await run_auth_checks(tool.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(tool)
return authorized
async def _get_tool(
self, name: str, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name via aggregation from providers.
Extends AggregateProvider._get_tool() with component-level auth checks.
Args:
name: The tool name.
version: Version filter (None returns highest version).
Returns:
The tool if found and authorized, None if not found or unauthorized.
"""
# Get tool from AggregateProvider (handles aggregation and namespacing)
tool = await super()._get_tool(name, version)
if tool is None:
return None
# Component auth - return None if unauthorized (consistent with list filtering)
skip_auth, token = _get_auth_context()
if not skip_auth and tool.auth is not None:
ctx = AuthContext(token=token, component=tool)
try:
if not await run_auth_checks(tool.auth, ctx):
return None
except AuthorizationError:
return None
return tool
async def get_tool(
self, name: str, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name, filtering disabled tools.
Overrides Provider.get_tool() to add visibility filtering after all
transforms (including session-level) have been applied. This ensures
session transforms can override provider-level disables.
When the highest version is disabled and no explicit version was
requested, falls back to the next-highest enabled version.
Args:
name: The tool name.
version: Version filter (None returns highest version).
Returns:
The tool if found and enabled, None otherwise.
"""
tool = await super().get_tool(name, version)
if tool is None:
return None
# Apply session transforms to single item
tools = await apply_session_transforms([tool])
if tools and is_enabled(tools[0]) and not _is_backend_tool(tools[0]):
return tools[0]
# The highest version is disabled (or app-only). If an explicit version
# was requested, respect that. Otherwise fall back to the next-highest
# enabled, model-visible version.
if version is not None:
return None
all_tools = [t for t in await super().list_tools() if t.name == name]
all_tools = list(await apply_session_transforms(all_tools))
enabled = [t for t in all_tools if is_enabled(t) and not _is_backend_tool(t)]
skip_auth, token = _get_auth_context()
authorized: list[Tool] = []
for t in enabled:
if not skip_auth and t.auth is not None:
ctx = AuthContext(token=token, component=t)
try:
if not await run_auth_checks(t.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(t)
if not authorized:
return None
return max(authorized, key=version_sort_key)
async def list_resources(
self, *, run_middleware: bool = True
) -> Sequence[Resource]:
"""List all enabled resources from providers.
Overrides Provider.list_resources() to add visibility filtering, auth filtering,
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message={},
source="client",
type="request",
method="resources/list",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.list_resources(run_middleware=False),
)
# Core logic: list resources
with server_span(
"resources/list", "resources/list", self.name, "resource", ""
):
# Get all resources, apply session transforms, then filter enabled
resources = list(await super().list_resources())
resources = await apply_session_transforms(resources)
resources = [r for r in resources if is_enabled(r)]
# Append synthetic Prefab renderer resources — one per
# prefab tool, hashed by mount address. These don't live on
# any provider's storage; they're computed on demand.
from fastmcp.server.providers.prefab_synthesis import (
synthesize_prefab_resources,
)
resources.extend(await synthesize_prefab_resources(self))
skip_auth, token = _get_auth_context()
authorized: list[Resource] = []
for resource in resources:
if not skip_auth and resource.auth is not None:
ctx = AuthContext(token=token, component=resource)
try:
if not await run_auth_checks(resource.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(resource)
return authorized
async def _get_resource(
self, uri: str, version: VersionSpec | None = None
) -> Resource | None:
"""Get a resource by URI via aggregation from providers.
Extends AggregateProvider._get_resource() with component-level auth checks.
Args:
uri: The resource URI.
version: Version filter (None returns highest version).
Returns:
The resource if found and authorized, None if not found or unauthorized.
"""
# Get resource from AggregateProvider (handles aggregation and namespacing)
resource = await super()._get_resource(uri, version)
if resource is None:
return None
# Component auth - return None if unauthorized (consistent with list filtering)
skip_auth, token = _get_auth_context()
if not skip_auth and resource.auth is not None:
ctx = AuthContext(token=token, component=resource)
try:
if not await run_auth_checks(resource.auth, ctx):
return None
except AuthorizationError:
return None
return resource
async def get_resource(
self, uri: str, version: VersionSpec | None = None
) -> Resource | None:
"""Get a resource by URI, filtering disabled resources.
Overrides Provider.get_resource() to add visibility filtering after all
transforms (including session-level) have been applied.
When the highest version is disabled and no explicit version was
requested, falls back to the next-highest enabled version.
Args:
uri: The resource URI.
version: Version filter (None returns highest version).
Returns:
The resource if found and enabled, None otherwise.
"""
resource = await super().get_resource(uri, version)
if resource is None:
return None
# Apply session transforms to single item
resources = await apply_session_transforms([resource])
if resources and is_enabled(resources[0]):
return resources[0]
if version is not None:
return None
all_resources = [r for r in await super().list_resources() if str(r.uri) == uri]
all_resources = list(await apply_session_transforms(all_resources))
enabled = [r for r in all_resources if is_enabled(r)]
skip_auth, token = _get_auth_context()
authorized: list[Resource] = []
for r in enabled:
if not skip_auth and r.auth is not None:
ctx = AuthContext(token=token, component=r)
try:
if not await run_auth_checks(r.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(r)
if not authorized:
return None
return max(authorized, key=version_sort_key)
async def list_resource_templates(
self, *, run_middleware: bool = True
) -> Sequence[ResourceTemplate]:
"""List all enabled resource templates from providers.
Overrides Provider.list_resource_templates() to add visibility filtering,
auth filtering, and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message={},
source="client",
type="request",
method="resources/templates/list",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.list_resource_templates(
run_middleware=False
),
)
# Core logic: list resource templates
with server_span(
"resources/templates/list",
"resources/templates/list",
self.name,
"resource_template",
"",
):
# Get all templates, apply session transforms, then filter enabled
templates = list(await super().list_resource_templates())
templates = await apply_session_transforms(templates)
templates = [t for t in templates if is_enabled(t)]
skip_auth, token = _get_auth_context()
authorized: list[ResourceTemplate] = []
for template in templates:
if not skip_auth and template.auth is not None:
ctx = AuthContext(token=token, component=template)
try:
if not await run_auth_checks(template.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(template)
return authorized
async def _get_resource_template(
self, uri: str, version: VersionSpec | None = None
) -> ResourceTemplate | None:
"""Get a resource template by URI via aggregation from providers.
Extends AggregateProvider._get_resource_template() with component-level auth checks.
Args:
uri: The template URI to match.
version: Version filter (None returns highest version).
Returns:
The template if found and authorized, None if not found or unauthorized.
"""
# Get template from AggregateProvider (handles aggregation and namespacing)
template = await super()._get_resource_template(uri, version)
if template is None:
return None
# Component auth - return None if unauthorized (consistent with list filtering)
skip_auth, token = _get_auth_context()
if not skip_auth and template.auth is not None:
ctx = AuthContext(token=token, component=template)
try:
if not await run_auth_checks(template.auth, ctx):
return None
except AuthorizationError:
return None
return template
async def get_resource_template(
self, uri: str, version: VersionSpec | None = None
) -> ResourceTemplate | None:
"""Get a resource template by URI, filtering disabled templates.
Overrides Provider.get_resource_template() to add visibility filtering after
all transforms (including session-level) have been applied.
When the highest version is disabled and no explicit version was
requested, falls back to the next-highest enabled version.
Args:
uri: The template URI.
version: Version filter (None returns highest version).
Returns:
The template if found and enabled, None otherwise.
"""
template = await super().get_resource_template(uri, version)
if template is None:
return None
# Apply session transforms to single item
templates = await apply_session_transforms([template])
if templates and is_enabled(templates[0]):
return templates[0]
if version is not None:
return None
all_templates = [
t
for t in await super().list_resource_templates()
if t.matches(uri) is not None
]
all_templates = list(await apply_session_transforms(all_templates))
enabled = [t for t in all_templates if is_enabled(t)]
skip_auth, token = _get_auth_context()
authorized: list[ResourceTemplate] = []
for t in enabled:
if not skip_auth and t.auth is not None:
ctx = AuthContext(token=token, component=t)
try:
if not await run_auth_checks(t.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(t)
if not authorized:
return None
return max(authorized, key=version_sort_key)
async def list_prompts(self, *, run_middleware: bool = True) -> Sequence[Prompt]:
"""List all enabled prompts from providers.
Overrides Provider.list_prompts() to add visibility filtering, auth filtering,
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message={},
source="client",
type="request",
method="prompts/list",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.list_prompts(run_middleware=False),
)
# Core logic: list prompts
with server_span("prompts/list", "prompts/list", self.name, "prompt", ""):
# Get all prompts, apply session transforms, then filter enabled
prompts = list(await super().list_prompts())
prompts = await apply_session_transforms(prompts)
prompts = [p for p in prompts if is_enabled(p)]
skip_auth, token = _get_auth_context()
authorized: list[Prompt] = []
for prompt in prompts:
if not skip_auth and prompt.auth is not None:
ctx = AuthContext(token=token, component=prompt)
try:
if not await run_auth_checks(prompt.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(prompt)
return authorized
async def _get_prompt(
self, name: str, version: VersionSpec | None = None
) -> Prompt | None:
"""Get a prompt by name via aggregation from providers.
Extends AggregateProvider._get_prompt() with component-level auth checks.
Args:
name: The prompt name.
version: Version filter (None returns highest version).
Returns:
The prompt if found and authorized, None if not found or unauthorized.
"""
# Get prompt from AggregateProvider (handles aggregation and namespacing)
prompt = await super()._get_prompt(name, version)
if prompt is None:
return None
# Component auth - return None if unauthorized (consistent with list filtering)
skip_auth, token = _get_auth_context()
if not skip_auth and prompt.auth is not None:
ctx = AuthContext(token=token, component=prompt)
try:
if not await run_auth_checks(prompt.auth, ctx):
return None
except AuthorizationError:
return None
return prompt
async def get_prompt(
self, name: str, version: VersionSpec | None = None
) -> Prompt | None:
"""Get a prompt by name, filtering disabled prompts.
Overrides Provider.get_prompt() to add visibility filtering after all
transforms (including session-level) have been applied.
When the highest version is disabled and no explicit version was
requested, falls back to the next-highest enabled version.
Args:
name: The prompt name.
version: Version filter (None returns highest version).
Returns:
The prompt if found and enabled, None otherwise.
"""
prompt = await super().get_prompt(name, version)
if prompt is None:
return None
# Apply session transforms to single item
prompts = await apply_session_transforms([prompt])
if prompts and is_enabled(prompts[0]):
return prompts[0]
if version is not None:
return None
all_prompts = [p for p in await super().list_prompts() if p.name == name]
all_prompts = list(await apply_session_transforms(all_prompts))
enabled = [p for p in all_prompts if is_enabled(p)]
skip_auth, token = _get_auth_context()
authorized: list[Prompt] = []
for p in enabled:
if not skip_auth and p.auth is not None:
ctx = AuthContext(token=token, component=p)
try:
if not await run_auth_checks(p.auth, ctx):
continue
except AuthorizationError:
continue
authorized.append(p)
if not authorized:
return None
return max(authorized, key=version_sort_key)
@overload
async def call_tool(
self,
name: str,
arguments: dict[str, Any] | None = None,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: None = None,
) -> ToolResult: ...
@overload
async def call_tool(
self,
name: str,
arguments: dict[str, Any] | None = None,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: TaskMeta,
) -> mcp_types.CreateTaskResult: ...
async def call_tool(
self,
name: str,
arguments: dict[str, Any] | None = None,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: TaskMeta | None = None,
) -> ToolResult | mcp_types.CreateTaskResult:
"""Call a tool by name.
This is the public API for executing tools. By default, middleware is applied.
Args:
name: The tool name
arguments: Tool arguments (optional)
version: Specific version to call. If None, calls highest version.
run_middleware: If True (default), apply the middleware chain.
Set to False when called from middleware to avoid re-applying.
task_meta: If provided, execute as a background task and return
CreateTaskResult. If None (default), execute synchronously and
return ToolResult.
Returns:
ToolResult when task_meta is None.
CreateTaskResult when task_meta is provided.
Raises:
NotFoundError: If tool not found or disabled
ToolError: If tool execution fails
ValidationError: If arguments fail validation
"""
# Note: fn_key enrichment happens here after finding the tool.
# For mounted servers, the parent's provider sets fn_key to the
# namespaced key before delegating, ensuring correct Docket routing.
from fastmcp.server.providers.addressing import (
parse_hashed_backend_name,
)
# Two routing paths:
# 1. Hashed-name path — backend tools that opted into
# app-callable visibility. Recognized by their
# `<hash>_<local_name>` format and resolved via the
# reverse-hash map. Address is known eagerly.
# 2. Display-name path — everything else. Goes through normal
# `get_tool` aggregation/transforms. Address is determined
# after resolution by walking the registry.
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext[CallToolRequestParams](
message=mcp_types.CallToolRequestParams(
name=name,
arguments=arguments or {},
# `_meta` carries the app-level `fastmcp` version key, which the
# reserved-key RequestParamsMeta TypedDict can't express statically.
_meta=_version_request_meta(version), # type: ignore[unknown-argument] # ty: ignore[invalid-argument-type]
),
source="client",
type="request",
method="tools/call",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.call_tool(
context.message.name,
context.message.arguments or {},
version=version,
run_middleware=False,
task_meta=task_meta,
),
)
# Core logic: find and execute tool
with server_span(
f"tools/call {name}",
"tools/call",
self.name,
"tool",
name,
tool_name=name,
) as span:
# Try normal display-name resolution first.
tool: Tool | None = await self.get_tool(name, version=version)
# If that fails, try hashed-name dispatch. This walks
# the provider tree recursively (same pattern as the old
# get_app_tool) looking for a tool whose stored hash
# matches the parsed prefix.
if tool is None:
hashed = parse_hashed_backend_name(name)
if hashed is not None:
digest, local_name = hashed
tool = await self.get_tool_by_hash(digest, local_name)
if tool is not None:
# Auth still applies on the bypass path.
skip_auth, token = _get_auth_context()
if not skip_auth and tool.auth is not None:
try:
auth_ctx = AuthContext(token=token, component=tool)
if not await run_auth_checks(tool.auth, auth_ctx):
raise NotFoundError(f"Unknown tool: {name!r}")
except AuthorizationError:
raise NotFoundError(
f"Unknown tool: {name!r}"
) from None
if tool is None:
raise NotFoundError(f"Unknown tool: {name!r}")
span.set_attributes(tool.get_span_attributes())
if task_meta is not None and task_meta.fn_key is None:
task_meta = replace(task_meta, fn_key=tool.key)
try:
return await tool._run(arguments or {}, task_meta=task_meta)
except ValidationError as e:
# Argument-validation failure (a bad call). FunctionTool
# converts pydantic's call-validation error into fastmcp's
# ValidationError (see #4128) so it can be filtered as a
# client error. Log the underlying detail without a URL or
# traceback, matching the previous pydantic-error logging.
cause = e.__cause__
detail = (
cause.errors(include_url=False)
if isinstance(cause, PydanticValidationError)
else str(e)
)
logger.warning("Invalid arguments for tool %r: %s", name, detail)
raise
except FastMCPError as e:
logger.log(
e.log_level, f"Error calling tool {name!r}", exc_info=False
)
raise
except PydanticValidationError as e:
# A pydantic error that is NOT an argument-validation failure
# (e.g. raised by a non-FunctionTool's own validation). Kept
# for backward compatibility.
logger.warning(
"Invalid arguments for tool %r: %s",
name,
e.errors(include_url=False),
)
raise
except Exception as e:
logger.exception(f"Error calling tool {name!r}")
# Handle actionable errors that should reach the LLM
# even when masking is enabled
if isinstance(e, httpx.HTTPStatusError):
if e.response.status_code == 429:
raise ToolError(
"Rate limited by upstream API, please retry later"
) from e
if isinstance(e, httpx.TimeoutException):
raise ToolError(
"Upstream request timed out, please retry"
) from e
# Standard masking logic
if self._mask_error_details:
raise ToolError(f"Error calling tool {name!r}") from e
raise ToolError(f"Error calling tool {name!r}: {e}") from e
@overload
async def read_resource(
self,
uri: str,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: None = None,
) -> ResourceResult: ...
@overload
async def read_resource(
self,
uri: str,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: TaskMeta,
) -> mcp_types.CreateTaskResult: ...
async def read_resource(
self,
uri: str,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: TaskMeta | None = None,
) -> ResourceResult | mcp_types.CreateTaskResult:
"""Read a resource by URI.
This is the public API for reading resources. By default, middleware is applied.
Checks concrete resources first, then templates.
Args:
uri: The resource URI
version: Specific version to read. If None, reads highest version.
run_middleware: If True (default), apply the middleware chain.
Set to False when called from middleware to avoid re-applying.
task_meta: If provided, execute as a background task and return
CreateTaskResult. If None (default), execute synchronously and
return ResourceResult.
Returns:
ResourceResult when task_meta is None.
CreateTaskResult when task_meta is provided.
Raises:
NotFoundError: If resource not found or disabled
ResourceError: If resource read fails
"""
# Note: fn_key enrichment happens here after finding the resource/template.
# Resources and templates use different key formats:
# - Resources use resource.key (derived from the concrete URI)
# - Templates use template.key (the template pattern)
# For mounted servers, the parent's provider sets fn_key to the
# namespaced key before delegating, ensuring correct Docket routing.
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message=mcp_types.ReadResourceRequestParams(
uri=str(uri),
# `_meta` carries the app-level `fastmcp` version key, which the
# reserved-key RequestParamsMeta TypedDict can't express statically.
_meta=_version_request_meta(version), # type: ignore[unknown-argument] # ty: ignore[invalid-argument-type]
),
source="client",
type="request",
method="resources/read",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.read_resource(
str(context.message.uri),
version=version,
run_middleware=False,
task_meta=task_meta,
),
)
# Core logic: find and read resource (providers queried in parallel)
with server_span(
"resources/read",
"resources/read",
self.name,
"resource",
uri,
resource_uri=uri,
) as span:
# Intercept synthetic Prefab renderer URIs before normal
# resolution. The resource isn't stored anywhere — we
# build it on demand from the matching tool's CSP.
from fastmcp.server.providers.prefab_synthesis import (
synthesize_prefab_resource_by_uri,
)
synthesized = await synthesize_prefab_resource_by_uri(self, uri)
if synthesized is not None:
span.set_attributes(synthesized.get_span_attributes())
return await synthesized._read(task_meta=task_meta)
# Try concrete resources first (transforms + auth via _get_resource)
resource = await self.get_resource(uri, version=version)
if resource is not None:
span.set_attributes(resource.get_span_attributes())
if task_meta is not None and task_meta.fn_key is None:
task_meta = replace(task_meta, fn_key=resource.key)
try:
return await resource._read(task_meta=task_meta)
except FastMCPError as e:
logger.log(
e.log_level,
f"Error reading resource {uri!r}",
exc_info=True,
)
raise
except MCPError:
logger.exception(f"Error reading resource {uri!r}")
raise
except Exception as e:
logger.exception(f"Error reading resource {uri!r}")
# Handle actionable errors that should reach the LLM
if isinstance(e, httpx.HTTPStatusError):
if e.response.status_code == 429:
raise ResourceError(
"Rate limited by upstream API, please retry later"
) from e
if isinstance(e, httpx.TimeoutException):
raise ResourceError(
"Upstream request timed out, please retry"
) from e
# Standard masking logic
if self._mask_error_details:
raise ResourceError(
f"Error reading resource {uri!r}"
) from e
raise ResourceError(
f"Error reading resource {uri!r}: {e}"
) from e
# Try templates (transforms + auth via get_resource_template)
template = await self.get_resource_template(uri, version=version)
if template is None:
if version is None:
raise NotFoundError(f"Unknown resource: {uri!r}")
raise NotFoundError(
f"Unknown resource: {uri!r} version {version!r}"
)
span.set_attributes(template.get_span_attributes())
params = template.matches(uri)
assert params is not None
if task_meta is not None and task_meta.fn_key is None:
task_meta = replace(task_meta, fn_key=template.key)
try:
return await template._read(uri, params, task_meta=task_meta)
except FastMCPError as e:
logger.log(
e.log_level, f"Error reading resource {uri!r}", exc_info=True
)
raise
except MCPError:
logger.exception(f"Error reading resource {uri!r}")
raise
except Exception as e:
logger.exception(f"Error reading resource {uri!r}")
# Handle actionable errors that should reach the LLM
if isinstance(e, httpx.HTTPStatusError):
if e.response.status_code == 429:
raise ResourceError(
"Rate limited by upstream API, please retry later"
) from e
if isinstance(e, httpx.TimeoutException):
raise ResourceError(
"Upstream request timed out, please retry"
) from e
# Standard masking logic
if self._mask_error_details:
raise ResourceError(f"Error reading resource {uri!r}") from e
raise ResourceError(f"Error reading resource {uri!r}: {e}") from e
@overload
async def render_prompt(
self,
name: str,
arguments: dict[str, Any] | None = None,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: None = None,
) -> PromptResult: ...
@overload
async def render_prompt(
self,
name: str,
arguments: dict[str, Any] | None = None,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: TaskMeta,
) -> mcp_types.CreateTaskResult: ...
async def render_prompt(
self,
name: str,
arguments: dict[str, Any] | None = None,
*,
version: VersionSpec | None = None,
run_middleware: bool = True,
task_meta: TaskMeta | None = None,
) -> PromptResult | mcp_types.CreateTaskResult:
"""Render a prompt by name.
This is the public API for rendering prompts. By default, middleware is applied.
Use get_prompt() to retrieve the prompt definition without rendering.
Args:
name: The prompt name
arguments: Prompt arguments (optional)
version: Specific version to render. If None, renders highest version.
run_middleware: If True (default), apply the middleware chain.
Set to False when called from middleware to avoid re-applying.
task_meta: If provided, execute as a background task and return
CreateTaskResult. If None (default), execute synchronously and
return PromptResult.
Returns:
PromptResult when task_meta is None.
CreateTaskResult when task_meta is provided.
Raises:
NotFoundError: If prompt not found or disabled
PromptError: If prompt rendering fails
"""
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message=mcp_types.GetPromptRequestParams(
name=name,
arguments=arguments,
# `_meta` carries the app-level `fastmcp` version key, which the
# reserved-key RequestParamsMeta TypedDict can't express statically.
_meta=_version_request_meta(version), # type: ignore[unknown-argument] # ty: ignore[invalid-argument-type]
),
source="client",
type="request",
method="prompts/get",
fastmcp_context=ctx,
)
return await self._run_middleware(
context=mw_context,
call_next=lambda context: self.render_prompt(
context.message.name,
context.message.arguments,
version=version,
run_middleware=False,
task_meta=task_meta,
),
)
# Core logic: find and render prompt (providers queried in parallel)
# Use get_prompt to apply transforms and filter disabled
with server_span(
f"prompts/get {name}",
"prompts/get",
self.name,
"prompt",
name,
prompt_name=name,
) as span:
prompt = await self.get_prompt(name, version=version)
if prompt is None:
raise NotFoundError(f"Unknown prompt: {name!r}")
span.set_attributes(prompt.get_span_attributes())
if task_meta is not None and task_meta.fn_key is None:
task_meta = replace(task_meta, fn_key=prompt.key)
try:
return await prompt._render(arguments, task_meta=task_meta)
except FastMCPError as e:
logger.log(
e.log_level, f"Error rendering prompt {name!r}", exc_info=True
)
raise
except MCPError:
logger.exception(f"Error rendering prompt {name!r}")
raise
except Exception as e:
logger.exception(f"Error rendering prompt {name!r}")
if self._mask_error_details:
raise PromptError(f"Error rendering prompt {name!r}") from e
raise PromptError(f"Error rendering prompt {name!r}: {e}") from e
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 or @tool-decorated function to register
Returns:
The tool instance that was added to the server.
"""
return self._local_provider.add_tool(tool)
@overload
def tool(
self,
name_or_fn: F,
*,
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,
meta: dict[str, Any] | None = None,
app: AppConfig | dict[str, Any] | bool | None = None,
task: bool | TaskConfig | None = None,
timeout: float | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
run_in_thread: bool = True,
) -> F: ...
@overload
def tool(
self,
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,
meta: dict[str, Any] | None = None,
app: AppConfig | dict[str, Any] | bool | None = None,
task: bool | TaskConfig | None = None,
timeout: float | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
run_in_thread: bool = True,
) -> Callable[[F], F]: ...
def tool(
self,
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,
meta: dict[str, Any] | None = None,
app: AppConfig | dict[str, Any] | bool | None = None,
task: bool | TaskConfig | None = None,
timeout: float | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
run_in_thread: bool = True,
) -> (
Callable[[AnyFunction], FunctionTool]
| FunctionTool
| partial[Callable[[AnyFunction], FunctionTool] | FunctionTool]
):
"""Decorator to register a tool.
Tools can optionally request a Context object by adding a parameter with the
Context type annotation. The context provides access to MCP capabilities like
logging, progress reporting, and resource access.
This decorator supports multiple calling patterns:
- @server.tool (without parentheses)
- @server.tool (with empty parentheses)
- @server.tool("custom_name") (with name as first argument)
- @server.tool(name="custom_name") (with name as keyword argument)
- server.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)
description: Optional description of what the tool does
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
Examples:
Register a tool with a custom name:
```python
@server.tool
def my_tool(x: int) -> str:
return str(x)
# Register a tool with a custom name
@server.tool
def my_tool(x: int) -> str:
return str(x)
@server.tool("custom_name")
def my_tool(x: int) -> str:
return str(x)
@server.tool(name="custom_name")
def my_tool(x: int) -> str:
return str(x)
# Direct function call
server.tool(my_function, name="custom_name")
```
"""
# Merge app config into meta["ui"] (wire format) before passing to provider
if app is not None and app is not False:
meta = dict(meta) if meta else {}
if app is True:
meta["ui"] = True
else:
meta["ui"] = app_config_to_meta_dict(app)
# Delegate to LocalProvider with server-level defaults
result = self._local_provider.tool(
name_or_fn,
name=name,
version=version,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
task=task if task is not None else self._support_tasks_by_default,
timeout=timeout,
auth=auth,
run_in_thread=run_in_thread,
)
return result
def add_resource(
self, resource: Resource | Callable[..., Any]
) -> Resource | ResourceTemplate:
"""Add a resource to the server.
Args:
resource: A Resource instance or @resource-decorated function to add
Returns:
The resource instance that was added to the server.
"""
return self._local_provider.add_resource(resource)
def add_template(self, template: ResourceTemplate) -> ResourceTemplate:
"""Add a resource template to the server.
Args:
template: A ResourceTemplate instance to add
Returns:
The template instance that was added to the server.
"""
return self._local_provider.add_template(template)
def resource(
self,
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,
annotations: Annotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
app: AppConfig | dict[str, Any] | bool | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
) -> Callable[[F], F]:
"""Decorator to register a function as a resource.
The function will be called when the resource is read to generate its content.
The function can return:
- str for text content
- bytes for binary content
- other types will be converted to JSON
Resources can optionally request a Context object by adding a parameter with the
Context type annotation. The context provides access to MCP capabilities like
logging, progress reporting, and session information.
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
description: Optional description of 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
Examples:
Register a resource with a custom name:
```python
@server.resource("resource://my-resource")
def get_data() -> str:
return "Hello, world!"
@server.resource("resource://my-resource")
async get_data() -> str:
data = await fetch_data()
return f"Hello, world! {data}"
@server.resource("resource://{city}/weather")
def get_weather(city: str) -> str:
return f"Weather for {city}"
@server.resource("resource://{city}/weather")
async def get_weather_with_context(city: str, ctx: Context) -> str:
await ctx.info(f"Fetching weather for {city}")
return f"Weather for {city}"
@server.resource("resource://{city}/weather")
async def get_weather(city: str) -> str:
data = await fetch_weather(city)
return f"Weather for {city}: {data}"
```
"""
# Catch incorrect decorator usage early (before any processing)
if not isinstance(uri, str):
raise TypeError(
"The @resource decorator was used incorrectly. "
"It requires a URI as the first argument. "
"Use @resource('uri') instead of @resource"
)
# Apply default MIME type for ui:// scheme resources
mime_type = resolve_ui_mime_type(uri, mime_type)
# Validate app config for resources — resource_uri and visibility
# don't apply since the resource itself is the UI
if isinstance(app, AppConfig):
if app.resource_uri is not None:
raise ValueError(
"resource_uri cannot be set on resources — "
"the resource itself is the UI. "
"Use resource_uri on tools to point to a UI resource."
)
if app.visibility is not None:
raise ValueError(
"visibility cannot be set on resources — it only applies to tools."
)
# Merge app config into meta["ui"] (wire format) before passing to provider
if app is not None and app is not False:
meta = dict(meta) if meta else {}
if app is True:
meta["ui"] = True
else:
meta["ui"] = app_config_to_meta_dict(app)
# Delegate to LocalProvider with server-level defaults
inner_decorator = self._local_provider.resource(
uri,
name=name,
version=version,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
task=task if task is not None else self._support_tasks_by_default,
auth=auth,
)
return inner_decorator
def add_prompt(self, prompt: Prompt | Callable[..., Any]) -> Prompt:
"""Add a prompt to the server.
Args:
prompt: A Prompt instance or @prompt-decorated function to add
Returns:
The prompt instance that was added to the server.
"""
return self._local_provider.add_prompt(prompt)
@overload
def prompt(
self,
name_or_fn: F,
*,
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,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
) -> F: ...
@overload
def prompt(
self,
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,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
) -> Callable[[F], F]: ...
def prompt(
self,
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,
meta: dict[str, Any] | None = None,
task: bool | TaskConfig | None = None,
auth: AuthCheck | list[AuthCheck] | None = None,
) -> (
Callable[[AnyFunction], FunctionPrompt]
| FunctionPrompt
| partial[Callable[[AnyFunction], FunctionPrompt] | FunctionPrompt]
):
"""Decorator to register a prompt.
Prompts can optionally request a Context object by adding a parameter with the
Context type annotation. The context provides access to MCP capabilities like
logging, progress reporting, and session information.
This decorator supports multiple calling patterns:
- @server.prompt (without parentheses)
- @server.prompt() (with empty parentheses)
- @server.prompt("custom_name") (with name as first argument)
- @server.prompt(name="custom_name") (with name as keyword argument)
- server.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)
description: Optional description of what the prompt does
tags: Optional set of tags for categorizing the prompt
meta: Optional meta information about the prompt
Examples:
```python
@server.prompt
def analyze_table(table_name: str) -> list[Message]:
schema = read_table_schema(table_name)
return [
{
"role": "user",
"content": f"Analyze this schema:\n{schema}"
}
]
@server.prompt()
async def analyze_with_context(table_name: str, ctx: Context) -> list[Message]:
await ctx.info(f"Analyzing table {table_name}")
schema = read_table_schema(table_name)
return [
{
"role": "user",
"content": f"Analyze this schema:\n{schema}"
}
]
@server.prompt("custom_name")
async def analyze_file(path: str) -> list[Message]:
content = await read_file(path)
return [
{
"role": "user",
"content": {
"type": "resource",
"resource": {
"uri": f"file://{path}",
"text": content
}
}
}
]
@server.prompt(name="custom_name")
def another_prompt(data: str) -> list[Message]:
return [{"role": "user", "content": data}]
# Direct function call
server.prompt(my_function, name="custom_name")
```
"""
# Delegate to LocalProvider with server-level defaults
return self._local_provider.prompt(
name_or_fn,
name=name,
version=version,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
task=task if task is not None else self._support_tasks_by_default,
auth=auth,
)
def mount(
self,
server: FastMCP[LifespanResultT],
namespace: str | None = None,
tool_names: dict[str, str] | None = None,
) -> None:
"""Mount another FastMCP server on this server with an optional namespace.
Mounting establishes a dynamic connection between servers. When a client
interacts with a mounted server's objects through the parent server, requests
are forwarded to the mounted server in real-time. This means changes to the
mounted server are immediately reflected when accessed through the parent.
When a server is mounted with a namespace:
- Tools from the mounted server are accessible with namespaced names.
Example: If server has a tool named "get_weather", it will be available as "namespace_get_weather".
- Resources are accessible with namespaced URIs.
Example: If server has a resource with URI "weather://forecast", it will be available as
"weather://namespace/forecast".
- Templates are accessible with namespaced URI templates.
Example: If server has a template with URI "weather://location/{id}", it will be available
as "weather://namespace/location/{id}".
- Prompts are accessible with namespaced names.
Example: If server has a prompt named "weather_prompt", it will be available as
"namespace_weather_prompt".
When a server is mounted without a namespace (namespace=None), its tools, resources, templates,
and prompts are accessible with their original names. Multiple servers can be mounted
without namespaces, and they will be tried in order until a match is found.
The mounted server's lifespan is executed when the parent server starts, and its
middleware chain is invoked for all operations (tool calls, resource reads, prompts).
Args:
server: The FastMCP server to mount.
namespace: Optional namespace to use for the mounted server's objects. If None,
the server's objects are accessible with their original names.
tool_names: Optional mapping of original tool names to custom names. Use this
to override namespaced names. Keys are the original tool names from the
mounted server.
"""
from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
if server is self:
raise ValueError("Cannot mount a server onto itself")
# Warn if parent masks errors but child doesn't (or vice versa)
if self._mask_error_details and not server._mask_error_details:
logger.warning(
f"Parent server {self.name!r} has mask_error_details=True but "
f"mounted server {server.name!r} does not. Error details from "
f"{server.name!r} may leak through to clients. Set "
f"mask_error_details=True on the child server to prevent this."
)
# Create provider and add it with namespace
provider: Provider = FastMCPProvider(server)
# Apply tool renames first (scoped to this provider), then namespace
# So foo → bar with namespace="baz" becomes baz_bar
if tool_names:
transforms = {
old_name: ToolTransformConfig(name=new_name)
for old_name, new_name in tool_names.items()
}
provider = provider.wrap_transform(ToolTransform(transforms))
# Use add_provider with namespace (applies namespace in AggregateProvider)
self.add_provider(provider, namespace=namespace or "")
@classmethod
def from_openapi(
cls,
openapi_spec: dict[str, Any],
client: httpx.AsyncClient | None = None,
name: str = "OpenAPI Server",
route_maps: list[RouteMap] | None = None,
route_map_fn: OpenAPIRouteMapFn | None = None,
mcp_component_fn: OpenAPIComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
tags: set[str] | None = None,
validate_output: bool = True,
**settings: Any,
) -> Self:
"""
Create a FastMCP server from an OpenAPI specification.
Args:
openapi_spec: OpenAPI schema as a dictionary
client: Optional httpx AsyncClient for making HTTP requests.
If not provided, a default client is created using the first
server URL from the OpenAPI spec with a 30-second timeout.
name: Name for the MCP server
route_maps: Optional list of RouteMap objects defining route mappings
route_map_fn: Optional callable for advanced route type mapping
mcp_component_fn: Optional callable for component customization
mcp_names: Optional dictionary mapping operationId to component names
tags: Optional set of tags to add to all components
validate_output: If True (default), tools use the output schema
extracted from the OpenAPI spec for response validation. If
False, a permissive schema is used instead, allowing any
response structure while still returning structured JSON.
**settings: Additional settings passed to FastMCP
Returns:
A FastMCP server with an OpenAPIProvider attached.
"""
from .providers.openapi import OpenAPIProvider
provider: Provider = OpenAPIProvider(
openapi_spec=openapi_spec,
client=client,
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
validate_output=validate_output,
)
return cls(name=name, providers=[provider], **settings)
@classmethod
def from_fastapi(
cls,
app: Any,
name: str | None = None,
route_maps: list[RouteMap] | None = None,
route_map_fn: OpenAPIRouteMapFn | None = None,
mcp_component_fn: OpenAPIComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
httpx_client_kwargs: dict[str, Any] | None = None,
tags: set[str] | None = None,
**settings: Any,
) -> Self:
"""
Create a FastMCP server from a FastAPI application.
Args:
app: FastAPI application instance
name: Name for the MCP server (defaults to app.title)
route_maps: Optional list of RouteMap objects defining route mappings
route_map_fn: Optional callable for advanced route type mapping
mcp_component_fn: Optional callable for component customization
mcp_names: Optional dictionary mapping operationId to component names
httpx_client_kwargs: Optional kwargs passed to httpx.AsyncClient.
Use this to configure timeout and other client settings.
tags: Optional set of tags to add to all components
**settings: Additional settings passed to FastMCP
Returns:
A FastMCP server with an OpenAPIProvider attached.
"""
from .providers.openapi import OpenAPIProvider
if httpx_client_kwargs is None:
httpx_client_kwargs = {}
httpx_client_kwargs.setdefault("base_url", "http://fastapi")
client = httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
**httpx_client_kwargs,
)
server_name = name or app.title
provider: Provider = OpenAPIProvider(
openapi_spec=app.openapi(),
client=client,
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
)
return cls(name=server_name, providers=[provider], **settings)
@classmethod
def generate_name(cls, name: str | None = None) -> str:
class_name = cls.__name__
if name is None:
return f"{class_name}-{secrets.token_hex(2)}"
else:
return f"{class_name}-{name}-{secrets.token_hex(2)}"
# -----------------------------------------------------------------------------
# Module-level Factory Functions
# -----------------------------------------------------------------------------
def create_proxy(
target: (
Client[ClientTransportT]
| ClientTransport
| FastMCP[Any]
| SDKServer
| AnyUrl
| Path
| MCPConfig
| dict[str, Any]
| str
),
**settings: Any,
) -> FastMCPProxy:
"""Create a FastMCP proxy server for the given target.
This is the recommended way to create a proxy server. For lower-level control,
use `FastMCPProxy` or `ProxyProvider` directly from `fastmcp.server.providers.proxy`.
Args:
target: The backend to proxy to. Can be:
- A Client instance (connected or disconnected)
- A ClientTransport
- A FastMCP server instance
- A URL string or AnyUrl
- A Path to a server script
- An MCPConfig or dict
**settings: Additional settings passed to FastMCPProxy (name, etc.)
Returns:
A FastMCPProxy server that proxies to the target.
Example:
```python
from fastmcp.server import create_proxy
# Create a proxy to a remote server
proxy = create_proxy("http://remote-server/mcp")
# Create a proxy to another FastMCP server
proxy = create_proxy(other_server)
```
"""
from fastmcp.server.providers.proxy import (
FastMCPProxy,
_create_client_factory,
)
client_factory = _create_client_factory(target)
return FastMCPProxy(
client_factory=client_factory,
**settings,
)
Reference in a new issue View git blame Copy permalink