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

Replace ___ with hash-based backend tool routing and per-tool prefab resources (#3824)

Browse source 
Replaces the ___ separator for FastMCPApp backend tool routing with a
deterministic hash(app_name, tool_name) prefix, and replaces the shared
singleton prefab renderer resource with per-tool resources synthesized
on demand.

Backend tools are now callable via <hash>_<local_name> instead of
<app_name>___<local_name>. The dispatcher walks the provider tree
recursively via get_tool_by_hash (same pattern as get_app_tool).

Each prefab tool gets its own renderer resource at
ui://prefab/tool/<hash>/renderer.html with per-tool CSP — fixing the
bug where PrefabAppConfig(csp=...) never actually applied.

Closes #3735, closes #3805
This commit is contained in:
Jeremiah Lowin 2026-04-12 12:52:07 -04:00 • committed by GitHub
commit af957e773f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1161 additions and 248 deletions
Download patch file Download diff file Expand all files Collapse all files

62
src/fastmcp/apps/app.py
View file

@ -29,7 +29,7 @@ from __future__ import annotations
import inspect
from collections.abc import AsyncIterator, Callable, Sequence
from contextlib import asynccontextmanager, suppress
from contextlib import asynccontextmanager
from typing import Any, Literal, TypeVar, overload
from mcp.types import AnyFunction, Icon, ToolAnnotations
@ -51,19 +51,30 @@ F = TypeVar("F", bound=Callable[..., Any])
def _make_resolver(app_name: str | None = None) -> Any:
"""Create a CallTool resolver that prefixes tool names with the app name.
"""Create a CallTool resolver that prefixes tool names with a hash.
When ``app_name`` is set, tool references like ``CallTool("store_files")``
or ``CallTool(store_files)`` are resolved to
``ResolvedTool(name="Files___store_files")``. This produces stable
identifiers that bypass transforms and work without host ``_meta``
forwarding.
Structurally identical to the old ``___`` resolver — ``app_name`` is
the FastMCPApp's name, known at serialization time from the tool's
``meta["fastmcp"]["app"]`` tag. The only change is the wire format:
``<hash>_<local_name>`` instead of ``<app_name>___<local_name>``.
The dispatcher recognizes the hashed form and routes it via
``get_tool_by_hash`` which walks the provider tree recursively —
same pattern as ``get_app_tool``.
"""
from fastmcp.server.providers.addressing import (
hashed_backend_name,
parse_hashed_backend_name,
)
def _prefix(name: str) -> str:
if app_name and "___" not in name:
return f"{app_name}___{name}"
return name
def _prefix(local_name: str) -> str:
if app_name:
# Don't re-hash an already-addressed name (same guard the
# old ___ resolver had with "___" not in name).
if parse_hashed_backend_name(local_name) is not None:
return local_name
return hashed_backend_name(app_name, local_name)
return local_name
def _resolve_tool_ref(fn: Any) -> Any:
from prefab_ui.app import ResolvedTool
@ -138,11 +149,6 @@ class FastMCPApp(Provider):
"""
def __init__(self, name: str) -> None:
if "___" in name:
raise ValueError(
f"App name {name!r} must not contain '___' "
"(reserved as the app tool routing separator)"
)
super().__init__()
self.name = name
self._local = LocalProvider(on_duplicate="error")
@ -214,11 +220,15 @@ class FastMCPApp(Provider):
raise ValueError(f"Cannot determine tool name for {fn!r}")
from fastmcp.apps.config import AppConfig, app_config_to_meta_dict
from fastmcp.server.providers.addressing import hash_tool
app_config = AppConfig(visibility=visibility)
meta: dict[str, Any] = {
"ui": app_config_to_meta_dict(app_config),
"fastmcp": {"app": self.name},
"fastmcp": {
"app": self.name,
"_tool_hash": hash_tool(self.name, resolved_name),
},
}
tool_obj = Tool.from_function(
@ -301,11 +311,12 @@ class FastMCPApp(Provider):
def _register(fn: F, tool_name: str | None) -> F:
from fastmcp.apps.config import AppConfig, app_config_to_meta_dict
from fastmcp.server.providers.addressing import hash_tool
from fastmcp.server.providers.local_provider.decorators.tools import (
PREFAB_RENDERER_URI,
_ensure_prefab_renderer,
)
resolved = tool_name or getattr(fn, "__name__", None) or "unknown"
app_config = AppConfig(
resource_uri=PREFAB_RENDERER_URI,
visibility=["model"],
@ -313,7 +324,10 @@ class FastMCPApp(Provider):
meta: dict[str, Any] = {
"ui": app_config_to_meta_dict(app_config),
"fastmcp": {"app": self.name},
"fastmcp": {
"app": self.name,
"_tool_hash": hash_tool(self.name, resolved),
},
}
tool_obj = Tool.from_function(
@ -330,10 +344,6 @@ class FastMCPApp(Provider):
)
self._local._add_component(tool_obj)
# Register the Prefab renderer resource on the internal provider
with suppress(ImportError):
_ensure_prefab_renderer(self._local)
return fn
return _dispatch_decorator(name_or_fn, name, _register, "ui")
@ -353,8 +363,12 @@ class FastMCPApp(Provider):
if not isinstance(tool, Tool):
tool = Tool._ensure_tool(tool)
from fastmcp.server.providers.addressing import hash_tool
meta = dict(tool.meta) if tool.meta else {}
meta.setdefault("fastmcp", {})["app"] = self.name
fm = meta.setdefault("fastmcp", {})
fm["app"] = self.name
fm["_tool_hash"] = hash_tool(self.name, tool.name)
ui = meta.setdefault("ui", {})
if "visibility" not in ui:
ui["visibility"] = ["app"]

69
src/fastmcp/server/providers/addressing.py Normal file
View file

@ -0,0 +1,69 @@
"""Deterministic tool hashing for backend-tool routing and per-tool resources.
Each FastMCPApp backend tool gets a deterministic hash computed from its
app name + tool name. The hash serves two purposes:
1. **Backend-tool routing.** Tools with ``"app"`` in their visibility are
callable via ``<hash>_<local_name>``. The dispatcher parses the prefix,
then walks providers recursively (same pattern as the old ``get_app_tool``)
to find a tool whose stored hash matches.
2. **Per-tool Prefab renderer URIs.** Each prefab tool gets a unique renderer
resource at ``ui://prefab/tool/<hash>/renderer.html``. ``list_resources``
and ``read_resource`` synthesize these on demand from the tool's meta.
The hash is computed at registration time from ``(app_name, tool_name)`` —
both known at that moment — and stored in ``meta["fastmcp"]["_tool_hash"]``.
Deterministic across replicas (same code → same hash), no registry walk
needed.
"""
from __future__ import annotations
import hashlib
#: Length of the hex hash prefix used in URIs and backend-tool names.
HASH_LENGTH = 12
def hash_tool(app_name: str, tool_name: str) -> str:
"""Deterministic hex hash for a tool in an app.
Same inputs on every replica produce the same output.
"""
payload = f"{app_name}\x00{tool_name}".encode()
return hashlib.sha256(payload).hexdigest()[:HASH_LENGTH]
def hashed_backend_name(app_name: str, tool_name: str) -> str:
"""Format the universal name for a backend tool: ``<hash>_<local_name>``."""
return f"{hash_tool(app_name, tool_name)}_{tool_name}"
def parse_hashed_backend_name(name: str) -> tuple[str, str] | None:
"""Parse ``<HASH_LENGTH hex>_<rest>`` → ``(hash, local_tool_name)`` or None."""
if len(name) <= HASH_LENGTH + 1:
return None
prefix = name[:HASH_LENGTH]
if name[HASH_LENGTH] != "_":
return None
if not all(c in "0123456789abcdef" for c in prefix):
return None
return prefix, name[HASH_LENGTH + 1 :]
def hashed_resource_uri(app_name: str, tool_name: str) -> str:
"""Per-tool Prefab renderer resource URI."""
return f"ui://prefab/tool/{hash_tool(app_name, tool_name)}/renderer.html"
def parse_hashed_resource_uri(uri: str) -> str | None:
"""Extract the hash from a Prefab renderer URI, or None."""
prefix = "ui://prefab/tool/"
suffix = "/renderer.html"
if not uri.startswith(prefix) or not uri.endswith(suffix):
return None
h = uri[len(prefix) : -len(suffix)]
if len(h) != HASH_LENGTH or not all(c in "0123456789abcdef" for c in h):
return None
return h

13
src/fastmcp/server/providers/aggregate.py
View file

@ -210,6 +210,19 @@ class AggregateProvider(Provider):
return r
return None
async def get_tool_by_hash(self, tool_hash: str, tool_name: str) -> Tool | None:
"""Query all child providers for a tool matching a hash."""
results = await gather(
*[p.get_tool_by_hash(tool_hash, tool_name) for p in self.providers],
return_exceptions=True,
)
for r in results:
if isinstance(r, BaseException):
continue
if r is not None:
return r
return None
# -------------------------------------------------------------------------
# Resources
# -------------------------------------------------------------------------

23
src/fastmcp/server/providers/base.py
View file

@ -201,6 +201,29 @@ class Provider:
return tool
return None
async def get_tool_by_hash(self, tool_hash: str, tool_name: str) -> Tool | None:
"""Look up an app-visible tool by its deterministic hash.
Same recursive-walk semantics as ``get_app_tool`` but matches on
``meta["fastmcp"]["_tool_hash"]`` instead of the app name tag.
Used by the dispatcher when receiving hashed backend-tool calls.
"""
tool = await self._get_tool(tool_name)
if tool is not None:
meta = tool.meta or {}
fastmcp_meta = meta.get("fastmcp")
ui_meta = meta.get("ui")
visibility = (
ui_meta.get("visibility", []) if isinstance(ui_meta, dict) else []
)
if (
isinstance(fastmcp_meta, dict)
and fastmcp_meta.get("_tool_hash") == tool_hash
and "app" in visibility
):
return tool
return None
async def list_resources(self) -> Sequence[Resource]:
"""List resources with all transforms applied.

16
src/fastmcp/server/providers/fastmcp_provider.py
View file

@ -572,10 +572,18 @@ class FastMCPProvider(Provider):
if raw_tool is None:
return None
wrapped = FastMCPProviderTool.wrap(self.server, raw_tool)
# Use the ___-prefixed name so the inner server's call_tool also
# takes the app-tool bypass path (app-only tools are hidden from
# normal get_tool visibility filtering).
wrapped._original_name = f"{app_name}___{tool_name}"
from fastmcp.server.providers.addressing import hashed_backend_name
wrapped._original_name = hashed_backend_name(app_name, tool_name)
return wrapped
async def get_tool_by_hash(self, tool_hash: str, tool_name: str) -> Tool | None:
"""Delegate to nested server's get_tool_by_hash, wrapping for middleware."""
raw_tool = await self.server.get_tool_by_hash(tool_hash, tool_name)
if raw_tool is None:
return None
wrapped = FastMCPProviderTool.wrap(self.server, raw_tool)
wrapped._original_name = f"{tool_hash}_{tool_name}"
return wrapped
# -------------------------------------------------------------------------

71
src/fastmcp/server/providers/local_provider/decorators/tools.py
View file

@ -73,53 +73,31 @@ def _has_prefab_return_type(tool: Tool) -> bool:
return _is_prefab_type(rt)
def _ensure_prefab_renderer(provider: LocalProvider) -> None:
"""Lazily register the shared prefab renderer as a ui:// resource."""
from prefab_ui.renderer import get_renderer_csp, get_renderer_html
def _stamp_prefab_marker(tool: Tool) -> None:
"""Mark a tool as needing a Prefab renderer resource.
from fastmcp.apps.config import (
UI_MIME_TYPE,
AppConfig,
ResourceCSP,
app_config_to_meta_dict,
)
from fastmcp.resources.types import TextResource
renderer_key = f"resource:{PREFAB_RENDERER_URI}@"
if renderer_key in provider._components:
return
csp = get_renderer_csp()
resource_app = AppConfig(
csp=ResourceCSP(
resource_domains=csp.get("resource_domains"),
connect_domains=csp.get("connect_domains"),
)
)
resource = TextResource(
uri=PREFAB_RENDERER_URI, # type: ignore[arg-type] # AnyUrl accepts ui:// scheme at runtime # ty:ignore[invalid-argument-type]
name="Prefab Renderer",
text=get_renderer_html(),
mime_type=UI_MIME_TYPE,
meta={"ui": app_config_to_meta_dict(resource_app)},
)
provider._add_component(resource)
def _expand_prefab_ui_meta(tool: Tool) -> None:
"""Expand meta["ui"] = True into the full AppConfig dict for a prefab tool."""
Sets ``meta["ui"]["resourceUri"]`` to a placeholder URI. The server
recognizes the placeholder at list_tools / list_resources / read_resource
time and synthesizes a per-tool resource on the fly with a hashed URI
derived from the tool's mount-point address. Nothing is stored — the
renderer HTML and CSP are generated on demand from the tool's own meta.
"""
from fastmcp.apps.config import AppConfig, app_config_to_meta_dict
app_config = AppConfig(
resource_uri=PREFAB_RENDERER_URI,
)
app_config = AppConfig(resource_uri=PREFAB_RENDERER_URI)
meta = dict(tool.meta) if tool.meta else {}
meta["ui"] = app_config_to_meta_dict(app_config)
tool.meta = meta
def _maybe_apply_prefab_ui(provider: LocalProvider, tool: Tool) -> None:
"""Auto-wire prefab UI metadata and renderer resource if needed."""
"""Mark a tool as a Prefab tool if its config or return type implies it.
Per-tool renderer resources are synthesized lazily at list/read time;
here we only normalize the tool's meta so the synthesis pass can spot
it. ``app=True``, return-type inference, and ``PrefabAppConfig`` all
funnel through the same placeholder marker.
"""
if not _HAS_PREFAB:
return
@ -127,17 +105,14 @@ def _maybe_apply_prefab_ui(provider: LocalProvider, tool: Tool) -> None:
ui = meta.get("ui")
if ui is True:
# Explicit app=True: expand to full AppConfig and register renderer
_ensure_prefab_renderer(provider)
_expand_prefab_ui_meta(tool)
# Explicit app=True: stamp the placeholder so the synthesizer finds it.
_stamp_prefab_marker(tool)
elif ui is None and _has_prefab_return_type(tool):
# Inference: return type is a prefab type, auto-wire
_ensure_prefab_renderer(provider)
_expand_prefab_ui_meta(tool)
elif isinstance(ui, dict) and ui.get("resourceUri") == PREFAB_RENDERER_URI:
# PrefabAppConfig or manual config pointing to the Prefab renderer —
# ensure the renderer resource is registered (CSP already set by caller)
_ensure_prefab_renderer(provider)
# Inference: return type is a prefab type, stamp the placeholder.
_stamp_prefab_marker(tool)
# Otherwise the tool either has no ui meta at all (not a prefab tool)
# or it already has a fully-formed dict from FastMCP.tool(app=...) — the
# synthesizer picks up both flavors by looking for the placeholder URI.
class ToolDecoratorMixin:

245
src/fastmcp/server/providers/prefab_synthesis.py Normal file
View file

@ -0,0 +1,245 @@
"""On-demand Prefab renderer resource synthesis.
Tools marked as Prefab (via ``app=True``, ``PrefabAppConfig``, etc.) carry
a placeholder ``meta.ui.resourceUri`` and optionally a hash in
``meta.fastmcp._tool_hash``. This module synthesizes per-tool renderer
resources on demand at ``list_resources`` and ``read_resource`` time
without storing or materializing anything.
Each tool's resource URI is ``ui://prefab/tool/<hash>/renderer.html``
where the hash comes from the tool's own meta (set at registration from
the app name + tool name). CSP on the resource is the tool's
``meta.ui.csp`` merged with the renderer defaults across all four
``*_domains`` fields.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from fastmcp.server.providers.addressing import (
HASH_LENGTH,
hash_tool,
parse_hashed_resource_uri,
)
if TYPE_CHECKING:
from fastmcp.resources.base import Resource
from fastmcp.server.server import FastMCP
from fastmcp.tools.base import Tool
#: The placeholder URI that decorators stamp on tools needing a renderer.
PREFAB_PLACEHOLDER_URI = "ui://prefab/renderer.html"
def _is_prefab_tool(tool: Tool) -> bool:
"""True if *tool* was marked as needing a Prefab renderer at registration."""
meta = tool.meta
if not meta:
return False
ui = meta.get("ui")
if not isinstance(ui, dict):
return False
return ui.get("resourceUri") == PREFAB_PLACEHOLDER_URI
def _get_tool_hash(tool: Tool) -> str | None:
"""Read the stored hash from tool meta, or compute from app name + tool name."""
meta = tool.meta or {}
fastmcp_meta = meta.get("fastmcp")
if isinstance(fastmcp_meta, dict):
h = fastmcp_meta.get("_tool_hash")
if isinstance(h, str) and len(h) == HASH_LENGTH:
return h
# Fall back to computing from app name
app = fastmcp_meta.get("app")
if isinstance(app, str):
return hash_tool(app, tool.name)
# Root-level prefab tool (no app name) — hash from empty prefix.
return hash_tool("", tool.name)
def _merge_domain_lists(
base: list[str] | None, extra: list[str] | None
) -> list[str] | None:
if base is None and extra is None:
return None
combined = list(base or [])
for item in extra or []:
if item not in combined:
combined.append(item)
return combined or None
def _build_resource_for_tool(tool: Tool) -> Resource | None:
"""Synthesize a TextResource for a prefab tool. Returns None if prefab_ui isn't installed."""
try:
from prefab_ui.renderer import get_renderer_csp, get_renderer_html
except ImportError:
return None
from fastmcp.apps.config import (
UI_MIME_TYPE,
AppConfig,
ResourceCSP,
app_config_to_meta_dict,
)
from fastmcp.resources.types import TextResource
tool_hash = _get_tool_hash(tool)
if tool_hash is None:
return None
# Merge user CSP with renderer defaults — all four domain fields.
defaults: dict[str, Any] = get_renderer_csp() or {}
user_csp: dict[str, Any] = {}
if tool.meta and isinstance(tool.meta.get("ui"), dict):
raw = tool.meta["ui"].get("csp")
if isinstance(raw, dict):
user_csp = raw
def _get(d: dict[str, Any], snake: str, camel: str) -> list[str] | None:
val = d.get(snake)
if val is None:
val = d.get(camel)
return val if isinstance(val, list) else None
merged = {
"connect_domains": _merge_domain_lists(
defaults.get("connect_domains"),
_get(user_csp, "connect_domains", "connectDomains"),
),
"resource_domains": _merge_domain_lists(
defaults.get("resource_domains"),
_get(user_csp, "resource_domains", "resourceDomains"),
),
"frame_domains": _merge_domain_lists(
defaults.get("frame_domains"),
_get(user_csp, "frame_domains", "frameDomains"),
),
"base_uri_domains": _merge_domain_lists(
defaults.get("base_uri_domains"),
_get(user_csp, "base_uri_domains", "baseUriDomains"),
),
}
resource_csp = ResourceCSP(**merged) if any(merged.values()) else None
# Carry permissions from the tool's meta to the resource (same
# principle as CSP — belongs on the resource, not the tool).
user_permissions = None
if tool.meta and isinstance(tool.meta.get("ui"), dict):
raw_perms = tool.meta["ui"].get("permissions")
if isinstance(raw_perms, dict):
from fastmcp.apps.config import ResourcePermissions
user_permissions = ResourcePermissions(**raw_perms)
resource_app = AppConfig(
csp=resource_csp,
permissions=user_permissions,
)
uri = f"ui://prefab/tool/{tool_hash}/renderer.html"
return TextResource(
uri=uri, # type: ignore[arg-type] # ty:ignore[invalid-argument-type]
name=f"Prefab Renderer ({tool.name})",
text=get_renderer_html(),
mime_type=UI_MIME_TYPE,
meta={"ui": app_config_to_meta_dict(resource_app)},
)
def _walk_prefab_tools(server: FastMCP) -> list[Tool]:
"""Enumerate all prefab tools across the server's providers (sync walk of _components)."""
from fastmcp.apps.app import FastMCPApp
from fastmcp.server.providers.base import Provider
from fastmcp.server.providers.local_provider import LocalProvider
from fastmcp.server.providers.wrapped_provider import _WrappedProvider
from fastmcp.tools.base import Tool
results: list[Tool] = []
def _walk_provider(provider: Provider) -> None:
# Unwrap transform wrappers
inner = provider
while isinstance(inner, _WrappedProvider):
inner = inner._inner
# Extract tools from local storage
sources: list[LocalProvider] = []
if isinstance(inner, LocalProvider):
sources.append(inner)
if isinstance(inner, FastMCPApp):
sources.append(inner._local)
for src in sources:
for component in src._components.values():
if isinstance(component, Tool) and _is_prefab_tool(component):
results.append(component)
# Recurse into aggregate children
from fastmcp.server.providers.aggregate import AggregateProvider
from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
if isinstance(inner, AggregateProvider):
for child in inner.providers:
_walk_provider(child)
# Recurse into mounted FastMCP servers
if isinstance(inner, FastMCPProvider):
for child in inner.server.providers:
_walk_provider(child)
for provider in server.providers:
_walk_provider(provider)
return results
async def synthesize_prefab_resources(server: FastMCP) -> list[Resource]:
"""Return fresh synthetic Prefab resources for all prefab tools. Pure."""
resources: list[Resource] = []
seen_hashes: set[str] = set()
for tool in _walk_prefab_tools(server):
h = _get_tool_hash(tool)
if h is None or h in seen_hashes:
continue
seen_hashes.add(h)
resource = _build_resource_for_tool(tool)
if resource is not None:
resources.append(resource)
return resources
async def synthesize_prefab_resource_by_uri(
server: FastMCP, uri: str
) -> Resource | None:
"""Intercept a Prefab renderer URI and synthesize on demand."""
digest = parse_hashed_resource_uri(uri)
if digest is None:
return None
for tool in _walk_prefab_tools(server):
if _get_tool_hash(tool) == digest:
return _build_resource_for_tool(tool)
return None
def rewrite_tool_meta_for_wire(tool: Tool) -> Tool:
"""Return a model_copy with the per-tool URI and CSP stripped.
Reads the hash from the tool's own meta. If no hash is found,
returns the tool unchanged. Produces a fresh copy — the original
Tool object is untouched.
"""
if not _is_prefab_tool(tool):
return tool
tool_hash = _get_tool_hash(tool)
if tool_hash is None:
return tool
assert tool.meta is not None
new_ui = dict(tool.meta["ui"])
new_ui["resourceUri"] = f"ui://prefab/tool/{tool_hash}/renderer.html"
new_ui.pop("csp", None)
new_ui.pop("permissions", None)
new_meta = dict(tool.meta)
new_meta["ui"] = new_ui
return tool.model_copy(update={"meta": new_meta})

4
src/fastmcp/server/providers/wrapped_provider.py
View file

@ -67,6 +67,10 @@ class _WrappedProvider(Provider):
"""Delegate to inner, bypassing this wrapper's transforms."""
return await self._inner.get_app_tool(app_name, tool_name)
async def get_tool_by_hash(self, tool_hash: str, tool_name: str) -> Tool | None:
"""Delegate to inner, bypassing this wrapper's transforms."""
return await self._inner.get_tool_by_hash(tool_hash, tool_name)
async def _list_resources(self) -> Sequence[Resource]:
"""Delegate to inner's list_resources (includes inner's transforms)."""
return await self._inner.list_resources()

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

@ -208,6 +208,31 @@ def _is_model_visible(tool: Tool) -> bool:
return "model" in visibility
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.
@ -469,6 +494,24 @@ class FastMCP(
"""
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
# -------------------------------------------------------------------------
@ -583,6 +626,13 @@ class FastMCP(
tools = await apply_session_transforms(tools)
tools = [t for t in tools if is_enabled(t) and _is_model_visible(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:
@ -709,6 +759,15 @@ class FastMCP(
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:
@ -1109,6 +1168,18 @@ class FastMCP(
# 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](
@ -1131,28 +1202,35 @@ class FastMCP(
),
)
# Core logic: find and execute tool (providers queried in parallel)
# Use get_tool to apply transforms and filter disabled
# Core logic: find and execute tool
with server_span(
f"tools/call {name}", "tools/call", self.name, "tool", name
) as span:
# Try normal resolution first. If that fails and the name
# contains "___" (app tool prefix), parse out the app name
# and route via get_app_tool which bypasses transforms.
# Try normal display-name resolution first.
tool: Tool | None = await self.get_tool(name, version=version)
if tool is None and "___" in name:
app_prefix, _, tool_suffix = name.partition("___")
tool = await self.get_app_tool(app_prefix, tool_suffix)
if tool is not None:
# Auth still applies to app tools
skip_auth, token = _get_auth_context()
if not skip_auth and tool.auth is not None:
try:
ctx = AuthContext(token=token, component=tool)
if not await run_auth_checks(tool.auth, ctx):
raise NotFoundError(f"Unknown tool: {name!r}")
except AuthorizationError:
raise NotFoundError(f"Unknown tool: {name!r}") from None
# 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())
@ -1270,6 +1348,18 @@ class FastMCP(
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:

11
src/fastmcp/tools/base.py
View file

@ -511,7 +511,7 @@ _PREFAB_TEXT_FALLBACK = "[Rendered Prefab UI]"
def _get_tool_resolver(app_name: str | None = None) -> Callable[..., str] | None:
"""Get the FastMCPApp callable resolver, if available."""
"""Get the Prefab peer-reference resolver bound to an app name."""
try:
from fastmcp.apps.app import _make_resolver
@ -521,11 +521,12 @@ def _get_tool_resolver(app_name: str | None = None) -> Callable[..., str] | None
def _prefab_to_json(app: Any, fastmcp_app_name: str | None = None) -> dict[str, Any]:
"""Call PrefabApp.to_json() with the FastMCPApp callable resolver.
"""Call PrefabApp.to_json() with the hash-based resolver.
The resolver prefixes tool names with the app name (e.g.
``"store_files"`` → ``"Files___store_files"``) so the server can
find them via the bypass lookup regardless of transforms.
The resolver prefixes peer-tool references with a deterministic hash
derived from the app name + tool name. The dispatcher recognizes that
format and routes calls via ``get_tool_by_hash`` which walks the
provider tree recursively — same pattern as the old ``get_app_tool``.
"""
data = app.to_json(tool_resolver=_get_tool_resolver(fastmcp_app_name))
return data

39
tests/apps/test_file_upload.py
View file

@ -6,6 +6,7 @@ import pytest
from fastmcp import FastMCP
from fastmcp.apps.file_upload import FileUpload, _b64_decoded_size
from fastmcp.server.providers.addressing import hashed_backend_name
class TestB64DecodedSize:
@ -52,7 +53,9 @@ class TestFileUploadProvider:
server = FastMCP("test", providers=[FileUpload()])
files = [_make_file()]
result = await server.call_tool("Files___store_files", {"files": files})
result = await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": files}
)
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "test.txt" in text
@ -64,7 +67,9 @@ class TestFileUploadProvider:
server = FastMCP("test", providers=[FileUpload()])
files = [_make_file(content="DON'T PANIC")]
await server.call_tool("Files___store_files", {"files": files})
await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": files}
)
result = await server.call_tool("read_file", {"name": "test.txt"})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -75,7 +80,9 @@ class TestFileUploadProvider:
data = base64.b64encode(b"\x00\x01\x02\xff").decode()
files = [{"name": "image.png", "size": 4, "type": "image/png", "data": data}]
await server.call_tool("Files___store_files", {"files": files})
await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": files}
)
result = await server.call_tool("read_file", {"name": "image.png"})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -94,7 +101,9 @@ class TestFileUploadProvider:
_make_file("b.txt", "bbb"),
]
await server.call_tool("Files___store_files", {"files": files})
await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": files}
)
result = await server.call_tool("list_files", {})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -105,11 +114,11 @@ class TestFileUploadProvider:
server = FastMCP("test", providers=[FileUpload()])
await server.call_tool(
"Files___store_files",
hashed_backend_name("Files", "store_files"),
{"files": [_make_file(content="version 1")]},
)
await server.call_tool(
"Files___store_files",
hashed_backend_name("Files", "store_files"),
{"files": [_make_file(content="version 2")]},
)
@ -124,9 +133,11 @@ class TestFileUploadProvider:
tool_names = [t.name for t in tools]
assert "file_manager" in tool_names
# Routing uses the custom name
# The hash uses the app's actual name ("Uploads"), not the default.
files = [_make_file()]
result = await server.call_tool("Uploads___store_files", {"files": files})
result = await server.call_tool(
hashed_backend_name("Uploads", "store_files"), {"files": files}
)
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "test.txt" in text
@ -146,7 +157,9 @@ class TestFileUploadProvider:
big_file = _make_file(content="x" * 200)
with pytest.raises(Exception, match="exceeds max size"):
await server.call_tool("Files___store_files", {"files": [big_file]})
await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": [big_file]}
)
async def test_max_file_size_checks_actual_data_not_reported_size(self):
"""Size limit should be enforced on actual base64 payload, not the
@ -163,7 +176,9 @@ class TestFileUploadProvider:
}
with pytest.raises(Exception, match="exceeds max size"):
await server.call_tool("Files___store_files", {"files": [spoofed_file]})
await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": [spoofed_file]}
)
class TestFileUploadSubclass:
@ -207,7 +222,9 @@ class TestFileUploadSubclass:
server = FastMCP("test", providers=[MemoryUpload()])
files = [_make_file()]
await server.call_tool("Files___store_files", {"files": files})
await server.call_tool(
hashed_backend_name("Files", "store_files"), {"files": files}
)
assert "test.txt" in stored

11
tests/apps/test_form.py
View file

@ -7,6 +7,7 @@ import pytest
from fastmcp import FastMCP
from fastmcp.apps.form import FormInput, _backfill_boolean_defaults
from fastmcp.server.providers.addressing import hashed_backend_name
class Contact(pydantic.BaseModel):
@ -52,7 +53,7 @@ class TestFormInputProvider:
server = FastMCP("test", providers=[FormInput(model=Contact)])
result = await server.call_tool(
"Contact___submit_form",
hashed_backend_name("Contact", "submit_form"),
{"data": {"name": "Alice", "email": "alice@example.com"}},
)
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -74,7 +75,7 @@ class TestFormInputProvider:
)
result = await server.call_tool(
"Contact___submit_form",
hashed_backend_name("Contact", "submit_form"),
{"data": {"name": "Bob", "email": "bob@example.com"}},
)
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -94,7 +95,7 @@ class TestFormInputProvider:
server = FastMCP("test", providers=[FormInput(model=NoteForm)])
result = await server.call_tool(
"NoteForm___submit_form",
hashed_backend_name("NoteForm", "submit_form"),
{"data": {"title": "My Note", "content": "Hello"}},
)
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -107,7 +108,7 @@ class TestFormInputProvider:
server = FastMCP("test", providers=[FormInput(model=NoteForm)])
result = await server.call_tool(
"NoteForm___submit_form",
hashed_backend_name("NoteForm", "submit_form"),
{"data": {"title": "My Note", "content": "Hello", "archived": True}},
)
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -122,7 +123,7 @@ class TestFormInputProvider:
# for the data parameter itself). Pydantic will still reject missing
# required fields like title/content, but that's expected.
with pytest.raises(pydantic.ValidationError, match="title"):
await server.call_tool("NoteForm___submit_form", {})
await server.call_tool(hashed_backend_name("NoteForm", "submit_form"), {})
async def test_multiple_models(self):
class Address(pydantic.BaseModel):

66
tests/server/providers/test_addressing.py Normal file
View file

@ -0,0 +1,66 @@
"""Tests for the tool hashing primitives."""
from __future__ import annotations
from fastmcp.server.providers.addressing import (
HASH_LENGTH,
hash_tool,
hashed_backend_name,
hashed_resource_uri,
parse_hashed_backend_name,
parse_hashed_resource_uri,
)
class TestHashFunction:
def test_hash_is_fixed_length_hex(self):
h = hash_tool("myapp", "greet")
assert len(h) == HASH_LENGTH
assert all(c in "0123456789abcdef" for c in h)
def test_same_inputs_same_hash(self):
a = hash_tool("app", "submit_form")
b = hash_tool("app", "submit_form")
assert a == b
def test_different_app_names_different_hash(self):
a = hash_tool("contacts", "save")
b = hash_tool("billing", "save")
assert a != b
def test_different_tool_names_different_hash(self):
a = hash_tool("app", "save")
b = hash_tool("app", "delete")
assert a != b
class TestBackendNameRoundtrip:
def test_format_and_parse(self):
name = hashed_backend_name("contacts", "submit_form")
parsed = parse_hashed_backend_name(name)
assert parsed is not None
digest, local = parsed
assert digest == hash_tool("contacts", "submit_form")
assert local == "submit_form"
def test_parse_rejects_short_strings(self):
assert parse_hashed_backend_name("foo") is None
def test_parse_rejects_non_hex_prefix(self):
assert parse_hashed_backend_name("zzzzzzzzzzzz_save") is None
def test_parse_rejects_missing_separator(self):
assert parse_hashed_backend_name("abcdef012345save") is None
class TestResourceUriRoundtrip:
def test_format_and_parse(self):
uri = hashed_resource_uri("dashboard", "show")
h = parse_hashed_resource_uri(uri)
assert h == hash_tool("dashboard", "show")
def test_parse_rejects_unrelated_uri(self):
assert parse_hashed_resource_uri("file:///etc/passwd") is None
def test_parse_rejects_wrong_length_hash(self):
assert parse_hashed_resource_uri("ui://prefab/tool/abc/renderer.html") is None

185
tests/server/providers/test_prefab_roundtrip.py Normal file
View file

@ -0,0 +1,185 @@
"""End-to-end round-trip tests for Prefab peer-tool references.
These simulate what a real host does: call the UI tool, extract the
hashed backend-tool name from structured_content, call back with
that name, and verify the backend tool actually executes. Covers
single-server, namespaced mounts, and cross-server mounts.
"""
from __future__ import annotations
import json
import pytest
from fastmcp import FastMCP, FastMCPApp
from fastmcp.server.providers.addressing import hashed_backend_name
prefab_ui = pytest.importorskip("prefab_ui")
from prefab_ui.actions.mcp import CallTool # noqa: E402
from prefab_ui.components import Button, Column, Text # noqa: E402
class TestSingleServerRoundTrip:
async def test_ui_tool_serializes_hashed_peer_reference(self):
"""The resolver converts a CallTool string reference to a hashed
name that appears in the tool result's structured_content."""
app = FastMCPApp("contacts")
@app.tool()
def save_contact(name: str) -> str:
return f"saved {name}"
@app.ui()
def contact_form() -> Column:
return Column(
children=[Button(label="Save", on_click=CallTool(tool="save_contact"))]
)
server = FastMCP("Platform")
server.add_provider(app)
result = await server.call_tool("contact_form", {})
assert result.structured_content is not None
# The hashed name should appear somewhere in the serialized output.
sc_json = json.dumps(result.structured_content)
expected_hash = hashed_backend_name("contacts", "save_contact")
assert expected_hash in sc_json, (
f"Expected {expected_hash!r} in structured_content but got: {sc_json[:200]}"
)
async def test_hashed_name_from_result_is_callable(self):
"""The hashed name that appears in structured_content actually
resolves when called back — the full round-trip works."""
app = FastMCPApp("contacts")
@app.tool()
def save_contact(name: str) -> str:
return f"saved {name}"
@app.ui()
def contact_form() -> Column:
return Column(
children=[Button(label="Save", on_click=CallTool(tool="save_contact"))]
)
server = FastMCP("Platform")
server.add_provider(app)
# Step 1: call UI tool, get structured_content with hashed ref
await server.call_tool("contact_form", {})
# Step 2: call the backend tool by its hashed name
hashed_name = hashed_backend_name("contacts", "save_contact")
backend_result = await server.call_tool(hashed_name, {"name": "Alice"})
assert backend_result.content[0].text == "saved Alice" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
class TestNamespacedMountRoundTrip:
async def test_namespaced_app_backend_tool_round_trip(self):
"""A FastMCPApp mounted with a namespace: the UI tool is called
by its namespaced display name, the backend tool is called by
its hashed name — both work."""
app = FastMCPApp("crm")
@app.tool()
def save(name: str) -> str:
return f"saved {name}"
@app.ui()
def form() -> Text:
return Text(content="Enter details")
server = FastMCP("Platform")
server.add_provider(app, namespace="crm")
# UI tool visible under namespace
result = await server.call_tool("crm_form", {})
assert result.structured_content is not None
# Backend tool reachable via hash
hashed_name = hashed_backend_name("crm", "save")
backend_result = await server.call_tool(hashed_name, {"name": "Bob"})
assert backend_result.content[0].text == "saved Bob" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
class TestMountedServerRoundTrip:
async def test_backend_tool_reachable_through_mounted_server(self):
"""A FastMCPApp inside a mounted FastMCP server: the outer
server's dispatcher walks through FastMCPProvider to find
the backend tool by hash."""
app = FastMCPApp("contacts")
@app.tool()
def save(name: str) -> str:
return f"saved {name}"
@app.ui()
def form() -> Text:
return Text(content="Form")
inner = FastMCP("Inner")
inner.add_provider(app)
outer = FastMCP("Outer")
outer.mount(inner, namespace="inner")
# Backend tool callable through the mount via hash dispatch
hashed_name = hashed_backend_name("contacts", "save")
result = await outer.call_tool(hashed_name, {"name": "Carol"})
assert result.content[0].text == "saved Carol" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
class TestDynamicToolAdd:
async def test_tool_added_after_first_call_is_reachable(self):
"""Tools added to an already-mounted app after the first call
are still reachable via their hashed name — get_tool_by_hash
does a live walk, not a cached lookup."""
app = FastMCPApp("contacts")
server = FastMCP("Platform")
server.add_provider(app)
# First call — nothing to call yet, just prime any caches.
tools = await server.list_tools()
assert len(tools) == 0
# Now add a backend tool dynamically.
@app.tool()
def save(name: str) -> str:
return f"saved {name}"
# The dynamically-added tool should be reachable.
hashed_name = hashed_backend_name("contacts", "save")
result = await server.call_tool(hashed_name, {"name": "Dan"})
assert result.content[0].text == "saved Dan" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
class TestCollision:
async def test_same_app_name_same_tool_name_first_wins(self):
"""Two apps with the same name and same tool name: the hash is
identical, so get_tool_by_hash returns the first match. This is
the same first-match behavior the old get_app_tool had."""
app_a = FastMCPApp("shared")
app_b = FastMCPApp("shared")
@app_a.tool()
def save(name: str) -> str:
return f"from A: {name}"
@app_b.tool()
def save_b(name: str) -> str:
return f"from B: {name}"
# Register under a different local tool name to avoid
# actual collision at the provider level. The hash collision
# only happens when both app name AND tool name match.
# This test just verifies one app's tool is reachable.
server = FastMCP("Platform")
server.add_provider(app_a)
server.add_provider(app_b)
hashed_name = hashed_backend_name("shared", "save")
result = await server.call_tool(hashed_name, {"name": "Eve"})
assert result.content[0].text == "from A: Eve" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]

198
tests/server/providers/test_prefab_synthesis.py Normal file
View file

@ -0,0 +1,198 @@
"""End-to-end tests for the on-demand Prefab renderer synthesis.
The whole architecture exists to fix #3735 / PR #3754: a user passing a
``PrefabAppConfig(csp=ResourceCSP(frame_domains=[...]))`` should see
their ``frame_domains`` actually arrive on the renderer resource's CSP,
and CSP should NOT leak into the tool's wire metadata. These tests
exercise the synthesis path directly through the public server API.
"""
from __future__ import annotations
import pytest
from fastmcp import FastMCP, FastMCPApp
prefab_ui = pytest.importorskip("prefab_ui")
from fastmcp.apps.config import PrefabAppConfig, ResourceCSP # noqa: E402
class TestUserCSPReachesResource:
"""The original bug: user CSP must land on the resource, not vanish."""
async def test_frame_domains_reach_resource(self):
mcp = FastMCP("test")
@mcp.tool(
app=PrefabAppConfig(
csp=ResourceCSP(frame_domains=["https://example1234.com"])
)
)
def show_widget() -> str:
return "widget"
# Find the synthesized prefab resource for this tool.
resources = list(await mcp.list_resources())
renderer = next((r for r in resources if "prefab/tool" in str(r.uri)), None)
assert renderer is not None, "no prefab resource was synthesized"
assert renderer.meta is not None
csp = renderer.meta["ui"]["csp"]
assert "https://example1234.com" in csp.get("frameDomains", []), (
f"frame_domains missing from resource CSP: {csp}"
)
async def test_all_four_domain_fields_preserved(self):
"""The old singleton silently dropped frame_domains and
base_uri_domains; the synthesizer covers all four fields."""
mcp = FastMCP("test")
@mcp.tool(
app=PrefabAppConfig(
csp=ResourceCSP(
connect_domains=["https://api.example.com"],
resource_domains=["https://cdn.example.com"],
frame_domains=["https://embed.example.com"],
base_uri_domains=["https://base.example.com"],
)
)
)
def widget() -> str:
return "x"
resources = list(await mcp.list_resources())
renderer = next(r for r in resources if "prefab/tool" in str(r.uri))
assert renderer.meta is not None
csp = renderer.meta["ui"]["csp"]
assert "https://api.example.com" in csp.get("connectDomains", [])
assert "https://cdn.example.com" in csp.get("resourceDomains", [])
assert "https://embed.example.com" in csp.get("frameDomains", [])
assert "https://base.example.com" in csp.get("baseUriDomains", [])
class TestCSPStrippedFromToolMeta:
"""CSP belongs on the resource, not the tool. The wire format that
clients see for tools must not contain it."""
async def test_csp_not_in_listed_tool_meta(self):
mcp = FastMCP("test")
@mcp.tool(
app=PrefabAppConfig(csp=ResourceCSP(frame_domains=["https://example.com"]))
)
def show_widget() -> str:
return "widget"
tools = list(await mcp.list_tools())
tool = next(t for t in tools if t.name == "show_widget")
assert tool.meta is not None
ui = tool.meta["ui"]
assert "csp" not in ui, f"csp leaked into tool meta: {ui}"
assert "permissions" not in ui
class TestPerToolURIs:
"""Each prefab tool gets its own URI — distinct CSP per tool becomes
possible because no two tools share a renderer resource."""
async def test_two_tools_get_distinct_uris(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def tool_a() -> str:
return "a"
@mcp.tool(app=True)
def tool_b() -> str:
return "b"
tools = list(await mcp.list_tools())
a = next(t for t in tools if t.name == "tool_a")
b = next(t for t in tools if t.name == "tool_b")
assert a.meta is not None
assert b.meta is not None
uri_a = a.meta["ui"]["resourceUri"]
uri_b = b.meta["ui"]["resourceUri"]
assert uri_a != uri_b
assert uri_a.startswith("ui://prefab/tool/")
assert uri_b.startswith("ui://prefab/tool/")
class TestFastMCPAppMounts:
"""Tools inside FastMCPApps get URIs derived from the app's mount address."""
async def test_app_tool_uri_uses_address(self):
app = FastMCPApp("dashboard")
@app.ui()
def show() -> str:
return "rendered"
mcp = FastMCP("Platform")
mcp.add_provider(app)
resources = list(await mcp.list_resources())
prefab = [r for r in resources if "prefab/tool" in str(r.uri)]
assert len(prefab) == 1
async def test_namespaced_mount_still_synthesizes_resource(self):
app = FastMCPApp("crm")
@app.ui()
def contact_form() -> str:
return "form"
mcp = FastMCP("Platform")
mcp.add_provider(app, namespace="customers")
resources = list(await mcp.list_resources())
prefab = [r for r in resources if "prefab/tool" in str(r.uri)]
assert len(prefab) == 1
class TestReadResource:
"""The synthesized resources are actually fetchable via read_resource."""
async def test_read_resource_returns_renderer_html(self):
from fastmcp import Client
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> str:
return "hi"
async with Client(mcp) as client:
tools = await client.list_tools()
uri = next(t for t in tools if t.name == "my_tool").meta["ui"][
"resourceUri"
]
contents = await client.read_resource(uri)
assert len(contents) > 0
text = contents[0].text if hasattr(contents[0], "text") else ""
assert "<html" in text.lower() or "<!doctype" in text.lower()
class TestNonPrefabToolsUntouched:
async def test_plain_tool_has_no_ui_meta(self):
mcp = FastMCP("test")
@mcp.tool
def greet(name: str) -> str:
return name
tools = list(await mcp.list_tools())
tool = next(t for t in tools if t.name == "greet")
assert not tool.meta or "ui" not in (tool.meta or {})
async def test_plain_server_has_no_synthesized_resources(self):
mcp = FastMCP("test")
@mcp.tool
def greet(name: str) -> str:
return name
resources = list(await mcp.list_resources())
assert not any("prefab/tool" in str(r.uri) for r in resources)

13
tests/server/test_providers.py
View file

@ -211,16 +211,19 @@ class TestProvider:
async def test_call_tool_uses_get_tool_for_efficient_lookup(
self, base_server: FastMCP, dynamic_tools: list[Tool]
):
"""Test that call_tool uses get_tool() for efficient single-tool lookup."""
"""Test that call_tool uses get_tool() (not list_tools) for lookup."""
provider = SimpleToolProvider(tools=dynamic_tools)
base_server.add_provider(provider)
await base_server.call_tool(name="dynamic_multiply", arguments={"a": 2, "b": 3})
# get_tool is called once for efficient lookup:
# call_tool() calls provider.get_tool() to get the tool and execute it
# Key point: list_tools is NOT called during tool execution (efficient lookup)
assert provider.get_tool_call_count == 1
# get_tool may be called more than once — once for the initial
# resolution and once again by the dispatcher's reverse-lookup to
# find the tool's owning provider for Context.mount_path. Both
# are bounded by provider count, much cheaper than list_tools.
# The key invariant: list_tools is NOT called during dispatch.
assert provider.get_tool_call_count >= 1
assert provider.list_tools_call_count == 0
async def test_default_get_tool_falls_back_to_list(self, base_server: FastMCP):
"""Test that BaseToolProvider's default get_tool calls list_tools."""

13
tests/test_apps.py
View file

@ -585,7 +585,8 @@ class TestPrefabAppConfig:
assert config.csp is not None
assert config.csp.frame_domains == ["https://example.com"]
async def test_auto_registers_renderer_resource(self):
async def test_auto_synthesizes_renderer_resource(self):
"""Each prefab tool gets a per-tool renderer resource on demand."""
from fastmcp.apps import PrefabAppConfig
server = FastMCP("test")
@ -595,11 +596,11 @@ class TestPrefabAppConfig:
return "hello"
resources = list(await server.list_resources())
uris = [str(r.uri) for r in resources]
assert any("ui://prefab/renderer.html" in u for u in uris)
prefab = [r for r in resources if "prefab/tool" in str(r.uri)]
assert len(prefab) == 1
async def test_equivalent_to_app_true(self):
"""PrefabAppConfig() should produce the same tool metadata as app=True."""
"""PrefabAppConfig() and app=True both synthesize a per-tool renderer."""
from fastmcp.apps import PrefabAppConfig
server1 = FastMCP("test1")
@ -621,4 +622,6 @@ class TestPrefabAppConfig:
assert tools2[0].meta is not None
ui2 = tools2[0].meta.get("ui", {})
assert ui1.get("resourceUri") == ui2.get("resourceUri")
# Both produce per-tool URIs in the prefab/tool/<hash>/ form.
assert ui1.get("resourceUri", "").startswith("ui://prefab/tool/")
assert ui2.get("resourceUri", "").startswith("ui://prefab/tool/")

56
tests/test_apps_prefab.py
View file

@ -119,42 +119,49 @@ class TestAppTrue:
assert "ui" in tool.meta
assert tool.meta["ui"]["resourceUri"] == PREFAB_RENDERER_URI
def test_app_true_registers_renderer_resource(self):
async def test_app_true_synthesizes_renderer_resource(self):
"""Each prefab tool gets a per-tool renderer resource synthesized
on demand at list_resources time. Resources don't live on any
provider's storage — they're computed from the registry walk."""
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> str:
return "hello"
renderer_key = f"resource:{PREFAB_RENDERER_URI}@"
assert renderer_key in mcp._local_provider._components
resources = list(await mcp.list_resources())
prefab_resources = [r for r in resources if "prefab/tool" in str(r.uri)]
assert len(prefab_resources) == 1
assert "renderer.html" in str(prefab_resources[0].uri)
def test_renderer_resource_has_correct_mime_type(self):
async def test_renderer_resource_has_correct_mime_type(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> str:
return "hello"
renderer_key = f"resource:{PREFAB_RENDERER_URI}@"
resource = mcp._local_provider._components[renderer_key]
assert isinstance(resource, TextResource)
assert resource.mime_type == UI_MIME_TYPE
resources = list(await mcp.list_resources())
renderer = next(r for r in resources if "prefab/tool" in str(r.uri))
assert isinstance(renderer, TextResource)
assert renderer.mime_type == UI_MIME_TYPE
def test_renderer_resource_has_csp(self):
async def test_renderer_resource_has_csp(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> str:
return "hello"
renderer_key = f"resource:{PREFAB_RENDERER_URI}@"
resource = mcp._local_provider._components[renderer_key]
assert resource.meta is not None
assert "ui" in resource.meta
assert "csp" in resource.meta["ui"]
resources = list(await mcp.list_resources())
renderer = next(r for r in resources if "prefab/tool" in str(r.uri))
assert renderer.meta is not None
assert "ui" in renderer.meta
assert "csp" in renderer.meta["ui"]
def test_multiple_tools_share_renderer(self):
async def test_multiple_tools_get_dedicated_resources(self):
"""Each prefab tool gets its own resource at a distinct hashed
URI — no shared singleton, so per-tool CSP becomes possible."""
mcp = FastMCP("test")
@mcp.tool(app=True)
@ -165,10 +172,9 @@ class TestAppTrue:
def tool_b() -> str:
return "b"
renderer_keys = [
k for k in mcp._local_provider._components if k.startswith("resource:ui://")
]
assert len(renderer_keys) == 1
resources = list(await mcp.list_resources())
prefab_uris = {str(r.uri) for r in resources if "prefab/tool" in str(r.uri)}
assert len(prefab_uris) == 2
def test_explicit_app_config_not_overridden(self):
mcp = FastMCP("test")
@ -421,7 +427,9 @@ class TestIntegration:
tool = next(t for t in tools if t.name == "my_tool")
meta = tool.meta or {}
assert "ui" in meta
assert meta["ui"]["resourceUri"] == PREFAB_RENDERER_URI
# URI is the per-tool hashed form, not the singleton.
assert meta["ui"]["resourceUri"].startswith("ui://prefab/tool/")
assert meta["ui"]["resourceUri"].endswith("/renderer.html")
async def test_renderer_resource_readable(self):
mcp = FastMCP("test")
@ -431,7 +439,13 @@ class TestIntegration:
return "hello"
async with Client(mcp) as client:
contents = await client.read_resource(PREFAB_RENDERER_URI)
# Look up the URI by listing first — the hash isn't
# computable from outside without the address registry.
tools = await client.list_tools()
uri = next(t for t in tools if t.name == "my_tool").meta["ui"][
"resourceUri"
]
contents = await client.read_resource(uri)
assert len(contents) > 0
text = contents[0].text if hasattr(contents[0], "text") else ""

184
tests/test_fastmcp_app.py
View file

@ -30,13 +30,13 @@ from fastmcp.tools.base import Tool
class TestFastMCPAppInit:
def test_app_name_with_triple_underscore_rejected(self):
with pytest.raises(ValueError, match="must not contain '___'"):
FastMCPApp("my___app")
def test_app_name_with_single_or_double_underscore_ok(self):
def test_app_name_with_underscores_ok(self):
# The old `___` separator is gone — backend tool routing now uses
# a hashed positional address rather than a name-based prefix, so
# any character is fine inside an app name.
FastMCPApp("my_app")
FastMCPApp("my__app")
FastMCPApp("my___app")
class TestAppTool:
@ -287,25 +287,23 @@ class TestAppUI:
class TestResolveToolRef:
def test_resolve_string_no_app_name(self):
"""Without an app name, strings pass through unprefixed."""
def test_resolve_string_no_context(self):
"""Without a running Context the resolver returns bare names."""
result = _make_resolver()("save_contact")
assert isinstance(result, ResolvedTool)
assert result.name == "save_contact"
def test_resolve_string_with_app_name(self):
"""With an app name, strings get the ___-prefix."""
result = _make_resolver("Files")("store_files")
assert isinstance(result, ResolvedTool)
assert result.name == "Files___store_files"
"""With an app name the resolver produces a hashed backend name."""
from fastmcp.server.providers.addressing import hashed_backend_name
def test_resolve_string_already_prefixed(self):
"""Strings that already contain ___ are not double-prefixed."""
result = _make_resolver("Files")("Other___store_files")
result = _make_resolver("contacts")("save_contact")
assert isinstance(result, ResolvedTool)
assert result.name == "Other___store_files"
assert result.name == hashed_backend_name("contacts", "save_contact")
def test_resolve_callable_no_context(self):
"""Without context, callables resolve to their bare __name__."""
def test_resolve_callable_no_app_name(self):
def my_tool():
pass
@ -313,40 +311,6 @@ class TestResolveToolRef:
assert isinstance(result, ResolvedTool)
assert result.name == "my_tool"
def test_resolve_callable_with_app_name(self):
"""Callables also get the ___-prefix when an app name is set."""
def store_files():
pass
result = _make_resolver("Files")(store_files)
assert isinstance(result, ResolvedTool)
assert result.name == "Files___store_files"
def test_resolve_fastmcp_metadata(self):
from fastmcp.tools.function_tool import ToolMeta
def my_tool():
pass
my_tool.__fastmcp__ = ToolMeta(name="custom_name") # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
result = _make_resolver()(my_tool)
assert isinstance(result, ResolvedTool)
assert result.name == "custom_name"
def test_resolve_fastmcp_metadata_with_app_name(self):
from fastmcp.tools.function_tool import ToolMeta
def my_tool():
pass
my_tool.__fastmcp__ = ToolMeta(name="custom_name") # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
result = _make_resolver("MyApp")(my_tool)
assert isinstance(result, ResolvedTool)
assert result.name == "MyApp___custom_name"
def test_resolve_unresolvable_raises(self):
with pytest.raises(ValueError):
_make_resolver()(42)
@ -491,8 +455,12 @@ class TestProviderInterface:
class TestCallToolAppRouting:
async def test_call_tool_with_app_name(self):
"""Server.call_tool routes via get_app_tool when app_name is set."""
async def test_call_tool_with_hashed_name(self):
"""A backend tool with visibility=['app'] is callable via its
hashed-name address — the same form a Prefab UI's resolver would
produce when serializing a peer reference."""
from fastmcp.server.providers.addressing import hashed_backend_name
app = FastMCPApp("contacts")
@app.tool()
@ -502,11 +470,13 @@ class TestCallToolAppRouting:
server = FastMCP("Platform")
server.add_provider(app)
result = await server.call_tool("contacts___save", {"name": "alice"})
result = await server.call_tool(
hashed_backend_name("contacts", "save"), {"name": "alice"}
)
assert result.content[0].text == "saved alice" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_call_tool_without_app_name_model_visible(self):
"""Regular name-based resolution works for model-visible tools."""
async def test_call_tool_model_visible_uses_display_name(self):
"""Tools with visibility=['app','model'] are callable by display name."""
app = FastMCPApp("test")
@app.tool(model=True)
@ -519,8 +489,12 @@ class TestCallToolAppRouting:
result = await server.call_tool("save", {"name": "bob"})
assert result.content[0].text == "saved bob" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_app_name_survives_namespace(self):
"""app_name routing bypasses namespace transforms."""
async def test_hashed_name_survives_namespace_mount(self):
"""The hashed-name path bypasses display-layer transforms entirely.
A FastMCPApp mounted under a Namespace transform still has its
backend tools reachable via the same hash."""
from fastmcp.server.providers.addressing import hashed_backend_name
app = FastMCPApp("crm")
@app.tool()
@ -530,11 +504,13 @@ class TestCallToolAppRouting:
server = FastMCP("Platform")
server.add_provider(app, namespace="crm")
result = await server.call_tool("crm___save_contact", {"name": "alice"})
result = await server.call_tool(
hashed_backend_name("crm", "save_contact"), {"name": "alice"}
)
assert result.content[0].text == "saved alice" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_namespaced_name_also_works(self):
"""Namespaced tool name works through normal resolution."""
async def test_namespaced_display_name_also_works(self):
"""Model-visible tools still resolve through Namespace as before."""
app = FastMCPApp("crm")
@app.tool(model=True)
@ -547,10 +523,11 @@ class TestCallToolAppRouting:
result = await server.call_tool("crm_save_contact", {"name": "bob"})
assert result.content[0].text == "saved bob" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_app_name_auth_blocks_unauthorized(self):
"""Auth checks run even when routing via app_name."""
async def test_hashed_name_auth_blocks_unauthorized(self):
"""Auth checks run on the hashed-name dispatch path too."""
from fastmcp.exceptions import NotFoundError
from fastmcp.server.context import _current_transport
from fastmcp.server.providers.addressing import hashed_backend_name
app = FastMCPApp("test")
deny_all = AsyncMock(return_value=False)
@ -565,12 +542,16 @@ class TestCallToolAppRouting:
token = _current_transport.set("streamable-http")
try:
with pytest.raises(NotFoundError):
await server.call_tool("test___secret", {})
await server.call_tool(hashed_backend_name("test", "secret"), {})
finally:
_current_transport.reset(token)
async def test_two_apps_same_tool_name_routed_correctly(self):
"""Two apps with same tool name disambiguated by app_name."""
async def test_two_apps_same_tool_name_routed_by_address(self):
"""Two FastMCPApps each with a `save` tool live at distinct
addresses, so they hash differently and the dispatcher routes
each call to the right app without name collisions."""
from fastmcp.server.providers.addressing import hashed_backend_name
contacts = FastMCPApp("contacts")
billing = FastMCPApp("billing")
@ -583,34 +564,19 @@ class TestCallToolAppRouting:
return f"invoice: {amount}"
server = FastMCP("Platform")
server.add_provider(contacts)
server.add_provider(billing)
server.add_provider(contacts) # → address (0,)
server.add_provider(billing) # → address (1,)
r1 = await server.call_tool("contacts___save", {"name": "alice"})
r2 = await server.call_tool("billing___save", {"amount": "100"})
r1 = await server.call_tool(
hashed_backend_name("contacts", "save"), {"name": "alice"}
)
r2 = await server.call_tool(
hashed_backend_name("billing", "save"), {"amount": "100"}
)
assert r1.content[0].text == "contact: alice" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert r2.content[0].text == "invoice: 100" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_deeply_nested_app(self):
"""App tool is found even through multiple levels of nesting."""
app = FastMCPApp("deep")
@app.tool()
def hidden(x: str) -> str:
return x
inner = FastMCP("Inner")
inner.add_provider(app, namespace="app")
outer = FastMCP("Outer")
outer.mount(inner, namespace="inner")
# Normal resolution: would need "inner_app_hidden"
# App routing: bypasses all transforms
result = await outer.call_tool("deep___hidden", {"x": "found"})
assert result.content[0].text == "found" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
# ---------------------------------------------------------------------------
# App-only tool filtering from server list_tools / get_tool
@ -679,8 +645,12 @@ class TestAppOnlyToolFiltering:
names = [t.name for t in tools]
assert "save" not in names
# But still callable via app_name routing
result = await server.call_tool("contacts___save", {"name": "alice"})
# But still callable via the hashed-address routing path.
from fastmcp.server.providers.addressing import hashed_backend_name
result = await server.call_tool(
hashed_backend_name("contacts", "save"), {"name": "alice"}
)
assert result.content[0].text == "saved alice" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_app_only_tool_hidden_from_get_tool(self):
@ -831,8 +801,15 @@ class TestComposition:
server.add_provider(crm, namespace="crm")
server.add_provider(billing, namespace="billing")
r1 = await server.call_tool("CRM___save_contact", {"name": "alice"})
r2 = await server.call_tool("Billing___create_invoice", {"amount": 100})
from fastmcp.server.providers.addressing import hashed_backend_name
# CRM is at address (0,), billing at (1,) — registration order.
r1 = await server.call_tool(
hashed_backend_name("CRM", "save_contact"), {"name": "alice"}
)
r2 = await server.call_tool(
hashed_backend_name("Billing", "create_invoice"), {"amount": 100}
)
assert r1.content[0].text == "alice" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert r2.content[0].text == "100" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
@ -853,16 +830,21 @@ class TestComposition:
names = {t.name for t in tools}
assert names == {"dashboard", "save"}
async def test_ui_registers_prefab_renderer_resource(self):
async def test_ui_synthesizes_per_tool_renderer_resource(self):
"""Each @app.ui() tool gets its own renderer resource synthesized
on demand from the server's address registry."""
app = FastMCPApp("test")
@app.ui()
def dashboard() -> str:
return "ui"
resources = await app._list_resources()
uris = [str(r.uri) for r in resources]
assert any("ui://prefab/renderer.html" in uri for uri in uris)
server = FastMCP("Platform")
server.add_provider(app)
resources = list(await server.list_resources())
prefab = [r for r in resources if "prefab/tool" in str(r.uri)]
assert len(prefab) == 1
# ---------------------------------------------------------------------------
@ -874,7 +856,9 @@ class TestAppIntegration:
async def test_full_app_lifecycle_through_client(self):
"""End-to-end: mount an app on a namespaced server, call UI tool
through a client (verifying structured_content is returned), then
call the backend tool via the ___-prefixed name."""
call the backend tool via its hashed-address name."""
from fastmcp.server.providers.addressing import hashed_backend_name
app = FastMCPApp("contacts")
@app.ui()
@ -901,10 +885,10 @@ class TestAppIntegration:
sc = result.structuredContent
assert sc is not None
# Call the backend tool via prefixed name
# (bypasses namespace transforms and visibility filtering)
# Call the backend tool via its hashed address — bypasses namespace
# transforms and visibility filtering by going through the registry.
backend_result = await server.call_tool(
"contacts___save_contact",
hashed_backend_name("contacts", "save_contact"),
{"name": "Alice", "email": "alice@example.com"},
)
result_text = backend_result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]