mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Move app modules to fastmcp.apps package (#3616)
This commit is contained in:
parent
c04c9d0ce5
commit
1eabe7f74a
17 changed files with 630 additions and 568 deletions
|
|
@ -10,7 +10,7 @@ from fastmcp.utilities.logging import configure_logging as _configure_logging
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.client import Client as Client
|
||||
from fastmcp.server.app import FastMCPApp as FastMCPApp
|
||||
from fastmcp.apps.app import FastMCPApp as FastMCPApp
|
||||
|
||||
settings = Settings()
|
||||
if settings.log_enabled:
|
||||
|
|
@ -42,7 +42,7 @@ def __getattr__(name: str) -> object:
|
|||
|
||||
return Client
|
||||
if name == "FastMCPApp":
|
||||
from fastmcp.server.app import FastMCPApp
|
||||
from fastmcp.apps.app import FastMCPApp
|
||||
|
||||
return FastMCPApp
|
||||
if name == "client":
|
||||
|
|
|
|||
17
src/fastmcp/apps/__init__.py
Normal file
17
src/fastmcp/apps/__init__.py
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
"""FastMCP Apps — interactive UIs for MCP tools.
|
||||
|
||||
This package contains the app-related components:
|
||||
|
||||
- ``FastMCPApp`` — composable provider for interactive apps with backend tools
|
||||
- ``AppConfig`` — configuration for MCP App tools and resources
|
||||
- ``ResourceCSP`` / ``ResourcePermissions`` — security configuration
|
||||
"""
|
||||
|
||||
from fastmcp.apps.app import FastMCPApp as FastMCPApp
|
||||
from fastmcp.apps.config import AppConfig as AppConfig
|
||||
from fastmcp.apps.config import ResourceCSP as ResourceCSP
|
||||
from fastmcp.apps.config import ResourcePermissions as ResourcePermissions
|
||||
from fastmcp.apps.config import UI_EXTENSION_ID as UI_EXTENSION_ID
|
||||
from fastmcp.apps.config import app_config_to_meta_dict as app_config_to_meta_dict
|
||||
from fastmcp.utilities.mime import UI_MIME_TYPE as UI_MIME_TYPE
|
||||
from fastmcp.utilities.mime import resolve_ui_mime_type as resolve_ui_mime_type
|
||||
415
src/fastmcp/apps/app.py
Normal file
415
src/fastmcp/apps/app.py
Normal file
|
|
@ -0,0 +1,415 @@
|
|||
"""FastMCPApp — a Provider that represents a composable MCP application.
|
||||
|
||||
FastMCPApp binds entry-point tools (model calls these) together with backend
|
||||
tools (the UI calls these via CallTool). Backend tools are tagged with
|
||||
``meta["fastmcp"]["app"]`` so they can be found through the provider chain
|
||||
even when transforms (namespace, visibility, etc.) have renamed or hidden
|
||||
them — the server sets a context var that tells ``Provider.get_tool`` to
|
||||
fall back to a direct lookup for app-visible tools.
|
||||
|
||||
Usage::
|
||||
|
||||
from fastmcp import FastMCP, FastMCPApp
|
||||
|
||||
app = FastMCPApp("Dashboard")
|
||||
|
||||
@app.ui()
|
||||
def show_dashboard() -> Component:
|
||||
return Column(...)
|
||||
|
||||
@app.tool()
|
||||
def save_contact(name: str, email: str) -> dict:
|
||||
return {"name": name, "email": email}
|
||||
|
||||
server = FastMCP("Platform")
|
||||
server.add_provider(app)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from collections.abc import AsyncIterator, Callable, Sequence
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from typing import Any, Literal, TypeVar, overload
|
||||
|
||||
from mcp.types import AnyFunction, Icon, ToolAnnotations
|
||||
|
||||
from fastmcp.server.auth.authorization import AuthCheck
|
||||
from fastmcp.server.providers.base import Provider
|
||||
from fastmcp.server.providers.local_provider import LocalProvider
|
||||
from fastmcp.tools.base import Tool
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
F = TypeVar("F", bound=Callable[..., Any])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CallTool resolver
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _resolve_tool_ref(fn: Any) -> Any:
|
||||
"""Resolve a callable or string to a ``ResolvedTool`` for CallTool serialization.
|
||||
|
||||
For strings, passes them through as-is — the server resolves them at
|
||||
call time using ``_meta.fastmcp.app``.
|
||||
|
||||
For callables, extracts the tool name from ``__fastmcp__`` metadata
|
||||
or ``__name__``.
|
||||
"""
|
||||
from prefab_ui.app import ResolvedTool
|
||||
|
||||
if isinstance(fn, str):
|
||||
return ResolvedTool(name=fn)
|
||||
|
||||
fmeta: Any = None
|
||||
try:
|
||||
from fastmcp.decorators import get_fastmcp_meta
|
||||
|
||||
fmeta = get_fastmcp_meta(fn)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if fmeta is not None:
|
||||
name: str | None = getattr(fmeta, "name", None)
|
||||
if name is not None:
|
||||
return ResolvedTool(name=name)
|
||||
|
||||
fn_name = getattr(fn, "__name__", None)
|
||||
if fn_name is not None:
|
||||
return ResolvedTool(name=fn_name)
|
||||
|
||||
raise ValueError(f"Cannot resolve tool reference: {fn!r}")
|
||||
|
||||
|
||||
def _dispatch_decorator(
|
||||
name_or_fn: str | AnyFunction | None,
|
||||
name: str | None,
|
||||
register: Callable[[Any, str | None], Any],
|
||||
decorator_name: str,
|
||||
) -> Any:
|
||||
"""Shared dispatch logic for @app.tool() and @app.ui() calling patterns."""
|
||||
if inspect.isroutine(name_or_fn):
|
||||
return register(name_or_fn, name)
|
||||
|
||||
if isinstance(name_or_fn, str):
|
||||
if name is not None:
|
||||
raise TypeError(
|
||||
"Cannot specify both a name as first argument and as keyword argument."
|
||||
)
|
||||
tool_name: str | None = name_or_fn
|
||||
elif name_or_fn is None:
|
||||
tool_name = name
|
||||
else:
|
||||
raise TypeError(
|
||||
f"First argument to @{decorator_name} must be a function, string, or None, "
|
||||
f"got {type(name_or_fn)}"
|
||||
)
|
||||
|
||||
def decorator(fn: F) -> F:
|
||||
return register(fn, tool_name)
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FastMCPApp
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FastMCPApp(Provider):
|
||||
"""A Provider that represents an MCP application.
|
||||
|
||||
Binds together entry-point tools (``@app.ui``), backend tools
|
||||
(``@app.tool``), and the Prefab renderer resource. Backend tools
|
||||
are tagged with ``meta["fastmcp"]["app"]`` so ``Provider.get_tool``
|
||||
can find them by original name even when transforms have been applied.
|
||||
"""
|
||||
|
||||
def __init__(self, name: str) -> None:
|
||||
super().__init__()
|
||||
self.name = name
|
||||
self._local = LocalProvider(on_duplicate="error")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"FastMCPApp({self.name!r})"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# @app.tool() — backend tools called by the UI
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@overload
|
||||
def tool(
|
||||
self,
|
||||
name_or_fn: F,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
model: bool = False,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> F: ...
|
||||
|
||||
@overload
|
||||
def tool(
|
||||
self,
|
||||
name_or_fn: str | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
model: bool = False,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Callable[[F], F]: ...
|
||||
|
||||
def tool(
|
||||
self,
|
||||
name_or_fn: str | AnyFunction | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
model: bool = False,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Any:
|
||||
"""Register a backend tool that the UI calls via CallTool.
|
||||
|
||||
Backend tools default to ``visibility=["app"]``. Pass ``model=True``
|
||||
to also expose the tool to the model (``visibility=["app", "model"]``).
|
||||
|
||||
Supports multiple calling patterns::
|
||||
|
||||
@app.tool
|
||||
def save(name: str): ...
|
||||
|
||||
@app.tool()
|
||||
def save(name: str): ...
|
||||
|
||||
@app.tool("custom_name")
|
||||
def save(name: str): ...
|
||||
"""
|
||||
visibility: list[Literal["app", "model"]] = (
|
||||
["app", "model"] if model else ["app"]
|
||||
)
|
||||
|
||||
def _register(fn: F, tool_name: str | None) -> F:
|
||||
resolved_name = tool_name or getattr(fn, "__name__", None)
|
||||
if resolved_name is None:
|
||||
raise ValueError(f"Cannot determine tool name for {fn!r}")
|
||||
|
||||
from fastmcp.apps.config import AppConfig, app_config_to_meta_dict
|
||||
|
||||
app_config = AppConfig(visibility=visibility)
|
||||
meta: dict[str, Any] = {
|
||||
"ui": app_config_to_meta_dict(app_config),
|
||||
"fastmcp": {"app": self.name},
|
||||
}
|
||||
|
||||
tool_obj = Tool.from_function(
|
||||
fn,
|
||||
name=resolved_name,
|
||||
description=description,
|
||||
meta=meta,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
self._local._add_component(tool_obj)
|
||||
return fn
|
||||
|
||||
return _dispatch_decorator(name_or_fn, name, _register, "tool")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# @app.ui() — entry-point tools the model calls to open the app
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@overload
|
||||
def ui(
|
||||
self,
|
||||
name_or_fn: F,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
title: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> F: ...
|
||||
|
||||
@overload
|
||||
def ui(
|
||||
self,
|
||||
name_or_fn: str | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
title: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Callable[[F], F]: ...
|
||||
|
||||
def ui(
|
||||
self,
|
||||
name_or_fn: str | AnyFunction | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
title: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Any:
|
||||
"""Register a UI entry-point tool that the model calls.
|
||||
|
||||
Entry-point tools default to ``visibility=["model"]`` and auto-wire
|
||||
the Prefab renderer resource and CSP. They are tagged with the app
|
||||
name so structured content includes ``_meta.fastmcp.app``.
|
||||
|
||||
Supports multiple calling patterns::
|
||||
|
||||
@app.ui
|
||||
def dashboard() -> Component: ...
|
||||
|
||||
@app.ui()
|
||||
def dashboard() -> Component: ...
|
||||
|
||||
@app.ui("my_dashboard")
|
||||
def dashboard() -> Component: ...
|
||||
"""
|
||||
|
||||
def _register(fn: F, tool_name: str | None) -> F:
|
||||
from fastmcp.apps.config import AppConfig, app_config_to_meta_dict
|
||||
from fastmcp.server.providers.local_provider.decorators.tools import (
|
||||
PREFAB_RENDERER_URI,
|
||||
_ensure_prefab_renderer,
|
||||
)
|
||||
|
||||
try:
|
||||
from prefab_ui.renderer import get_renderer_csp
|
||||
|
||||
from fastmcp.apps.config import ResourceCSP
|
||||
|
||||
csp = get_renderer_csp()
|
||||
app_config = AppConfig(
|
||||
resource_uri=PREFAB_RENDERER_URI,
|
||||
visibility=["model"],
|
||||
csp=ResourceCSP(
|
||||
resource_domains=csp.get("resource_domains"),
|
||||
connect_domains=csp.get("connect_domains"),
|
||||
),
|
||||
)
|
||||
except ImportError:
|
||||
app_config = AppConfig(
|
||||
resource_uri=PREFAB_RENDERER_URI,
|
||||
visibility=["model"],
|
||||
)
|
||||
|
||||
meta: dict[str, Any] = {
|
||||
"ui": app_config_to_meta_dict(app_config),
|
||||
"fastmcp": {"app": self.name},
|
||||
}
|
||||
|
||||
tool_obj = Tool.from_function(
|
||||
fn,
|
||||
name=tool_name,
|
||||
description=description,
|
||||
title=title,
|
||||
tags=tags,
|
||||
icons=icons,
|
||||
annotations=annotations,
|
||||
meta=meta,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
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")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Programmatic tool addition
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def add_tool(
|
||||
self,
|
||||
tool: Tool | Callable[..., Any],
|
||||
) -> Tool:
|
||||
"""Add a tool to this app programmatically.
|
||||
|
||||
The tool is tagged with this app's name for routing.
|
||||
"""
|
||||
if not isinstance(tool, Tool):
|
||||
tool = Tool._ensure_tool(tool)
|
||||
|
||||
# Tag with app name and visibility for routing
|
||||
meta = dict(tool.meta) if tool.meta else {}
|
||||
meta.setdefault("fastmcp", {})["app"] = self.name
|
||||
ui = meta.setdefault("ui", {})
|
||||
if "visibility" not in ui:
|
||||
ui["visibility"] = ["app"]
|
||||
tool.meta = meta
|
||||
|
||||
self._local._add_component(tool)
|
||||
return tool
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Provider interface — delegate to internal LocalProvider
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _list_tools(self) -> Sequence[Tool]:
|
||||
return await self._local._list_tools()
|
||||
|
||||
async def _get_tool(self, name: str, version: Any = None) -> Tool | None:
|
||||
return await self._local._get_tool(name, version)
|
||||
|
||||
async def _list_resources(self) -> Sequence[Any]:
|
||||
return await self._local._list_resources()
|
||||
|
||||
async def _get_resource(self, uri: str, version: Any = None) -> Any | None:
|
||||
return await self._local._get_resource(uri, version)
|
||||
|
||||
async def _list_resource_templates(self) -> Sequence[Any]:
|
||||
return await self._local._list_resource_templates()
|
||||
|
||||
async def _get_resource_template(self, uri: str, version: Any = None) -> Any | None:
|
||||
return await self._local._get_resource_template(uri, version)
|
||||
|
||||
async def _list_prompts(self) -> Sequence[Any]:
|
||||
return await self._local._list_prompts()
|
||||
|
||||
async def _get_prompt(self, name: str, version: Any = None) -> Any | None:
|
||||
return await self._local._get_prompt(name, version)
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(self) -> AsyncIterator[None]:
|
||||
async with self._local.lifespan():
|
||||
yield
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Convenience runner
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run(
|
||||
self,
|
||||
transport: Literal["stdio", "http", "sse", "streamable-http"] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""Create a temporary FastMCP server and run this app standalone."""
|
||||
from fastmcp.server.server import FastMCP
|
||||
|
||||
server = FastMCP(self.name)
|
||||
server.add_provider(self)
|
||||
server.run(transport=transport, **kwargs)
|
||||
121
src/fastmcp/apps/config.py
Normal file
121
src/fastmcp/apps/config.py
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
"""MCP Apps support — extension negotiation and typed UI metadata models.
|
||||
|
||||
Provides constants and Pydantic models for the MCP Apps extension
|
||||
(io.modelcontextprotocol/ui), enabling tools and resources to carry
|
||||
UI metadata for clients that support interactive app rendering.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from fastmcp.utilities.mime import UI_MIME_TYPE as UI_MIME_TYPE
|
||||
from fastmcp.utilities.mime import resolve_ui_mime_type as resolve_ui_mime_type
|
||||
|
||||
UI_EXTENSION_ID = "io.modelcontextprotocol/ui"
|
||||
|
||||
|
||||
class ResourceCSP(BaseModel):
|
||||
"""Content Security Policy for MCP App resources.
|
||||
|
||||
Declares which external origins the app is allowed to connect to or
|
||||
load resources from. Hosts use these declarations to build the
|
||||
``Content-Security-Policy`` header for the sandboxed iframe.
|
||||
"""
|
||||
|
||||
connect_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="connectDomains",
|
||||
description="Origins allowed for fetch/XHR/WebSocket (connect-src)",
|
||||
)
|
||||
resource_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="resourceDomains",
|
||||
description="Origins allowed for scripts, images, styles, fonts (script-src etc.)",
|
||||
)
|
||||
frame_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="frameDomains",
|
||||
description="Origins allowed for nested iframes (frame-src)",
|
||||
)
|
||||
base_uri_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="baseUriDomains",
|
||||
description="Allowed base URIs for the document (base-uri)",
|
||||
)
|
||||
|
||||
model_config = {"populate_by_name": True, "extra": "allow"}
|
||||
|
||||
|
||||
class ResourcePermissions(BaseModel):
|
||||
"""Iframe sandbox permissions for MCP App resources.
|
||||
|
||||
Each field, when set (typically to ``{}``), requests that the host
|
||||
grant the corresponding Permission Policy feature to the sandboxed
|
||||
iframe. Hosts MAY honour these; apps should use JS feature detection
|
||||
as a fallback.
|
||||
"""
|
||||
|
||||
camera: dict[str, Any] | None = Field(
|
||||
default=None, description="Request camera access"
|
||||
)
|
||||
microphone: dict[str, Any] | None = Field(
|
||||
default=None, description="Request microphone access"
|
||||
)
|
||||
geolocation: dict[str, Any] | None = Field(
|
||||
default=None, description="Request geolocation access"
|
||||
)
|
||||
clipboard_write: dict[str, Any] | None = Field(
|
||||
default=None,
|
||||
alias="clipboardWrite",
|
||||
description="Request clipboard-write access",
|
||||
)
|
||||
|
||||
model_config = {"populate_by_name": True, "extra": "allow"}
|
||||
|
||||
|
||||
class AppConfig(BaseModel):
|
||||
"""Configuration for MCP App tools and resources.
|
||||
|
||||
Controls how a tool or resource participates in the MCP Apps extension.
|
||||
On tools, ``resource_uri`` and ``visibility`` specify which UI resource
|
||||
to render and where the tool appears. On resources, those fields must
|
||||
be left unset (the resource itself is the UI).
|
||||
|
||||
All fields use ``exclude_none`` serialization so only explicitly-set
|
||||
values appear on the wire. Aliases match the MCP Apps wire format
|
||||
(camelCase).
|
||||
"""
|
||||
|
||||
resource_uri: str | None = Field(
|
||||
default=None,
|
||||
alias="resourceUri",
|
||||
description="URI of the UI resource (typically ui:// scheme). Tools only.",
|
||||
)
|
||||
visibility: list[Literal["app", "model"]] | None = Field(
|
||||
default=None,
|
||||
description="Where this tool is visible: 'app', 'model', or both. Tools only.",
|
||||
)
|
||||
csp: ResourceCSP | None = Field(
|
||||
default=None, description="Content Security Policy for the app iframe"
|
||||
)
|
||||
permissions: ResourcePermissions | None = Field(
|
||||
default=None, description="Iframe sandbox permissions"
|
||||
)
|
||||
domain: str | None = Field(default=None, description="Domain for the iframe")
|
||||
prefers_border: bool | None = Field(
|
||||
default=None,
|
||||
alias="prefersBorder",
|
||||
description="Whether the UI prefers a visible border",
|
||||
)
|
||||
|
||||
model_config = {"populate_by_name": True, "extra": "allow"}
|
||||
|
||||
|
||||
def app_config_to_meta_dict(app: AppConfig | dict[str, Any]) -> dict[str, Any]:
|
||||
"""Convert an AppConfig or dict to the wire-format dict for ``meta["ui"]``."""
|
||||
if isinstance(app, AppConfig):
|
||||
return app.model_dump(by_alias=True, exclude_none=True)
|
||||
return app
|
||||
|
|
@ -16,7 +16,6 @@ from pydantic.json_schema import SkipJsonSchema
|
|||
import fastmcp
|
||||
from fastmcp.decorators import resolve_task_config
|
||||
from fastmcp.resources.base import Resource, ResourceResult
|
||||
from fastmcp.server.apps import resolve_ui_mime_type
|
||||
from fastmcp.server.auth.authorization import AuthCheck
|
||||
from fastmcp.server.dependencies import (
|
||||
transform_context_annotations,
|
||||
|
|
@ -27,6 +26,7 @@ from fastmcp.utilities.async_utils import (
|
|||
call_sync_fn_in_threadpool,
|
||||
is_coroutine_function,
|
||||
)
|
||||
from fastmcp.utilities.mime import resolve_ui_mime_type
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from docket import Docket
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ from pydantic import (
|
|||
)
|
||||
|
||||
from fastmcp.resources.base import Resource, ResourceResult
|
||||
from fastmcp.server.apps import resolve_ui_mime_type
|
||||
from fastmcp.server.auth.authorization import AuthCheck
|
||||
from fastmcp.server.dependencies import (
|
||||
transform_context_annotations,
|
||||
|
|
@ -33,6 +32,7 @@ from fastmcp.server.dependencies import (
|
|||
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
|
||||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.json_schema import compress_schema
|
||||
from fastmcp.utilities.mime import resolve_ui_mime_type
|
||||
from fastmcp.utilities.types import get_cached_typeadapter
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,415 +1,18 @@
|
|||
"""FastMCPApp — a Provider that represents a composable MCP application.
|
||||
"""Backward-compatible re-exports from fastmcp.apps.app.
|
||||
|
||||
FastMCPApp binds entry-point tools (model calls these) together with backend
|
||||
tools (the UI calls these via CallTool). Backend tools are tagged with
|
||||
``meta["fastmcp"]["app"]`` so they can be found through the provider chain
|
||||
even when transforms (namespace, visibility, etc.) have renamed or hidden
|
||||
them — the server sets a context var that tells ``Provider.get_tool`` to
|
||||
fall back to a direct lookup for app-visible tools.
|
||||
|
||||
Usage::
|
||||
|
||||
from fastmcp import FastMCP, FastMCPApp
|
||||
|
||||
app = FastMCPApp("Dashboard")
|
||||
|
||||
@app.ui()
|
||||
def show_dashboard() -> Component:
|
||||
return Column(...)
|
||||
|
||||
@app.tool()
|
||||
def save_contact(name: str, email: str) -> dict:
|
||||
return {"name": name, "email": email}
|
||||
|
||||
server = FastMCP("Platform")
|
||||
server.add_provider(app)
|
||||
.. deprecated:: 3.2.0
|
||||
Import from ``fastmcp.apps.app`` or ``fastmcp`` instead.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from collections.abc import AsyncIterator, Callable, Sequence
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from typing import Any, Literal, TypeVar, overload
|
||||
|
||||
from mcp.types import AnyFunction, Icon, ToolAnnotations
|
||||
|
||||
from fastmcp.server.auth.authorization import AuthCheck
|
||||
from fastmcp.server.providers.base import Provider
|
||||
from fastmcp.server.providers.local_provider import LocalProvider
|
||||
from fastmcp.tools.base import Tool
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
F = TypeVar("F", bound=Callable[..., Any])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CallTool resolver
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _resolve_tool_ref(fn: Any) -> Any:
|
||||
"""Resolve a callable or string to a ``ResolvedTool`` for CallTool serialization.
|
||||
|
||||
For strings, passes them through as-is — the server resolves them at
|
||||
call time using ``_meta.fastmcp.app``.
|
||||
|
||||
For callables, extracts the tool name from ``__fastmcp__`` metadata
|
||||
or ``__name__``.
|
||||
"""
|
||||
from prefab_ui.app import ResolvedTool
|
||||
|
||||
if isinstance(fn, str):
|
||||
return ResolvedTool(name=fn)
|
||||
|
||||
fmeta: Any = None
|
||||
try:
|
||||
from fastmcp.decorators import get_fastmcp_meta
|
||||
|
||||
fmeta = get_fastmcp_meta(fn)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if fmeta is not None:
|
||||
name: str | None = getattr(fmeta, "name", None)
|
||||
if name is not None:
|
||||
return ResolvedTool(name=name)
|
||||
|
||||
fn_name = getattr(fn, "__name__", None)
|
||||
if fn_name is not None:
|
||||
return ResolvedTool(name=fn_name)
|
||||
|
||||
raise ValueError(f"Cannot resolve tool reference: {fn!r}")
|
||||
|
||||
|
||||
def _dispatch_decorator(
|
||||
name_or_fn: str | AnyFunction | None,
|
||||
name: str | None,
|
||||
register: Callable[[Any, str | None], Any],
|
||||
decorator_name: str,
|
||||
) -> Any:
|
||||
"""Shared dispatch logic for @app.tool() and @app.ui() calling patterns."""
|
||||
if inspect.isroutine(name_or_fn):
|
||||
return register(name_or_fn, name)
|
||||
|
||||
if isinstance(name_or_fn, str):
|
||||
if name is not None:
|
||||
raise TypeError(
|
||||
"Cannot specify both a name as first argument and as keyword argument."
|
||||
)
|
||||
tool_name: str | None = name_or_fn
|
||||
elif name_or_fn is None:
|
||||
tool_name = name
|
||||
else:
|
||||
raise TypeError(
|
||||
f"First argument to @{decorator_name} must be a function, string, or None, "
|
||||
f"got {type(name_or_fn)}"
|
||||
)
|
||||
|
||||
def decorator(fn: F) -> F:
|
||||
return register(fn, tool_name)
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FastMCPApp
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FastMCPApp(Provider):
|
||||
"""A Provider that represents an MCP application.
|
||||
|
||||
Binds together entry-point tools (``@app.ui``), backend tools
|
||||
(``@app.tool``), and the Prefab renderer resource. Backend tools
|
||||
are tagged with ``meta["fastmcp"]["app"]`` so ``Provider.get_tool``
|
||||
can find them by original name even when transforms have been applied.
|
||||
"""
|
||||
|
||||
def __init__(self, name: str) -> None:
|
||||
super().__init__()
|
||||
self.name = name
|
||||
self._local = LocalProvider(on_duplicate="error")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"FastMCPApp({self.name!r})"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# @app.tool() — backend tools called by the UI
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@overload
|
||||
def tool(
|
||||
self,
|
||||
name_or_fn: F,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
model: bool = False,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> F: ...
|
||||
|
||||
@overload
|
||||
def tool(
|
||||
self,
|
||||
name_or_fn: str | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
model: bool = False,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Callable[[F], F]: ...
|
||||
|
||||
def tool(
|
||||
self,
|
||||
name_or_fn: str | AnyFunction | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
model: bool = False,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Any:
|
||||
"""Register a backend tool that the UI calls via CallTool.
|
||||
|
||||
Backend tools default to ``visibility=["app"]``. Pass ``model=True``
|
||||
to also expose the tool to the model (``visibility=["app", "model"]``).
|
||||
|
||||
Supports multiple calling patterns::
|
||||
|
||||
@app.tool
|
||||
def save(name: str): ...
|
||||
|
||||
@app.tool()
|
||||
def save(name: str): ...
|
||||
|
||||
@app.tool("custom_name")
|
||||
def save(name: str): ...
|
||||
"""
|
||||
visibility: list[Literal["app", "model"]] = (
|
||||
["app", "model"] if model else ["app"]
|
||||
)
|
||||
|
||||
def _register(fn: F, tool_name: str | None) -> F:
|
||||
resolved_name = tool_name or getattr(fn, "__name__", None)
|
||||
if resolved_name is None:
|
||||
raise ValueError(f"Cannot determine tool name for {fn!r}")
|
||||
|
||||
from fastmcp.server.apps import AppConfig, app_config_to_meta_dict
|
||||
|
||||
app_config = AppConfig(visibility=visibility)
|
||||
meta: dict[str, Any] = {
|
||||
"ui": app_config_to_meta_dict(app_config),
|
||||
"fastmcp": {"app": self.name},
|
||||
}
|
||||
|
||||
tool_obj = Tool.from_function(
|
||||
fn,
|
||||
name=resolved_name,
|
||||
description=description,
|
||||
meta=meta,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
self._local._add_component(tool_obj)
|
||||
return fn
|
||||
|
||||
return _dispatch_decorator(name_or_fn, name, _register, "tool")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# @app.ui() — entry-point tools the model calls to open the app
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@overload
|
||||
def ui(
|
||||
self,
|
||||
name_or_fn: F,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
title: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> F: ...
|
||||
|
||||
@overload
|
||||
def ui(
|
||||
self,
|
||||
name_or_fn: str | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
title: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Callable[[F], F]: ...
|
||||
|
||||
def ui(
|
||||
self,
|
||||
name_or_fn: str | AnyFunction | None = None,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
title: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> Any:
|
||||
"""Register a UI entry-point tool that the model calls.
|
||||
|
||||
Entry-point tools default to ``visibility=["model"]`` and auto-wire
|
||||
the Prefab renderer resource and CSP. They are tagged with the app
|
||||
name so structured content includes ``_meta.fastmcp.app``.
|
||||
|
||||
Supports multiple calling patterns::
|
||||
|
||||
@app.ui
|
||||
def dashboard() -> Component: ...
|
||||
|
||||
@app.ui()
|
||||
def dashboard() -> Component: ...
|
||||
|
||||
@app.ui("my_dashboard")
|
||||
def dashboard() -> Component: ...
|
||||
"""
|
||||
|
||||
def _register(fn: F, tool_name: str | None) -> F:
|
||||
from fastmcp.server.apps import AppConfig, app_config_to_meta_dict
|
||||
from fastmcp.server.providers.local_provider.decorators.tools import (
|
||||
PREFAB_RENDERER_URI,
|
||||
_ensure_prefab_renderer,
|
||||
)
|
||||
|
||||
try:
|
||||
from prefab_ui.renderer import get_renderer_csp
|
||||
|
||||
from fastmcp.server.apps import ResourceCSP
|
||||
|
||||
csp = get_renderer_csp()
|
||||
app_config = AppConfig(
|
||||
resource_uri=PREFAB_RENDERER_URI,
|
||||
visibility=["model"],
|
||||
csp=ResourceCSP(
|
||||
resource_domains=csp.get("resource_domains"),
|
||||
connect_domains=csp.get("connect_domains"),
|
||||
),
|
||||
)
|
||||
except ImportError:
|
||||
app_config = AppConfig(
|
||||
resource_uri=PREFAB_RENDERER_URI,
|
||||
visibility=["model"],
|
||||
)
|
||||
|
||||
meta: dict[str, Any] = {
|
||||
"ui": app_config_to_meta_dict(app_config),
|
||||
"fastmcp": {"app": self.name},
|
||||
}
|
||||
|
||||
tool_obj = Tool.from_function(
|
||||
fn,
|
||||
name=tool_name,
|
||||
description=description,
|
||||
title=title,
|
||||
tags=tags,
|
||||
icons=icons,
|
||||
annotations=annotations,
|
||||
meta=meta,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
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")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Programmatic tool addition
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def add_tool(
|
||||
self,
|
||||
tool: Tool | Callable[..., Any],
|
||||
) -> Tool:
|
||||
"""Add a tool to this app programmatically.
|
||||
|
||||
The tool is tagged with this app's name for routing.
|
||||
"""
|
||||
if not isinstance(tool, Tool):
|
||||
tool = Tool._ensure_tool(tool)
|
||||
|
||||
# Tag with app name and visibility for routing
|
||||
meta = dict(tool.meta) if tool.meta else {}
|
||||
meta.setdefault("fastmcp", {})["app"] = self.name
|
||||
ui = meta.setdefault("ui", {})
|
||||
if "visibility" not in ui:
|
||||
ui["visibility"] = ["app"]
|
||||
tool.meta = meta
|
||||
|
||||
self._local._add_component(tool)
|
||||
return tool
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Provider interface — delegate to internal LocalProvider
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _list_tools(self) -> Sequence[Tool]:
|
||||
return await self._local._list_tools()
|
||||
|
||||
async def _get_tool(self, name: str, version: Any = None) -> Tool | None:
|
||||
return await self._local._get_tool(name, version)
|
||||
|
||||
async def _list_resources(self) -> Sequence[Any]:
|
||||
return await self._local._list_resources()
|
||||
|
||||
async def _get_resource(self, uri: str, version: Any = None) -> Any | None:
|
||||
return await self._local._get_resource(uri, version)
|
||||
|
||||
async def _list_resource_templates(self) -> Sequence[Any]:
|
||||
return await self._local._list_resource_templates()
|
||||
|
||||
async def _get_resource_template(self, uri: str, version: Any = None) -> Any | None:
|
||||
return await self._local._get_resource_template(uri, version)
|
||||
|
||||
async def _list_prompts(self) -> Sequence[Any]:
|
||||
return await self._local._list_prompts()
|
||||
|
||||
async def _get_prompt(self, name: str, version: Any = None) -> Any | None:
|
||||
return await self._local._get_prompt(name, version)
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(self) -> AsyncIterator[None]:
|
||||
async with self._local.lifespan():
|
||||
yield
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Convenience runner
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run(
|
||||
self,
|
||||
transport: Literal["stdio", "http", "sse", "streamable-http"] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""Create a temporary FastMCP server and run this app standalone."""
|
||||
from fastmcp.server.server import FastMCP
|
||||
|
||||
server = FastMCP(self.name)
|
||||
server.add_provider(self)
|
||||
server.run(transport=transport, **kwargs)
|
||||
import warnings
|
||||
|
||||
from fastmcp.apps.app import FastMCPApp as FastMCPApp
|
||||
from fastmcp.apps.app import _dispatch_decorator as _dispatch_decorator
|
||||
from fastmcp.apps.app import _resolve_tool_ref as _resolve_tool_ref
|
||||
|
||||
warnings.warn(
|
||||
"'fastmcp.server.app' is deprecated. "
|
||||
"Use 'fastmcp.apps.app' or 'from fastmcp import FastMCPApp' instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,142 +1,21 @@
|
|||
"""MCP Apps support — extension negotiation and typed UI metadata models.
|
||||
"""Backward-compatible re-exports from fastmcp.apps.
|
||||
|
||||
Provides constants and Pydantic models for the MCP Apps extension
|
||||
(io.modelcontextprotocol/ui), enabling tools and resources to carry
|
||||
UI metadata for clients that support interactive app rendering.
|
||||
.. deprecated:: 3.2.0
|
||||
Import from ``fastmcp.apps`` instead.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
import warnings
|
||||
|
||||
from typing import Any, Literal
|
||||
from fastmcp.apps.config import UI_EXTENSION_ID as UI_EXTENSION_ID
|
||||
from fastmcp.apps.config import AppConfig as AppConfig
|
||||
from fastmcp.apps.config import ResourceCSP as ResourceCSP
|
||||
from fastmcp.apps.config import ResourcePermissions as ResourcePermissions
|
||||
from fastmcp.apps.config import app_config_to_meta_dict as app_config_to_meta_dict
|
||||
from fastmcp.utilities.mime import UI_MIME_TYPE as UI_MIME_TYPE
|
||||
from fastmcp.utilities.mime import resolve_ui_mime_type as resolve_ui_mime_type
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
UI_EXTENSION_ID = "io.modelcontextprotocol/ui"
|
||||
UI_MIME_TYPE = "text/html;profile=mcp-app"
|
||||
|
||||
|
||||
class ResourceCSP(BaseModel):
|
||||
"""Content Security Policy for MCP App resources.
|
||||
|
||||
Declares which external origins the app is allowed to connect to or
|
||||
load resources from. Hosts use these declarations to build the
|
||||
``Content-Security-Policy`` header for the sandboxed iframe.
|
||||
"""
|
||||
|
||||
connect_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="connectDomains",
|
||||
description="Origins allowed for fetch/XHR/WebSocket (connect-src)",
|
||||
)
|
||||
resource_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="resourceDomains",
|
||||
description="Origins allowed for scripts, images, styles, fonts (script-src etc.)",
|
||||
)
|
||||
frame_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="frameDomains",
|
||||
description="Origins allowed for nested iframes (frame-src)",
|
||||
)
|
||||
base_uri_domains: list[str] | None = Field(
|
||||
default=None,
|
||||
alias="baseUriDomains",
|
||||
description="Allowed base URIs for the document (base-uri)",
|
||||
)
|
||||
|
||||
model_config = {"populate_by_name": True, "extra": "allow"}
|
||||
|
||||
|
||||
class ResourcePermissions(BaseModel):
|
||||
"""Iframe sandbox permissions for MCP App resources.
|
||||
|
||||
Each field, when set (typically to ``{}``), requests that the host
|
||||
grant the corresponding Permission Policy feature to the sandboxed
|
||||
iframe. Hosts MAY honour these; apps should use JS feature detection
|
||||
as a fallback.
|
||||
"""
|
||||
|
||||
camera: dict[str, Any] | None = Field(
|
||||
default=None, description="Request camera access"
|
||||
)
|
||||
microphone: dict[str, Any] | None = Field(
|
||||
default=None, description="Request microphone access"
|
||||
)
|
||||
geolocation: dict[str, Any] | None = Field(
|
||||
default=None, description="Request geolocation access"
|
||||
)
|
||||
clipboard_write: dict[str, Any] | None = Field(
|
||||
default=None,
|
||||
alias="clipboardWrite",
|
||||
description="Request clipboard-write access",
|
||||
)
|
||||
|
||||
model_config = {"populate_by_name": True, "extra": "allow"}
|
||||
|
||||
|
||||
class AppConfig(BaseModel):
|
||||
"""Configuration for MCP App tools and resources.
|
||||
|
||||
Controls how a tool or resource participates in the MCP Apps extension.
|
||||
On tools, ``resource_uri`` and ``visibility`` specify which UI resource
|
||||
to render and where the tool appears. On resources, those fields must
|
||||
be left unset (the resource itself is the UI).
|
||||
|
||||
All fields use ``exclude_none`` serialization so only explicitly-set
|
||||
values appear on the wire. Aliases match the MCP Apps wire format
|
||||
(camelCase).
|
||||
"""
|
||||
|
||||
resource_uri: str | None = Field(
|
||||
default=None,
|
||||
alias="resourceUri",
|
||||
description="URI of the UI resource (typically ui:// scheme). Tools only.",
|
||||
)
|
||||
visibility: list[Literal["app", "model"]] | None = Field(
|
||||
default=None,
|
||||
description="Where this tool is visible: 'app', 'model', or both. Tools only.",
|
||||
)
|
||||
csp: ResourceCSP | None = Field(
|
||||
default=None, description="Content Security Policy for the app iframe"
|
||||
)
|
||||
permissions: ResourcePermissions | None = Field(
|
||||
default=None, description="Iframe sandbox permissions"
|
||||
)
|
||||
domain: str | None = Field(default=None, description="Domain for the iframe")
|
||||
prefers_border: bool | None = Field(
|
||||
default=None,
|
||||
alias="prefersBorder",
|
||||
description="Whether the UI prefers a visible border",
|
||||
)
|
||||
|
||||
model_config = {"populate_by_name": True, "extra": "allow"}
|
||||
|
||||
|
||||
def app_config_to_meta_dict(app: AppConfig | dict[str, Any]) -> dict[str, Any]:
|
||||
"""Convert an AppConfig or dict to the wire-format dict for ``meta["ui"]``."""
|
||||
if isinstance(app, AppConfig):
|
||||
return app.model_dump(by_alias=True, exclude_none=True)
|
||||
return app
|
||||
|
||||
|
||||
def resolve_ui_mime_type(uri: str, explicit_mime_type: str | None) -> str | None:
|
||||
"""Return the appropriate MIME type for a resource URI.
|
||||
|
||||
For ``ui://`` scheme resources, defaults to ``UI_MIME_TYPE`` when no
|
||||
explicit MIME type is provided. This ensures UI resources are correctly
|
||||
identified regardless of how they're registered (via FastMCP.resource,
|
||||
the standalone @resource decorator, or resource templates).
|
||||
|
||||
Args:
|
||||
uri: The resource URI string
|
||||
explicit_mime_type: The MIME type explicitly provided by the user
|
||||
|
||||
Returns:
|
||||
The resolved MIME type (explicit value, UI default, or None)
|
||||
"""
|
||||
if explicit_mime_type is not None:
|
||||
return explicit_mime_type
|
||||
# Case-insensitive scheme check per RFC 3986
|
||||
if uri.lower().startswith("ui://"):
|
||||
return UI_MIME_TYPE
|
||||
return None
|
||||
warnings.warn(
|
||||
"'fastmcp.server.apps' is deprecated. Use 'from fastmcp.apps import ...' instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -587,7 +587,7 @@ class Context:
|
|||
|
||||
Example::
|
||||
|
||||
from fastmcp.server.apps import UI_EXTENSION_ID
|
||||
from fastmcp.apps.config import UI_EXTENSION_ID
|
||||
|
||||
@mcp.tool
|
||||
async def my_tool(ctx: Context) -> str:
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ from mcp.shared.message import SessionMessage
|
|||
from mcp.shared.session import RequestResponder
|
||||
from pydantic import AnyUrl
|
||||
|
||||
from fastmcp.server.apps import UI_EXTENSION_ID
|
||||
from fastmcp.apps.config import UI_EXTENSION_ID
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
|
|||
|
|
@ -76,13 +76,13 @@ 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
|
||||
|
||||
from fastmcp.resources.types import TextResource
|
||||
from fastmcp.server.apps import (
|
||||
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:
|
||||
|
|
@ -109,7 +109,7 @@ def _expand_prefab_ui_meta(tool: Tool) -> None:
|
|||
"""Expand meta["ui"] = True into the full AppConfig dict for a prefab tool."""
|
||||
from prefab_ui.renderer import get_renderer_csp
|
||||
|
||||
from fastmcp.server.apps import AppConfig, ResourceCSP, app_config_to_meta_dict
|
||||
from fastmcp.apps.config import AppConfig, ResourceCSP, app_config_to_meta_dict
|
||||
|
||||
csp = get_renderer_csp()
|
||||
app_config = AppConfig(
|
||||
|
|
@ -169,7 +169,7 @@ class ToolDecoratorMixin:
|
|||
# Merge ToolMeta.app into the meta dict
|
||||
tool_meta = fmeta.meta
|
||||
if fmeta.app is not None:
|
||||
from fastmcp.server.apps import app_config_to_meta_dict
|
||||
from fastmcp.apps.config import app_config_to_meta_dict
|
||||
|
||||
tool_meta = dict(tool_meta) if tool_meta else {}
|
||||
if fmeta.app is True:
|
||||
|
|
|
|||
|
|
@ -41,6 +41,11 @@ 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,
|
||||
|
|
@ -56,11 +61,6 @@ 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.apps import (
|
||||
AppConfig,
|
||||
app_config_to_meta_dict,
|
||||
resolve_ui_mime_type,
|
||||
)
|
||||
from fastmcp.server.auth import AuthCheck, AuthContext, AuthProvider, run_auth_checks
|
||||
from fastmcp.server.lifespan import Lifespan
|
||||
from fastmcp.server.low_level import LowLevelServer
|
||||
|
|
|
|||
|
|
@ -491,7 +491,7 @@ _PREFAB_TEXT_FALLBACK = "[Rendered Prefab UI]"
|
|||
def _get_tool_resolver() -> Callable[..., str] | None:
|
||||
"""Get the FastMCPApp callable resolver, if available."""
|
||||
try:
|
||||
from fastmcp.server.app import _resolve_tool_ref
|
||||
from fastmcp.apps.app import _resolve_tool_ref
|
||||
|
||||
return _resolve_tool_ref
|
||||
except ImportError:
|
||||
|
|
|
|||
27
src/fastmcp/utilities/mime.py
Normal file
27
src/fastmcp/utilities/mime.py
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
"""MIME type constants and helpers for MCP Apps UI resources.
|
||||
|
||||
This module has no dependencies on the server or resource packages,
|
||||
so it can be safely imported from anywhere.
|
||||
"""
|
||||
|
||||
UI_MIME_TYPE = "text/html;profile=mcp-app"
|
||||
|
||||
|
||||
def resolve_ui_mime_type(uri: str, explicit_mime_type: str | None) -> str | None:
|
||||
"""Return the appropriate MIME type for a resource URI.
|
||||
|
||||
For ``ui://`` scheme resources, defaults to ``UI_MIME_TYPE`` when no
|
||||
explicit MIME type is provided.
|
||||
|
||||
Args:
|
||||
uri: The resource URI string
|
||||
explicit_mime_type: The MIME type explicitly provided by the user
|
||||
|
||||
Returns:
|
||||
The resolved MIME type (explicit value, UI default, or None)
|
||||
"""
|
||||
if explicit_mime_type is not None:
|
||||
return explicit_mime_type
|
||||
if uri.lower().startswith("ui://"):
|
||||
return UI_MIME_TYPE
|
||||
return None
|
||||
|
|
@ -11,7 +11,7 @@ from typing import Any
|
|||
import pytest
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.server.apps import (
|
||||
from fastmcp.apps import (
|
||||
UI_EXTENSION_ID,
|
||||
UI_MIME_TYPE,
|
||||
AppConfig,
|
||||
|
|
|
|||
|
|
@ -14,8 +14,8 @@ from prefab_ui.components import Column, Heading, Text
|
|||
from prefab_ui.components.base import Component
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.apps import UI_MIME_TYPE, AppConfig
|
||||
from fastmcp.resources.types import TextResource
|
||||
from fastmcp.server.apps import UI_MIME_TYPE, AppConfig
|
||||
from fastmcp.server.providers.local_provider.decorators.tools import (
|
||||
PREFAB_RENDERER_URI,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ from prefab_ui.app import ResolvedTool
|
|||
from prefab_ui.components import Text
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.server.app import (
|
||||
from fastmcp.apps.app import (
|
||||
FastMCPApp,
|
||||
_resolve_tool_ref,
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue