app/server.py

"""Phast public landing and fail-closed owner control plane."""
from __future__ import annotations

import os
from pathlib import Path

import uvicorn
from fastmcp import FastMCP
from prefab_ui.app import PrefabApp
from prefab_ui.components import Badge, Button, Card, Code, Column, Heading, Row, Separator, Text
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse

try:
    from .registry import AppRegistry, RegistryUnavailable, registry_from_runtime
except ImportError:
    from registry import AppRegistry, RegistryUnavailable, registry_from_runtime

OWNER_HOST = "my.phast.app"
AUTH_BLOCKER = "Owner authentication is not configured; app mutations are disabled."

try:
    registry: AppRegistry | None = registry_from_runtime()
    registry_blocker: str | None = None
except RegistryUnavailable as error:
    registry = None
    registry_blocker = str(error)

mcp = FastMCP(
    "My Phast",
    instructions="Management MCP for portable Phast tool belts and Prefab homes. Mutations fail closed until owner auth and durable provisioning are installed.",
)

def _require_registry() -> AppRegistry:
    if registry is None:
        raise RegistryUnavailable(registry_blocker or "durable registry is offline")
    return registry

def _require_owner_auth() -> None:
    raise PermissionError(AUTH_BLOCKER)

@mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True})
def phast_status() -> dict:
    return {
        "ok": True,
        "service": "phast",
        "url": "https://phast.app",
        "owner_url": "https://my.phast.app",
        "mcp": "https://my.phast.app/mcp",
        "registry_ready": registry is not None,
        "registry_blocker": registry_blocker,
        "mutations_ready": False,
        "auth_blocker": AUTH_BLOCKER,
    }

@mcp.tool(name="phasts_list", annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True})
def phasts_list() -> dict:
    return {"apps": _require_registry().list()}

@mcp.tool(name="phasts_get", annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True})
def phasts_get(slug: str) -> dict:
    result = _require_registry().get(slug)
    if result is None:
        raise KeyError(f"unknown app: {slug}")
    return result

@mcp.tool(name="phasts_create", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": False})
def phasts_create(slug: str, name: str, base_url: str, human_route: str = "/", mcp_route: str = "/mcp") -> dict:
    _require_owner_auth()
    return _require_registry().create(slug=slug, name=name, base_url=base_url, human_route=human_route, mcp_route=mcp_route)

@mcp.tool(name="phasts_update", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True})
def phasts_update(slug: str, name: str | None = None, base_url: str | None = None,
                human_route: str | None = None, mcp_route: str | None = None) -> dict:
    _require_owner_auth()
    return _require_registry().update(slug, name=name, base_url=base_url, human_route=human_route, mcp_route=mcp_route)

@mcp.tool(name="providers_search", annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True})
def providers_search(query: str) -> dict:
    """V0 seam: provider discovery belongs to Oracle, not a hard-coded Phast catalog."""
    return {"query": query, "source": "oracle", "status": "discovery-route-pending"}

@mcp.tool(name="widgets_list", annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True})
def widgets_list(slug: str) -> dict:
    item = phasts_get(slug)
    return {"phast": slug, "widgets": item.get("prefab", {}).get("home", {}).get("widgets", [])}

@mcp.tool(name="widgets_add", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": False})
def widgets_add(slug: str, widget: dict) -> dict:
    _require_owner_auth()
    raise NotImplementedError("durable phast.json widget persistence is not installed yet")

@mcp.tool(name="widgets_update", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True})
def widgets_update(slug: str, widget_id: str, widget: dict) -> dict:
    _require_owner_auth()
    raise NotImplementedError("durable phast.json widget persistence is not installed yet")

@mcp.tool(name="widgets_remove", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True})
def widgets_remove(slug: str, widget_id: str) -> dict:
    _require_owner_auth()
    raise NotImplementedError("durable phast.json widget persistence is not installed yet")

def control_app() -> PrefabApp:
    apps = registry.list() if registry is not None else []
    with PrefabApp(title="My Phast", css_class="max-w-4xl mx-auto p-6", mode="light") as result:
        with Column(gap=5):
            with Row(gap=3, justify="between", align="center"):
                with Column(gap=1):
                    Text("PHAST · OWNER CONTROL PLANE", code=[REDACTED]
                    Heading("My Phast", level=1)
                Badge("read-only" if registry is not None else "setup required", variant="warning")
            Text("Your apps share one boring convention: a human surface at / and an agent surface at /mcp.")
            with Card(css_class="p-5"):
                with Column(gap=3):
                    with Row(gap=2, justify="between", align="center"):
                        Heading("Apps", level=2)
                        Badge(str(len(apps)), variant="secondary")
                    if apps:
                        for item in apps:
                            with Card(css_class="p-4"):
                                with Row(gap=3, justify="between", align="center"):
                                    with Column(gap=1):
                                        Heading(item["name"], level=3)
                                        Text(item["base_url"], code=[REDACTED]
                                    Badge(item["runtime"], variant="outline")
                                Text(f"human {item['human']['route']} · mcp {item['mcp']['route']}")
                    else:
                        Text("No durable app registry is available yet.")
            with Card(css_class="p-5"):
                with Column(gap=3):
                    Heading("Control-plane readiness", level=2)
                    Text(registry_blocker or "Durable SQLite registry is mounted and readable.")
                    Text(AUTH_BLOCKER)
                    Button("Create app", disabled=True)
            Separator()
            Text("Agent interface", bold=True)
            Code("https://my.phast.app/mcp", language="text")
            Text("Mutations stay disabled until both a Fly volume and accepted owner authentication exist.")
    return result

@mcp.tool(name="my_phast_app", app=True, annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True})
def my_phast_app() -> PrefabApp:
    return control_app()

@mcp.custom_route("/", methods=["GET"])
async def homepage(request: Request):
    host = request.headers.get("host", "").split(":", 1)[0].lower()
    if host == OWNER_HOST:
        return HTMLResponse(control_app().html(renderer_mode="bundled"))
    return HTMLResponse(Path(__file__).with_name("index.html").read_text())

@mcp.custom_route("/healthz", methods=["GET"])
async def health(_: Request):
    return JSONResponse({
        "ok": True,
        "service": "phast",
        "source_sha": os.environ.get("SOURCE_SHA", "local"),
        "control_plane": {
            "registry_ready": registry is not None,
            "mutations_ready": False,
            "blockers": [value for value in (registry_blocker, AUTH_BLOCKER) if value],
        },
    })

app = mcp.http_app(
    path="/mcp",
    transport="streamable-http",
    stateless_http=True,
    allowed_hosts=["phast.app", "www.phast.app", "my.phast.app", "localhost", "testserver"],
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))