"""Dev server for previewing FastMCPApp UIs locally.
Starts the user's MCP server on a configurable port, then starts a lightweight
Starlette dev server that:
- Serves a Prefab-based tool picker at GET /
- Proxies /mcp to the user's server (avoids browser CORS restrictions)
- Serves the AppBridge host page at GET /launch
The host page uses @modelcontextprotocol/ext-apps to connect to the MCP server
and render the selected UI tool inside an iframe.
Startup sequence
----------------
1. Download ext-apps app-bridge.js from npm and patch its bare
``@modelcontextprotocol/sdk/…`` imports to use concrete esm.sh URLs.
2. Detect the exact Zod v4 module URL that esm.sh serves for that SDK version
and build an import-map entry that redirects the broken ``v4.mjs`` (which
only re-exports ``{z, default}``) to ``v4/classic/index.mjs`` (which
correctly exports every named Zod v4 function). Import maps apply to the
full module graph in the document, including cross-origin esm.sh modules.
3. Serve both the patched JS and the import-map JSON from the dev server.
"""
from __future__ import annotations
import asyncio
import contextlib
import io
import json
import logging
import os
import re
import signal
import sys
import tarfile
import tempfile
import time
import webbrowser
from pathlib import Path
from typing import Any
from urllib.parse import quote
import httpcore
import httpx
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import HTMLResponse, Response, StreamingResponse
from starlette.routing import Route
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
# ---------------------------------------------------------------------------
# MCP message log (captures proxy traffic for the dev UI log panel)
# ---------------------------------------------------------------------------
class _MessageLog:
"""In-memory buffer of MCP JSON-RPC messages flowing through the proxy."""
def __init__(self) -> None:
self._entries: list[dict[str, Any]] = []
self._counter = 0
self._request_methods: dict[int | str, str] = {}
self._request_times: dict[int | str, float] = {}
def log_request(self, body: dict[str, Any]) -> None:
method = body.get("method", "unknown")
jsonrpc_id = body.get("id")
timestamp = time.time()
if jsonrpc_id is not None:
self._request_methods[jsonrpc_id] = method
self._request_times[jsonrpc_id] = timestamp
self._counter += 1
self._entries.append(
{
"id": self._counter,
"timestamp": timestamp,
"direction": "request",
"method": method,
"body": body,
}
)
def log_response(self, body: dict[str, Any]) -> None:
# Server-initiated notifications have "method" but no "id"
if "method" in body and "id" not in body:
self._counter += 1
self._entries.append(
{
"id": self._counter,
"timestamp": time.time(),
"direction": "notification",
"method": body.get("method", "unknown"),
"body": body,
}
)
return
jsonrpc_id = body.get("id")
method = (
self._request_methods.pop(jsonrpc_id, None)
if jsonrpc_id is not None
else None
)
request_time = (
self._request_times.pop(jsonrpc_id, None)
if jsonrpc_id is not None
else None
)
timestamp = time.time()
duration_ms = (
round((timestamp - request_time) * 1000, 1) if request_time else None
)
self._counter += 1
self._entries.append(
{
"id": self._counter,
"timestamp": timestamp,
"direction": "response",
"method": method,
"body": body,
"duration_ms": duration_ms,
}
)
def get_since(self, since_id: int = 0) -> list[dict[str, Any]]:
return [e for e in self._entries if e["id"] > since_id]
def log_bridge(self, body: dict[str, Any]) -> None:
method = body.get("method", "unknown")
self._counter += 1
self._entries.append(
{
"id": self._counter,
"timestamp": time.time(),
"direction": "bridge",
"method": method,
"body": body,
}
)
def clear(self) -> None:
self._entries.clear()
self._request_methods.clear()
self._request_times.clear()
def _log_response_bytes(log: _MessageLog, raw: bytes, content_type: str) -> None:
"""Parse accumulated proxy response bytes and log as message entries."""
if not raw:
return
try:
if "text/event-stream" in content_type:
for line in raw.decode("utf-8", errors="replace").splitlines():
if line.startswith("data: "):
with contextlib.suppress(json.JSONDecodeError):
log.log_response(json.loads(line[6:]))
else:
body = json.loads(raw)
if isinstance(body, list):
for item in body:
log.log_response(item)
else:
log.log_response(body)
except (json.JSONDecodeError, TypeError):
pass
_EXT_APPS_VERSION = "1.0.1"
# Pin to the SDK version ext-apps 1.0.1 was compiled against so the client
# and transport modules are API-compatible with the app-bridge internals.
_MCP_SDK_VERSION = "1.25.2"
# ---------------------------------------------------------------------------
# Shared AppBridge host shell
# ---------------------------------------------------------------------------
# Both the picker and the app launcher use the same host-page structure: an
# iframe that hosts a Prefab renderer, wired to the MCP server via AppBridge.
# The only differences are (a) which URL loads in the iframe and (b) what
# oninitialized does.
#
# app-bridge.js is served locally (see _fetch_app_bridge_bundle).
# Client/Transport are loaded from esm.sh.
# The import map (injected as {import_map_tag}) patches the broken esm.sh
# Zod v4 module so all Zod named exports are visible to the SDK at runtime.
_HOST_SHELL = """\
{title}
{import_map_tag}
{status_text}
"""
# ---------------------------------------------------------------------------
# Host page HTML
# ---------------------------------------------------------------------------
_HOST_HTML_TEMPLATE = """\
FastMCP Dev — {tool_name}
{import_map_tag}
Launching {tool_name}…
"""
# ---------------------------------------------------------------------------
# Dev log panel (injected into host pages)
# ---------------------------------------------------------------------------
_LOG_PANEL_HTML = """\
FastMCP Apps\u00b7 0
Show
"""
def _inject_log_panel(html: str) -> str:
"""Inject the MCP message log panel before
."""
return html.replace("", _LOG_PANEL_HTML + "\n")
# ---------------------------------------------------------------------------
# Picker UI (Prefab-based, built in Python)
# ---------------------------------------------------------------------------
def _has_ui_resource(tool: dict[str, Any]) -> bool:
"""Return True if the tool has a UI resourceUri in its metadata."""
for key in ("meta", "_meta"):
m = tool.get(key)
if isinstance(m, dict):
ui = m.get("ui")
if isinstance(ui, dict) and ui.get("resourceUri"):
return True
return False
def _model_from_schema(tool_name: str, input_schema: dict[str, Any]) -> type[Any]:
"""Dynamically create a Pydantic model from a JSON Schema for form generation."""
import pydantic
import pydantic.fields
properties: dict[str, Any] = input_schema.get("properties") or {}
required: list[str] = input_schema.get("required") or []
field_definitions: dict[str, Any] = {}
for prop_name, prop in properties.items():
json_type = prop.get("type", "string")
# Handle anyOf / oneOf (union types like str | dict | None)
for key in ("anyOf", "oneOf"):
if key in prop:
non_null = [
t
for t in prop[key]
if isinstance(t, dict) and t.get("type") != "null"
]
if non_null:
types = [t.get("type") for t in non_null if "type" in t]
# Prefer object/array (need textarea for JSON editing),
# then string (most versatile text input), then scalars.
for candidate in (
"object",
"array",
"string",
"integer",
"number",
"boolean",
):
if candidate in types:
json_type = candidate
break
break
match json_type:
case "integer":
py_type: type = int
case "number":
py_type = float
case "boolean":
py_type = bool
case "object" | "array":
# Render as a string textarea; api_launch parses JSON later
py_type = str
case _:
py_type = str
title = prop.get("title") or prop_name.replace("_", " ").title()
description = prop.get("description")
is_required = prop_name in required
if is_required:
default = pydantic.fields.PydanticUndefined
elif "default" in prop:
default = prop["default"]
else:
default = None
py_type = py_type | None # type: ignore[assignment] # ty:ignore[invalid-assignment]
extra: dict[str, Any] = {}
if prop.get("enum"):
from typing import Literal
py_type = Literal[tuple(prop["enum"])] # type: ignore[assignment] # ty:ignore[invalid-type-form]
# Textarea detection:
# 1. Explicit format: "textarea" in JSON schema
# 2. UI annotation: {"ui": {"type": "textarea"}} (json_schema_extra merged flat)
# 3. Object/array types need multiline JSON editing
use_textarea = (
prop.get("format") == "textarea"
or (
isinstance(prop.get("ui"), dict)
and prop["ui"].get("type") == "textarea"
)
or json_type in ("object", "array")
)
if use_textarea:
extra["json_schema_extra"] = {"ui": {"type": "textarea"}}
field_definitions[prop_name] = (
py_type,
pydantic.Field(
default=default, title=title, description=description, **extra
),
)
return pydantic.create_model(f"{tool_name.title()}Form", **field_definitions)
def _build_picker_html(tools: list[dict[str, Any]]) -> str:
"""Build Prefab picker page: dropdown selector with per-tool forms."""
try:
from prefab_ui.actions import Fetch, OpenLink, SetState, ShowToast
from prefab_ui.app import PrefabApp
from prefab_ui.components import (
Button,
Column,
Heading,
Label,
Markdown,
Muted,
Page,
Pages,
Select,
SelectOption,
Textarea,
)
from prefab_ui.components.form import Form
from prefab_ui.rx import RESULT, Rx
except ImportError:
return "