Compare commits

...

3 commits

Author SHA1 Message Date
Jeremiah Lowin
ae82aa511e
Merge branch 'main' into feature/apps-merge-main 2026-02-13 23:00:32 -05:00
Jeremiah Lowin
ccb74da45c
Merge origin/main into feature/apps 2026-02-13 20:41:53 -05:00
Jeremiah Lowin
aafd26d43a
Add prefab auto-wiring for MCP Apps (#3119)
Tools that return prefab types (UIResponse, Component) automatically get
wired to the shared prefab renderer resource. Works via app=True,
return type inference, or both.
2026-02-08 21:10:44 -05:00
10 changed files with 811 additions and 18 deletions

View file

@ -0,0 +1,101 @@
"""Chart MCP App — interactive data visualizations with Prefab.
Demonstrates `fastmcp[apps]` with Prefab chart components:
- `BarChart` and `LineChart` for categorical and trend data
- Multiple series, stacking, and curve styles
- Layout composition with `Column`, `Heading`, and `Muted`
Usage:
uv run python chart_server.py # HTTP (port 8000)
uv run python chart_server.py --stdio # stdio for MCP clients
"""
from __future__ import annotations
from prefab_ui.components import (
BarChart,
ChartSeries,
Column,
Heading,
LineChart,
Muted,
)
from prefab_ui.response import UIResponse
from fastmcp import FastMCP
mcp = FastMCP("Sales Dashboard")
MONTHLY_SALES = [
{"month": "Jan", "online": 4200, "retail": 2400},
{"month": "Feb", "online": 3800, "retail": 2100},
{"month": "Mar", "online": 5100, "retail": 2800},
{"month": "Apr", "online": 4600, "retail": 3200},
{"month": "May", "online": 5800, "retail": 3100},
{"month": "Jun", "online": 6200, "retail": 3500},
]
@mcp.tool(app=True)
def sales_overview(stacked: bool = False) -> UIResponse:
"""View monthly sales broken down by channel.
Args:
stacked: Stack bars to show total revenue per month.
"""
total = sum(row["online"] + row["retail"] for row in MONTHLY_SALES)
with Column(gap=6, css_class="p-6") as view:
with Column(gap=1):
Heading("Monthly Sales")
Muted(f"${total:,} total revenue")
BarChart(
data=MONTHLY_SALES,
series=[
ChartSeries(data_key="online", label="Online"),
ChartSeries(data_key="retail", label="Retail"),
],
x_axis="month",
stacked=stacked,
show_legend=True,
)
return UIResponse(
view=view,
text=f"Monthly sales: ${total:,} total revenue across 2 channels",
)
@mcp.tool(app=True)
def sales_trend(curve: str = "linear") -> UIResponse:
"""View sales trends over time as a line chart.
Args:
curve: Line style "linear", "smooth", or "step".
"""
with Column(gap=6, css_class="p-6") as view:
with Column(gap=1):
Heading("Sales Trend")
Muted("Online vs. retail over 6 months")
LineChart(
data=MONTHLY_SALES,
series=[
ChartSeries(data_key="online", label="Online"),
ChartSeries(data_key="retail", label="Retail"),
],
x_axis="month",
curve=curve,
show_dots=True,
show_legend=True,
)
return UIResponse(
view=view,
text="Sales trend across online and retail channels",
)
if __name__ == "__main__":
mcp.run()

View file

@ -0,0 +1,165 @@
"""DataTable MCP App — interactive, sortable data views with Prefab.
Demonstrates `fastmcp[apps]` with Prefab UI components:
- `app=True` for automatic renderer wiring
- `UIResponse` with `DataTable` for rich tabular output
- Searchable, sortable, paginated tables
- Layout composition with `Column`, `Heading`, `Text`, and `Badge`
Usage:
uv run python datatable_server.py # HTTP (port 8000)
uv run python datatable_server.py --stdio # stdio for MCP clients
"""
from __future__ import annotations
from prefab_ui.components import (
Badge,
Column,
DataTable,
DataTableColumn,
Heading,
Muted,
Row,
)
from prefab_ui.response import UIResponse
from fastmcp import FastMCP
mcp = FastMCP("Team Directory")
EMPLOYEES = [
{
"name": "Alice Chen",
"role": "Engineering",
"level": "Senior",
"location": "San Francisco",
"status": "active",
},
{
"name": "Bob Martinez",
"role": "Design",
"level": "Lead",
"location": "New York",
"status": "active",
},
{
"name": "Carol Johnson",
"role": "Engineering",
"level": "Staff",
"location": "London",
"status": "active",
},
{
"name": "David Kim",
"role": "Product",
"level": "Senior",
"location": "San Francisco",
"status": "away",
},
{
"name": "Eva Müller",
"role": "Engineering",
"level": "Mid",
"location": "Berlin",
"status": "active",
},
{
"name": "Frank Okafor",
"role": "Data Science",
"level": "Senior",
"location": "Lagos",
"status": "active",
},
{
"name": "Grace Liu",
"role": "Engineering",
"level": "Junior",
"location": "Singapore",
"status": "active",
},
{
"name": "Hassan Ali",
"role": "Design",
"level": "Senior",
"location": "Dubai",
"status": "away",
},
{
"name": "Iris Tanaka",
"role": "Product",
"level": "Lead",
"location": "Tokyo",
"status": "active",
},
{
"name": "James Wright",
"role": "Engineering",
"level": "Senior",
"location": "London",
"status": "inactive",
},
{
"name": "Karen Petrov",
"role": "Data Science",
"level": "Lead",
"location": "Berlin",
"status": "active",
},
{
"name": "Liam O'Brien",
"role": "Engineering",
"level": "Mid",
"location": "Dublin",
"status": "active",
},
]
@mcp.tool(app=True)
def list_team(department: str | None = None) -> UIResponse:
"""Browse the team directory with sorting and search.
Args:
department: Filter by department (e.g. "Engineering", "Design").
Leave empty to show everyone.
"""
if department:
rows = [e for e in EMPLOYEES if e["role"].lower() == department.lower()]
else:
rows = EMPLOYEES
active = sum(1 for e in rows if e["status"] == "active")
with Column(gap=6, css_class="p-6") as view:
with Column(gap=1):
Heading("Team Directory")
with Row(gap=2):
Muted(f"{len(rows)} members")
Muted(f"{active} active", css_class="text-success")
if department:
Badge(department, variant="outline")
DataTable(
columns=[
DataTableColumn(key="name", header="Name", sortable=True),
DataTableColumn(key="role", header="Department", sortable=True),
DataTableColumn(key="level", header="Level", sortable=True),
DataTableColumn(key="location", header="Location", sortable=True),
DataTableColumn(key="status", header="Status", sortable=True),
],
rows=rows,
searchable=True,
paginated=True,
page_size=10,
)
return UIResponse(
view=view,
state={"total": len(rows), "active": active},
text=f"Team directory: {len(rows)} members ({active} active)",
)
if __name__ == "__main__":
mcp.run()

View file

@ -52,6 +52,7 @@ classifiers = [
[project.optional-dependencies]
anthropic = ["anthropic>=0.40.0"]
apps = ["prefab-ui>=0.1.0"]
azure = ["azure-identity>=1.16.0"]
openai = ["openai>=1.102.0"]
tasks = ["pydocket>=0.17.2"]
@ -59,7 +60,7 @@ tasks = ["pydocket>=0.17.2"]
[dependency-groups]
dev = [
"dirty-equals>=0.9.0",
"fastmcp[anthropic,azure,openai,tasks]",
"fastmcp[anthropic,apps,azure,openai,tasks]",
# add optional dependencies for fastmcp dev
"fastapi>=0.115.12",
"opentelemetry-sdk>=1.20.0",
@ -104,6 +105,9 @@ source = "uv-dynamic-versioning"
[tool.hatch.metadata]
allow-direct-references = true
[tool.uv.sources]
prefab-ui = { path = "../prefab", editable = true }
[tool.uv-dynamic-versioning]
vcs = "git"
style = "pep440"
@ -190,18 +194,6 @@ known-first-party = ["fastmcp"]
"SIM", # flake8-simplify
]
[tool.basedpyright]
pythonVersion = "3.10"
typeCheckingMode = "standard"
reportMissingTypeStubs = false
reportUnknownParameterType = false
reportUnknownArgumentType = false
reportUnknownMemberType = false
reportUnknownVariableType = false
reportPrivateUsage = false
reportUnnecessaryIsInstance = false
reportUnnecessaryComparison = false
reportConstantRedefinition = false
[tool.codespell]
ignore-words-list = "asend,shttp,te"

View file

@ -26,7 +26,11 @@ class TextResource(Resource):
async def read(self) -> ResourceResult:
"""Read the text content."""
return ResourceResult(
contents=[ResourceContent(content=self.text, mime_type=self.mime_type)]
contents=[
ResourceContent(
content=self.text, mime_type=self.mime_type, meta=self.meta
)
]
)
@ -38,7 +42,11 @@ class BinaryResource(Resource):
async def read(self) -> ResourceResult:
"""Read the binary content."""
return ResourceResult(
contents=[ResourceContent(content=self.data, mime_type=self.mime_type)]
contents=[
ResourceContent(
content=self.data, mime_type=self.mime_type, meta=self.meta
)
]
)

View file

@ -22,12 +22,113 @@ from fastmcp.tools.function_tool import FunctionTool
from fastmcp.tools.tool import Tool
from fastmcp.utilities.types import NotSet, NotSetT
try:
from prefab_ui.components.base import Component as _PrefabComponent
from prefab_ui.response import UIResponse as _PrefabUIResponse
_HAS_PREFAB = True
except ImportError:
_HAS_PREFAB = False
if TYPE_CHECKING:
from fastmcp.server.providers.local_provider import LocalProvider
from fastmcp.tools.tool import ToolResultSerializerType
DuplicateBehavior = Literal["error", "warn", "replace", "ignore"]
PREFAB_RENDERER_URI = "ui://prefab/renderer.html"
def _has_prefab_return_type(tool: Tool) -> bool:
"""Check if a FunctionTool's return type annotation is a prefab type."""
if not _HAS_PREFAB or not isinstance(tool, FunctionTool):
return False
rt = tool.return_type
if rt is None or rt is inspect.Parameter.empty:
return False
# Direct type check
if isinstance(rt, type) and issubclass(rt, (_PrefabUIResponse, _PrefabComponent)):
return True
# Check Union args (e.g., UIResponse | None)
from typing import get_args
args = get_args(rt)
return any(
isinstance(a, type) and issubclass(a, (_PrefabUIResponse, _PrefabComponent))
for a in args
)
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 (
UI_MIME_TYPE,
AppConfig,
ResourceCSP,
app_config_to_meta_dict,
)
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
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."""
from prefab_ui.renderer import get_renderer_csp
from fastmcp.server.apps import AppConfig, ResourceCSP, app_config_to_meta_dict
csp = get_renderer_csp()
app_config = AppConfig(
resource_uri=PREFAB_RENDERER_URI,
csp=ResourceCSP(
resource_domains=csp.get("resource_domains"),
connect_domains=csp.get("connect_domains"),
),
)
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."""
if not _HAS_PREFAB:
return
meta = tool.meta or {}
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)
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)
# If ui is a dict, it's already manually configured — leave it alone
class ToolDecoratorMixin:
"""Mixin class providing tool decorator functionality for LocalProvider.
@ -85,6 +186,7 @@ class ToolDecoratorMixin:
self._add_component(tool)
if not enabled:
self.disable(keys={tool.key})
_maybe_apply_prefab_ui(self, tool)
return tool
@overload
@ -262,6 +364,7 @@ class ToolDecoratorMixin:
self._add_component(tool_obj)
if not enabled:
self.disable(keys={tool_obj.key})
_maybe_apply_prefab_ui(self, tool_obj)
return tool_obj
else:
from fastmcp.tools.function_tool import ToolMeta

View file

@ -27,6 +27,14 @@ from fastmcp.utilities.types import (
replace_type,
)
try:
from prefab_ui.components.base import Component as _PrefabComponent
from prefab_ui.response import UIResponse as _PrefabUIResponse
_PREFAB_TYPES: tuple[type, ...] = (_PrefabUIResponse, _PrefabComponent)
except ImportError:
_PREFAB_TYPES = ()
T = TypeVarExt("T", default=Any)
logger = get_logger(__name__)
@ -65,6 +73,7 @@ class ParsedFunction:
description: str | None
input_schema: dict[str, Any]
output_schema: dict[str, Any] | None
return_type: Any = None
@classmethod
def from_function(
@ -145,7 +154,20 @@ class ParsedFunction:
# If resolution fails, keep the string annotation
logger.debug("Failed to resolve type hint for return annotation: %s", e)
# Save original for return_type before any schema-related replacement
original_output_type = output_type
if output_type not in (inspect._empty, None, Any, ...):
# Prefab component subclasses (Column, Card, etc.) shouldn't
# produce output schemas — replace_type only does exact matching,
# so we handle subclass matching explicitly here.
if (
_PREFAB_TYPES
and isinstance(output_type, type)
and issubclass(output_type, _PREFAB_TYPES)
):
output_type = _UnserializableType
# there are a variety of types that we don't want to attempt to
# serialize because they are either used by FastMCP internally,
# or are MCP content types that explicitly don't form structured
@ -164,6 +186,7 @@ class ParsedFunction:
mcp.types.AudioContent,
mcp.types.ResourceLink,
mcp.types.EmbeddedResource,
*_PREFAB_TYPES,
),
_UnserializableType,
),
@ -198,4 +221,5 @@ class ParsedFunction:
description=fn_doc,
input_schema=input_schema,
output_schema=output_schema or None,
return_type=original_output_type,
)

View file

@ -8,6 +8,7 @@ from collections.abc import Callable
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Annotated,
Any,
Literal,
Protocol,
@ -20,6 +21,7 @@ import anyio
import mcp.types
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData, Icon, ToolAnnotations, ToolExecution
from pydantic import Field
from pydantic.json_schema import SkipJsonSchema
import fastmcp
@ -84,6 +86,7 @@ class ToolMeta:
class FunctionTool(Tool):
fn: SkipJsonSchema[Callable[..., Any]]
return_type: Annotated[SkipJsonSchema[Any], Field(exclude=True)] = None
def to_mcp_tool(
self,
@ -230,6 +233,7 @@ class FunctionTool(Tool):
return cls(
fn=parsed_fn.fn,
return_type=parsed_fn.return_type,
name=metadata.name or parsed_fn.name,
version=str(metadata.version) if metadata.version is not None else None,
title=metadata.title,

View file

@ -38,6 +38,18 @@ from fastmcp.utilities.types import (
NotSetT,
)
try:
from prefab_ui.components.base import Component as _PrefabComponent
from prefab_ui.response import UIResponse as _PrefabUIResponse
_HAS_PREFAB = True
except ImportError:
_HAS_PREFAB = False
# Runtime type alias for auth checks to avoid circular imports with authorization.py
# AuthCheck is Callable[[AuthContext], bool] but we use Any to avoid the import
AuthCheckCallable: TypeAlias = Callable[[Any], bool]
if TYPE_CHECKING:
from docket import Docket
from docket.execution import Execution
@ -248,6 +260,12 @@ class Tool(FastMCPComponent):
if isinstance(raw_value, ToolResult):
return raw_value
if _HAS_PREFAB:
if isinstance(raw_value, _PrefabUIResponse):
return _ui_response_to_tool_result(raw_value)
if isinstance(raw_value, _PrefabComponent):
return _ui_response_to_tool_result(_PrefabUIResponse(view=raw_value))
content = _convert_to_content(raw_value, serializer=self.serializer)
# Skip structured content for ContentBlock types only if no output_schema
@ -437,6 +455,16 @@ def _convert_to_single_content_block(
return TextContent(type="text", text=_serialize_with_fallback(item, serializer))
def _ui_response_to_tool_result(response: Any) -> ToolResult:
"""Convert a prefab UIResponse to a FastMCP ToolResult."""
text = response.text_fallback()
envelope = response.to_json()
return ToolResult(
content=[TextContent(type="text", text=text)],
structured_content=envelope,
)
def _convert_to_content(
result: Any,
serializer: ToolResultSerializerType | None = None,

335
tests/test_apps_prefab.py Normal file
View file

@ -0,0 +1,335 @@
"""Tests for MCP Apps Phase 2 — Prefab integration.
Covers ``convert_result`` for UIResponse/Component, ``app=True`` auto-wiring,
return type inference, output schema suppression, and end-to-end round trips.
"""
from __future__ import annotations
from mcp.types import TextContent
from prefab_ui.components import Column, Heading, Text
from prefab_ui.components.base import Component
from prefab_ui.response import UIResponse
from fastmcp import Client, FastMCP
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,
)
from fastmcp.tools.tool import Tool, ToolResult
# ---------------------------------------------------------------------------
# convert_result
# ---------------------------------------------------------------------------
class TestConvertResult:
def test_ui_response(self):
with Column() as view:
Heading("Hello")
response = UIResponse(view=view, state={"name": "Alice"})
tool = Tool(name="t", parameters={})
result = tool.convert_result(response)
assert isinstance(result, ToolResult)
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == response.text_fallback()
assert result.structured_content is not None
assert result.structured_content["version"] == "0.2"
assert result.structured_content["state"] == {"name": "Alice"}
assert result.structured_content["view"]["type"] == "Column"
def test_bare_component(self):
heading = Heading("World")
tool = Tool(name="t", parameters={})
result = tool.convert_result(heading)
assert isinstance(result, ToolResult)
assert result.structured_content is not None
assert result.structured_content["version"] == "0.2"
assert result.structured_content["view"]["type"] == "Heading"
def test_custom_text_fallback(self):
response = UIResponse(view=Heading("Hello"), text="Custom fallback text")
tool = Tool(name="t", parameters={})
result = tool.convert_result(response)
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Custom fallback text"
def test_tool_result_passthrough(self):
"""ToolResult should still pass through unchanged."""
original = ToolResult(content="hello")
tool = Tool(name="t", parameters={})
assert tool.convert_result(original) is original
# ---------------------------------------------------------------------------
# app=True auto-wiring
# ---------------------------------------------------------------------------
class TestAppTrue:
def test_app_true_sets_meta(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> str:
return "hello"
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta is not None
assert "ui" in tool.meta
assert tool.meta["ui"]["resourceUri"] == PREFAB_RENDERER_URI
def test_app_true_registers_renderer_resource(self):
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
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
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"]
def test_multiple_tools_share_renderer(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def tool_a() -> str:
return "a"
@mcp.tool(app=True)
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
def test_explicit_app_config_not_overridden(self):
mcp = FastMCP("test")
@mcp.tool(app=AppConfig(resource_uri="ui://custom/app.html"))
def my_tool() -> UIResponse:
return UIResponse(view=Heading("hi"))
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta["ui"]["resourceUri"] == "ui://custom/app.html"
# ---------------------------------------------------------------------------
# Return type inference
# ---------------------------------------------------------------------------
class TestInference:
def test_ui_response_annotation_inferred(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> UIResponse:
return UIResponse(view=Heading("hi"))
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta is not None
assert tool.meta["ui"]["resourceUri"] == PREFAB_RENDERER_URI
def test_component_annotation_inferred(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> Component:
return Heading("hi")
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta is not None
assert tool.meta["ui"]["resourceUri"] == PREFAB_RENDERER_URI
def test_no_annotation_no_inference(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool():
return "hello"
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta is None or "ui" not in (tool.meta or {})
def test_non_prefab_annotation_no_inference(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> str:
return "hello"
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta is None or "ui" not in (tool.meta or {})
def test_optional_ui_response_inferred(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> UIResponse | None:
return None
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.meta is not None
assert tool.meta["ui"]["resourceUri"] == PREFAB_RENDERER_URI
# ---------------------------------------------------------------------------
# Output schema suppression
# ---------------------------------------------------------------------------
class TestOutputSchema:
def test_ui_response_return_no_output_schema(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> UIResponse:
return UIResponse(view=Heading("hi"))
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.output_schema is None
def test_component_return_no_output_schema(self):
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> Column:
with Column() as view:
Heading("hi")
return view
tools = mcp._local_provider._components
tool = next(
v
for v in tools.values()
if hasattr(v, "parameters") and v.name == "my_tool"
)
assert tool.output_schema is None
# ---------------------------------------------------------------------------
# Integration — client-server round trip
# ---------------------------------------------------------------------------
class TestIntegration:
async def test_tool_call_returns_prefab_structured_content(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def greet(name: str) -> UIResponse:
with Column() as view:
Heading("Hello")
Text(f"Welcome, {name}!")
return UIResponse(view=view, state={"name": name})
async with Client(mcp) as client:
result = await client.call_tool("greet", {"name": "Alice"})
assert result.structured_content is not None
assert result.structured_content["version"] == "0.2"
assert result.structured_content["state"] == {"name": "Alice"}
async def test_tools_list_includes_app_meta(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> UIResponse:
return UIResponse(view=Heading("hi"))
async with Client(mcp) as client:
tools = await client.list_tools()
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
async def test_renderer_resource_readable(self):
mcp = FastMCP("test")
@mcp.tool(app=True)
def my_tool() -> str:
return "hello"
async with Client(mcp) as client:
contents = await client.read_resource(PREFAB_RENDERER_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()

39
uv.lock generated
View file

@ -736,6 +736,9 @@ dependencies = [
anthropic = [
{ name = "anthropic" },
]
apps = [
{ name = "prefab-ui" },
]
azure = [
{ name = "azure-identity" },
]
@ -750,7 +753,7 @@ tasks = [
dev = [
{ name = "dirty-equals" },
{ name = "fastapi" },
{ name = "fastmcp", extra = ["anthropic", "azure", "openai", "tasks"] },
{ name = "fastmcp", extra = ["anthropic", "apps", "azure", "openai", "tasks"] },
{ name = "inline-snapshot", extra = ["dirty-equals"] },
{ name = "ipython", version = "8.38.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
{ name = "ipython", version = "9.9.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
@ -792,6 +795,7 @@ requires-dist = [
{ name = "opentelemetry-api", specifier = ">=1.20.0" },
{ name = "packaging", specifier = ">=24.0" },
{ name = "platformdirs", specifier = ">=4.0.0" },
{ name = "prefab-ui", marker = "extra == 'apps'", editable = "../prefab" },
{ name = "py-key-value-aio", extras = ["disk", "keyring", "memory"], specifier = ">=0.4.0,<0.5.0" },
{ name = "pydantic", extras = ["email"], specifier = ">=2.11.7" },
{ name = "pydocket", marker = "extra == 'tasks'", specifier = ">=0.17.2" },
@ -803,13 +807,13 @@ requires-dist = [
{ name = "watchfiles", specifier = ">=1.0.0" },
{ name = "websockets", specifier = ">=15.0.1" },
]
provides-extras = ["anthropic", "azure", "openai", "tasks"]
provides-extras = ["anthropic", "apps", "azure", "openai", "tasks"]
[package.metadata.requires-dev]
dev = [
{ name = "dirty-equals", specifier = ">=0.9.0" },
{ name = "fastapi", specifier = ">=0.115.12" },
{ name = "fastmcp", extras = ["anthropic", "azure", "openai", "tasks"] },
{ name = "fastmcp", extras = ["anthropic", "apps", "azure", "openai", "tasks"] },
{ name = "inline-snapshot", extras = ["dirty-equals"], specifier = ">=0.27.2" },
{ name = "ipython", specifier = ">=8.12.3" },
{ name = "loq", specifier = ">=0.1.0a3" },
@ -1664,6 +1668,35 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]]
name = "prefab-ui"
source = { editable = "../prefab" }
dependencies = [
{ name = "cyclopts" },
{ name = "pydantic" },
{ name = "rich" },
]
[package.metadata]
requires-dist = [
{ name = "cyclopts", specifier = ">=4" },
{ name = "pydantic", specifier = ">=2.10" },
{ name = "rich", specifier = ">=13" },
]
[package.metadata.requires-dev]
dev = [
{ name = "jsonschema", specifier = ">=4.26.0" },
{ name = "loq", specifier = ">=0.1.0a3" },
{ name = "prek", specifier = ">=0.1.0" },
{ name = "pytest", specifier = ">=8.3.3" },
{ name = "pytest-asyncio", specifier = ">=1.2.0" },
{ name = "pytest-timeout", specifier = ">=2.4.0" },
{ name = "pytest-xdist", specifier = ">=3.6.1" },
{ name = "ruff", specifier = ">=0.12.8" },
{ name = "ty", specifier = ">=0.0.7" },
]
[[package]]
name = "prek"
version = "0.3.0"