mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Add MCP Apps Phase 1 — SDK compatibility (SEP-1865) (#3009)
This commit is contained in:
Jeremiah Lowin
GitHub
parent
d9cbc06ff1
commit
d52534ae32
10 changed files with 592 additions and 5 deletions
|
|
@ -634,6 +634,70 @@ STDIO transport bypasses all auth checks (no OAuth concept).
|
|||
|
||||
---
|
||||
|
||||
## MCP Apps (SDK Compatibility)
|
||||
|
||||
v3.0 adds Phase 1 support for [MCP Apps](https://modelcontextprotocol.io/specification/2025-06-18/server/apps) — the spec extension that lets MCP servers deliver interactive UIs via sandboxed iframes. Phase 1 is SDK compatibility only: extension negotiation, typed UI metadata on tools and resources, and the `ui://` resource scheme. No component DSL, renderer, or `FastMCPApp` class yet — those are future phases.
|
||||
|
||||
**Registering tools with UI metadata:**
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.apps import ToolUI, ResourceUI, UI_MIME_TYPE
|
||||
|
||||
mcp = FastMCP("My Server")
|
||||
|
||||
# Register the HTML bundle as a ui:// resource
|
||||
@mcp.resource("ui://my-app/view.html")
|
||||
def app_html() -> str:
|
||||
from pathlib import Path
|
||||
return Path("./dist/index.html").read_text()
|
||||
|
||||
# Tool with UI — clients render an iframe alongside the result
|
||||
@mcp.tool(ui=ToolUI(resource_uri="ui://my-app/view.html"))
|
||||
async def list_users() -> list[dict]:
|
||||
return [{"id": "1", "name": "Alice"}]
|
||||
|
||||
# App-only tool — visible to the UI but hidden from the model
|
||||
@mcp.tool(ui=ToolUI(resource_uri="ui://my-app/view.html", visibility=["app"]))
|
||||
async def delete_user(id: str) -> dict:
|
||||
return {"deleted": True}
|
||||
```
|
||||
|
||||
The `ui=` parameter accepts either a typed model (`ToolUI`, `ResourceUI`) or a raw dict for forward compatibility. It merges into `meta["ui"]` — alongside any other metadata you set.
|
||||
|
||||
**`ui://` resources** automatically get the correct MIME type (`text/html;profile=mcp-app`) unless you override it explicitly.
|
||||
|
||||
**Extension negotiation**: The server advertises `io.modelcontextprotocol/ui` in `capabilities.extensions`. UI metadata (`_meta.ui`) always flows through to clients — the MCP Apps spec assigns visibility enforcement to the host, not the server. Tools can check whether the connected client supports a given extension at runtime via `ctx.client_supports_extension()`:
|
||||
|
||||
```python
|
||||
from fastmcp import Context
|
||||
from fastmcp.server.apps import ToolUI, UI_EXTENSION_ID
|
||||
|
||||
@mcp.tool(ui=ToolUI(resource_uri="ui://dashboard"))
|
||||
async def dashboard(ctx: Context) -> dict:
|
||||
data = compute_dashboard()
|
||||
if ctx.client_supports_extension(UI_EXTENSION_ID):
|
||||
# Client will render the iframe with structured data
|
||||
return data
|
||||
# Fallback: text-only summary
|
||||
return {"summary": format_text(data)}
|
||||
```
|
||||
|
||||
**Key details:**
|
||||
- `ToolUI` fields: `resource_uri`, `visibility`, `csp`, `permissions`, `domain`, `prefers_border` (all optional except for typical usage of `resource_uri`)
|
||||
- `ResourceUI` fields: `csp`, `permissions`, `domain`, `prefers_border` — metadata for the resource itself when it's a UI bundle
|
||||
- Models use Pydantic aliases for wire format (`resourceUri`, `prefersBorder`)
|
||||
- `ctx.client_supports_extension(id)` is a general-purpose method — works for any extension, not just MCP Apps
|
||||
- `structuredContent` in tool results already works via `ToolResult` — MCP Apps clients use this to pass data into the iframe
|
||||
- Text content fallback already works — tools return both `content` and `structured_content`
|
||||
- The server does not strip `_meta.ui` for non-UI clients; per the spec, visibility enforcement is the host's responsibility
|
||||
|
||||
**Future phases** will add a component DSL for building UIs declaratively, an in-repo renderer, and a `FastMCPApp` class.
|
||||
|
||||
Implementation: `src/fastmcp/server/apps.py` (models and constants), with integration points in `server.py` (decorator parameters), `low_level.py` (extension advertisement), and `context.py` (`client_supports_extension` method).
|
||||
|
||||
---
|
||||
|
||||
## FileSystemProvider
|
||||
|
||||
v3.0 introduces `FileSystemProvider`, a fundamentally different approach to organizing MCP servers. Instead of importing a server instance and decorating functions with `@server.tool`, you use standalone decorators in separate files and let the provider discover them.
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from pydantic import AnyUrl
|
|||
import fastmcp
|
||||
from fastmcp.decorators import resolve_task_config
|
||||
from fastmcp.resources.resource import Resource, ResourceResult
|
||||
from fastmcp.server.apps import resolve_ui_mime_type
|
||||
from fastmcp.server.dependencies import (
|
||||
transform_context_annotations,
|
||||
without_injected_parameters,
|
||||
|
|
@ -180,6 +181,9 @@ class FunctionResource(Resource):
|
|||
# Wrap fn to handle dependency resolution internally
|
||||
wrapped_fn = without_injected_parameters(fn)
|
||||
|
||||
# Apply ui:// MIME default, then fall back to text/plain
|
||||
resolved_mime = resolve_ui_mime_type(metadata.uri, metadata.mime_type)
|
||||
|
||||
return cls(
|
||||
fn=wrapped_fn,
|
||||
uri=uri_obj,
|
||||
|
|
@ -188,7 +192,7 @@ class FunctionResource(Resource):
|
|||
title=metadata.title,
|
||||
description=metadata.description or inspect.getdoc(fn),
|
||||
icons=metadata.icons,
|
||||
mime_type=metadata.mime_type or "text/plain",
|
||||
mime_type=resolved_mime or "text/plain",
|
||||
tags=metadata.tags or set(),
|
||||
annotations=metadata.annotations,
|
||||
meta=metadata.meta,
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ from pydantic import (
|
|||
)
|
||||
|
||||
from fastmcp.resources.resource import Resource, ResourceResult
|
||||
from fastmcp.server.apps import resolve_ui_mime_type
|
||||
from fastmcp.server.dependencies import (
|
||||
transform_context_annotations,
|
||||
without_injected_parameters,
|
||||
|
|
@ -567,6 +568,9 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
# Use validate_call on wrapper for runtime type coercion
|
||||
fn = validate_call(wrapper_fn)
|
||||
|
||||
# Apply ui:// MIME default, then fall back to text/plain
|
||||
resolved_mime = resolve_ui_mime_type(uri_template, mime_type)
|
||||
|
||||
return cls(
|
||||
uri_template=uri_template,
|
||||
name=func_name,
|
||||
|
|
@ -574,7 +578,7 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
title=title,
|
||||
description=description,
|
||||
icons=icons,
|
||||
mime_type=mime_type or "text/plain",
|
||||
mime_type=resolved_mime or "text/plain",
|
||||
fn=fn,
|
||||
parameters=parameters,
|
||||
tags=tags or set(),
|
||||
|
|
|
|||
93
src/fastmcp/server/apps.py
Normal file
93
src/fastmcp/server/apps.py
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
"""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
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
UI_EXTENSION_ID = "io.modelcontextprotocol/ui"
|
||||
UI_MIME_TYPE = "text/html;profile=mcp-app"
|
||||
|
||||
|
||||
class ToolUI(BaseModel):
|
||||
"""Typed ``_meta.ui`` for tools — links a tool to its UI resource.
|
||||
|
||||
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)",
|
||||
)
|
||||
visibility: list[str] | None = Field(
|
||||
default=None,
|
||||
description="Where this tool is visible: 'app', 'model', or both",
|
||||
)
|
||||
csp: str | None = Field(default=None, description="Content Security Policy")
|
||||
permissions: list[str] | None = Field(
|
||||
default=None, description="iframe 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}
|
||||
|
||||
|
||||
class ResourceUI(BaseModel):
|
||||
"""Typed ``_meta.ui`` for resources — rendering hints for UI-capable clients."""
|
||||
|
||||
csp: str | None = Field(default=None, description="Content Security Policy")
|
||||
permissions: list[str] | None = Field(
|
||||
default=None, description="iframe 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}
|
||||
|
||||
|
||||
def ui_to_meta_dict(ui: ToolUI | ResourceUI | dict[str, Any]) -> dict[str, Any]:
|
||||
"""Convert a UI model or dict to the wire-format dict for ``meta["ui"]``."""
|
||||
if isinstance(ui, (ToolUI, ResourceUI)):
|
||||
return ui.model_dump(by_alias=True, exclude_none=True)
|
||||
return ui
|
||||
|
||||
|
||||
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
|
||||
|
|
@ -33,6 +33,7 @@ from fastmcp.server.elicitation import (
|
|||
handle_elicit_accept,
|
||||
parse_elicit_response_type,
|
||||
)
|
||||
from fastmcp.server.low_level import MiddlewareServerSession
|
||||
from fastmcp.server.sampling import SampleStep, SamplingResult, SamplingTool
|
||||
from fastmcp.server.sampling.run import (
|
||||
sample_impl,
|
||||
|
|
@ -455,6 +456,33 @@ class Context:
|
|||
"""
|
||||
return _current_transport.get()
|
||||
|
||||
def client_supports_extension(self, extension_id: str) -> bool:
|
||||
"""Check whether the connected client supports a given MCP extension.
|
||||
|
||||
Inspects the ``extensions`` extra field on ``ClientCapabilities``
|
||||
sent by the client during initialization.
|
||||
|
||||
Returns ``False`` when no session is available (e.g., outside a
|
||||
request context) or when the client did not advertise the extension.
|
||||
|
||||
Example::
|
||||
|
||||
from fastmcp.server.apps import UI_EXTENSION_ID
|
||||
|
||||
@mcp.tool
|
||||
async def my_tool(ctx: Context) -> str:
|
||||
if ctx.client_supports_extension(UI_EXTENSION_ID):
|
||||
return "UI-capable client"
|
||||
return "text-only client"
|
||||
"""
|
||||
rc = self.request_context
|
||||
if rc is None:
|
||||
return False
|
||||
session = rc.session
|
||||
if not isinstance(session, MiddlewareServerSession):
|
||||
return False
|
||||
return session.client_supports_extension(extension_id)
|
||||
|
||||
@property
|
||||
def client_id(self) -> str | None:
|
||||
"""Get the client ID if available."""
|
||||
|
|
|
|||
|
|
@ -24,6 +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.utilities.logging import get_logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -49,6 +50,25 @@ class MiddlewareServerSession(ServerSession):
|
|||
raise RuntimeError("FastMCP instance is no longer available")
|
||||
return fastmcp
|
||||
|
||||
def client_supports_extension(self, extension_id: str) -> bool:
|
||||
"""Check if the connected client supports a given MCP extension.
|
||||
|
||||
Inspects the ``extensions`` extra field on ``ClientCapabilities``
|
||||
sent by the client during initialization.
|
||||
"""
|
||||
client_params = self._client_params
|
||||
if client_params is None:
|
||||
return False
|
||||
caps = client_params.capabilities
|
||||
if caps is None:
|
||||
return False
|
||||
# ClientCapabilities uses extra="allow" — extensions is an extra field
|
||||
extras = caps.model_extra or {}
|
||||
extensions: dict[str, Any] | None = extras.get("extensions")
|
||||
if not extensions:
|
||||
return False
|
||||
return extension_id in extensions
|
||||
|
||||
async def _received_request(
|
||||
self,
|
||||
responder: RequestResponder[mcp.types.ClientRequest, mcp.types.ServerResult],
|
||||
|
|
@ -188,6 +208,15 @@ class LowLevelServer(_Server[LifespanResultT, RequestT]):
|
|||
# Set tasks as a first-class field (not experimental) per SEP-1686
|
||||
capabilities.tasks = get_task_capabilities()
|
||||
|
||||
# Advertise MCP Apps extension support (io.modelcontextprotocol/ui)
|
||||
# Uses the same extra-field pattern as tasks above — ServerCapabilities
|
||||
# has extra="allow" so this survives serialization.
|
||||
# Merge with any existing extensions to avoid clobbering other features.
|
||||
existing_extensions: dict[str, Any] = (
|
||||
getattr(capabilities, "extensions", None) or {}
|
||||
)
|
||||
capabilities.extensions = {**existing_extensions, UI_EXTENSION_ID: {}}
|
||||
|
||||
return capabilities
|
||||
|
||||
async def run(
|
||||
|
|
|
|||
|
|
@ -154,6 +154,7 @@ class MCPOperationsMixin:
|
|||
|
||||
tools = _dedupe_with_versions(list(await server.list_tools()), lambda t: t.name)
|
||||
sdk_tools = [tool.to_mcp_tool(name=tool.name) for tool in tools]
|
||||
|
||||
# SDK may pass None for internal cache refresh despite type hint
|
||||
cursor = (
|
||||
request.params.cursor if request is not None and request.params else None
|
||||
|
|
@ -177,6 +178,7 @@ class MCPOperationsMixin:
|
|||
sdk_resources = [
|
||||
resource.to_mcp_resource(uri=str(resource.uri)) for resource in resources
|
||||
]
|
||||
|
||||
cursor = request.params.cursor if request.params else None
|
||||
page, next_cursor = _apply_pagination(
|
||||
sdk_resources, cursor, server._list_page_size
|
||||
|
|
|
|||
|
|
@ -58,6 +58,12 @@ from fastmcp.prompts.function_prompt import FunctionPrompt
|
|||
from fastmcp.prompts.prompt import PromptResult
|
||||
from fastmcp.resources.resource import Resource, ResourceResult
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.server.apps import (
|
||||
ResourceUI,
|
||||
ToolUI,
|
||||
resolve_ui_mime_type,
|
||||
ui_to_meta_dict,
|
||||
)
|
||||
from fastmcp.server.auth import AuthContext, AuthProvider, run_auth_checks
|
||||
from fastmcp.server.dependencies import get_access_token
|
||||
from fastmcp.server.lifespan import Lifespan
|
||||
|
|
@ -1370,6 +1376,7 @@ class FastMCP(
|
|||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
ui: ToolUI | dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
|
|
@ -1390,6 +1397,7 @@ class FastMCP(
|
|||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
ui: ToolUI | dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
|
|
@ -1409,6 +1417,7 @@ class FastMCP(
|
|||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
ui: ToolUI | dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
|
|
@ -1465,6 +1474,11 @@ class FastMCP(
|
|||
server.tool(my_function, name="custom_name")
|
||||
```
|
||||
"""
|
||||
# Merge UI metadata into meta["ui"] before passing to provider
|
||||
if ui is not None:
|
||||
meta = dict(meta) if meta else {}
|
||||
meta["ui"] = ui_to_meta_dict(ui)
|
||||
|
||||
# Delegate to LocalProvider with server-level defaults
|
||||
result = self._local_provider.tool(
|
||||
name_or_fn,
|
||||
|
|
@ -1523,6 +1537,7 @@ class FastMCP(
|
|||
tags: set[str] | None = None,
|
||||
annotations: Annotations | dict[str, Any] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
ui: ResourceUI | dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[AnyFunction], Resource | ResourceTemplate | AnyFunction]:
|
||||
|
|
@ -1577,6 +1592,22 @@ class FastMCP(
|
|||
return f"Weather for {city}: {data}"
|
||||
```
|
||||
"""
|
||||
# Catch incorrect decorator usage early (before any processing)
|
||||
if not isinstance(uri, str):
|
||||
raise TypeError(
|
||||
"The @resource decorator was used incorrectly. "
|
||||
"It requires a URI as the first argument. "
|
||||
"Use @resource('uri') instead of @resource"
|
||||
)
|
||||
|
||||
# Apply default MIME type for ui:// scheme resources
|
||||
mime_type = resolve_ui_mime_type(uri, mime_type)
|
||||
|
||||
# Merge UI metadata into meta["ui"] before passing to provider
|
||||
if ui is not None:
|
||||
meta = dict(meta) if meta else {}
|
||||
meta["ui"] = ui_to_meta_dict(ui)
|
||||
|
||||
# Delegate to LocalProvider with server-level defaults
|
||||
inner_decorator = self._local_provider.resource(
|
||||
uri,
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
|
@ -202,9 +203,6 @@ async def download_skill(
|
|||
if isinstance(content, mcp.types.TextResourceContents):
|
||||
file_path.write_text(content.text)
|
||||
elif isinstance(content, mcp.types.BlobResourceContents):
|
||||
# Handle base64-encoded binary content
|
||||
import base64
|
||||
|
||||
file_path.write_bytes(base64.b64decode(content.blob))
|
||||
else:
|
||||
# Skip unknown content types
|
||||
|
|
|
|||
334
tests/test_apps.py
Normal file
334
tests/test_apps.py
Normal file
|
|
@ -0,0 +1,334 @@
|
|||
"""Tests for MCP Apps Phase 1 — SDK compatibility.
|
||||
|
||||
Covers UI metadata models, tool/resource registration with ``ui=``,
|
||||
extension negotiation, and the ``Context.client_supports_extension`` method.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.server.apps import (
|
||||
UI_EXTENSION_ID,
|
||||
UI_MIME_TYPE,
|
||||
ResourceUI,
|
||||
ToolUI,
|
||||
ui_to_meta_dict,
|
||||
)
|
||||
from fastmcp.server.context import Context
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Model serialization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestToolUI:
|
||||
def test_serializes_with_aliases(self):
|
||||
ui = ToolUI(resource_uri="ui://my-app/view.html", visibility=["app"])
|
||||
d = ui.model_dump(by_alias=True, exclude_none=True)
|
||||
assert d == {"resourceUri": "ui://my-app/view.html", "visibility": ["app"]}
|
||||
|
||||
def test_excludes_none_fields(self):
|
||||
ui = ToolUI(resource_uri="ui://foo")
|
||||
d = ui.model_dump(by_alias=True, exclude_none=True)
|
||||
assert d == {"resourceUri": "ui://foo"}
|
||||
|
||||
def test_all_fields(self):
|
||||
ui = ToolUI(
|
||||
resource_uri="ui://app",
|
||||
visibility=["app", "model"],
|
||||
csp="default-src 'self'",
|
||||
permissions=["clipboard-read"],
|
||||
domain="example.com",
|
||||
prefers_border=True,
|
||||
)
|
||||
d = ui.model_dump(by_alias=True, exclude_none=True)
|
||||
assert d == {
|
||||
"resourceUri": "ui://app",
|
||||
"visibility": ["app", "model"],
|
||||
"csp": "default-src 'self'",
|
||||
"permissions": ["clipboard-read"],
|
||||
"domain": "example.com",
|
||||
"prefersBorder": True,
|
||||
}
|
||||
|
||||
def test_populate_by_name(self):
|
||||
ui = ToolUI(resource_uri="ui://app")
|
||||
assert ui.resource_uri == "ui://app"
|
||||
|
||||
|
||||
class TestResourceUI:
|
||||
def test_serializes_with_aliases(self):
|
||||
ui = ResourceUI(prefers_border=True, csp="default-src 'self'")
|
||||
d = ui.model_dump(by_alias=True, exclude_none=True)
|
||||
assert d == {"prefersBorder": True, "csp": "default-src 'self'"}
|
||||
|
||||
def test_excludes_none_fields(self):
|
||||
ui = ResourceUI()
|
||||
d = ui.model_dump(by_alias=True, exclude_none=True)
|
||||
assert d == {}
|
||||
|
||||
|
||||
class TestUIToMetaDict:
|
||||
def test_from_tool_ui(self):
|
||||
ui = ToolUI(resource_uri="ui://app", visibility=["app"])
|
||||
result = ui_to_meta_dict(ui)
|
||||
assert result["resourceUri"] == "ui://app"
|
||||
assert result["visibility"] == ["app"]
|
||||
|
||||
def test_from_resource_ui(self):
|
||||
ui = ResourceUI(prefers_border=False)
|
||||
result = ui_to_meta_dict(ui)
|
||||
assert result == {"prefersBorder": False}
|
||||
|
||||
def test_passthrough_for_dict(self):
|
||||
raw: dict[str, Any] = {"resourceUri": "ui://app", "custom": "value"}
|
||||
result = ui_to_meta_dict(raw)
|
||||
assert result is raw
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool registration with ui=
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestToolRegistrationWithUI:
|
||||
async def test_tool_ui_model(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(ui=ToolUI(resource_uri="ui://my-app/view.html"))
|
||||
def my_tool() -> str:
|
||||
return "hello"
|
||||
|
||||
tools = list(await server.list_tools())
|
||||
assert len(tools) == 1
|
||||
assert tools[0].meta is not None
|
||||
assert tools[0].meta["ui"]["resourceUri"] == "ui://my-app/view.html"
|
||||
|
||||
async def test_tool_ui_dict(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(ui={"resourceUri": "ui://foo", "visibility": ["app"]})
|
||||
def my_tool() -> str:
|
||||
return "hello"
|
||||
|
||||
tools = list(await server.list_tools())
|
||||
assert tools[0].meta is not None
|
||||
assert tools[0].meta["ui"]["resourceUri"] == "ui://foo"
|
||||
assert tools[0].meta["ui"]["visibility"] == ["app"]
|
||||
|
||||
async def test_ui_merges_with_existing_meta(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(meta={"custom": "data"}, ui=ToolUI(resource_uri="ui://app"))
|
||||
def my_tool() -> str:
|
||||
return "hello"
|
||||
|
||||
tools = list(await server.list_tools())
|
||||
meta = tools[0].meta
|
||||
assert meta is not None
|
||||
assert meta["custom"] == "data"
|
||||
assert meta["ui"]["resourceUri"] == "ui://app"
|
||||
|
||||
async def test_ui_in_mcp_wire_format(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(ui=ToolUI(resource_uri="ui://app", visibility=["app"]))
|
||||
def my_tool() -> str:
|
||||
return "hello"
|
||||
|
||||
tools = list(await server.list_tools())
|
||||
mcp_tool = tools[0].to_mcp_tool()
|
||||
assert mcp_tool.meta is not None
|
||||
assert mcp_tool.meta["ui"]["resourceUri"] == "ui://app"
|
||||
assert mcp_tool.meta["ui"]["visibility"] == ["app"]
|
||||
|
||||
async def test_tool_without_ui_has_no_ui_meta(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool
|
||||
def my_tool() -> str:
|
||||
return "hello"
|
||||
|
||||
tools = list(await server.list_tools())
|
||||
meta = tools[0].meta
|
||||
assert meta is None or "ui" not in meta
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Resource registration with ui:// and ui=
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestResourceWithUI:
|
||||
async def test_ui_scheme_defaults_mime_type(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.resource("ui://my-app/view.html")
|
||||
def app_html() -> str:
|
||||
return "<html>hello</html>"
|
||||
|
||||
resources = list(await server.list_resources())
|
||||
assert len(resources) == 1
|
||||
assert resources[0].mime_type == UI_MIME_TYPE
|
||||
|
||||
async def test_explicit_mime_type_overrides_ui_default(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.resource("ui://my-app/view.html", mime_type="text/html")
|
||||
def app_html() -> str:
|
||||
return "<html>hello</html>"
|
||||
|
||||
resources = list(await server.list_resources())
|
||||
assert resources[0].mime_type == "text/html"
|
||||
|
||||
async def test_resource_ui_metadata(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.resource(
|
||||
"ui://my-app/view.html",
|
||||
ui=ResourceUI(prefers_border=True),
|
||||
)
|
||||
def app_html() -> str:
|
||||
return "<html>hello</html>"
|
||||
|
||||
resources = list(await server.list_resources())
|
||||
assert resources[0].meta is not None
|
||||
assert resources[0].meta["ui"]["prefersBorder"] is True
|
||||
|
||||
async def test_non_ui_scheme_no_mime_default(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.resource("resource://data")
|
||||
def data() -> str:
|
||||
return "data"
|
||||
|
||||
resources = list(await server.list_resources())
|
||||
assert resources[0].mime_type != UI_MIME_TYPE
|
||||
|
||||
async def test_standalone_decorator_ui_scheme_defaults_mime_type(self):
|
||||
"""Test that the standalone @resource decorator also applies ui:// MIME default."""
|
||||
from fastmcp.resources import resource
|
||||
|
||||
@resource("ui://standalone-app/view.html")
|
||||
def standalone_app() -> str:
|
||||
return "<html>standalone</html>"
|
||||
|
||||
server = FastMCP("test")
|
||||
server.add_resource(standalone_app)
|
||||
|
||||
resources = list(await server.list_resources())
|
||||
assert len(resources) == 1
|
||||
assert resources[0].mime_type == UI_MIME_TYPE
|
||||
|
||||
async def test_resource_template_ui_scheme_defaults_mime_type(self):
|
||||
"""Test that resource templates also apply ui:// MIME default."""
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.resource("ui://template-app/{view}")
|
||||
def template_app(view: str) -> str:
|
||||
return f"<html>{view}</html>"
|
||||
|
||||
templates = list(await server.list_resource_templates())
|
||||
assert len(templates) == 1
|
||||
assert templates[0].mime_type == UI_MIME_TYPE
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Extension advertisement
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExtensionAdvertisement:
|
||||
async def test_capabilities_include_ui_extension(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool
|
||||
def my_tool() -> str:
|
||||
return "hello"
|
||||
|
||||
async with Client(server) as client:
|
||||
init_result = client.initialize_result
|
||||
extras = init_result.capabilities.model_extra or {}
|
||||
extensions = extras.get("extensions", {})
|
||||
assert UI_EXTENSION_ID in extensions
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Context.client_supports_extension
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestContextClientSupportsExtension:
|
||||
async def test_returns_false_when_no_session(self):
|
||||
server = FastMCP("test")
|
||||
async with Context(fastmcp=server) as ctx:
|
||||
assert ctx.client_supports_extension(UI_EXTENSION_ID) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration — full client↔server round-trip
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestIntegration:
|
||||
async def test_tool_with_ui_roundtrip(self):
|
||||
"""UI metadata flows through to clients — no server-side stripping."""
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(ui=ToolUI(resource_uri="ui://app/view.html", visibility=["app"]))
|
||||
async def my_tool() -> dict[str, str]:
|
||||
return {"result": "ok"}
|
||||
|
||||
async with Client(server) as client:
|
||||
tools = await client.list_tools()
|
||||
assert len(tools) == 1
|
||||
# _meta.ui is preserved — the host decides what to do with it
|
||||
meta = tools[0].meta
|
||||
assert meta is not None
|
||||
assert meta["ui"]["resourceUri"] == "ui://app/view.html"
|
||||
assert meta["ui"]["visibility"] == ["app"]
|
||||
|
||||
async def test_resource_with_ui_scheme_roundtrip(self):
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.resource("ui://my-app/view.html")
|
||||
def app_html() -> str:
|
||||
return "<html><body>Hello</body></html>"
|
||||
|
||||
async with Client(server) as client:
|
||||
resources = await client.list_resources()
|
||||
assert len(resources) == 1
|
||||
assert str(resources[0].uri) == "ui://my-app/view.html"
|
||||
assert resources[0].mimeType == UI_MIME_TYPE
|
||||
|
||||
async def test_ui_tool_callable(self):
|
||||
"""A tool registered with ui= is still callable normally."""
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(ui=ToolUI(resource_uri="ui://app"))
|
||||
async def greet(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
|
||||
async with Client(server) as client:
|
||||
result = await client.call_tool("greet", {"name": "Alice"})
|
||||
assert any("Hello, Alice!" in str(c) for c in result.content)
|
||||
|
||||
async def test_extension_and_tool_together(self):
|
||||
"""Server advertises extension AND tool has UI meta (stored on FastMCP Tool)."""
|
||||
server = FastMCP("test")
|
||||
|
||||
@server.tool(ui=ToolUI(resource_uri="ui://dashboard", visibility=["app"]))
|
||||
def dashboard() -> str:
|
||||
return "data"
|
||||
|
||||
# Verify the stored FastMCP Tool still has full metadata
|
||||
tools = list(await server.list_tools())
|
||||
assert tools[0].meta is not None
|
||||
assert tools[0].meta["ui"]["resourceUri"] == "ui://dashboard"
|
||||
|
||||
# Verify the server advertises the extension
|
||||
async with Client(server) as client:
|
||||
extras = client.initialize_result.capabilities.model_extra or {}
|
||||
assert UI_EXTENSION_ID in extras.get("extensions", {})
|
||||
Loading…
Add table
Add a link
Reference in a new issue