diff --git a/server.py b/server.py new file mode 100644 index 0000000..25f7127 --- /dev/null +++ b/server.py @@ -0,0 +1,148 @@ +"""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="Owner control plane for Phast app manifests. Mutations fail closed until owner auth is 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="apps_list", annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True}) +def apps_list() -> dict: + return {"apps": _require_registry().list()} + +@mcp.tool(name="apps_get", annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True}) +def apps_get(slug: str) -> dict: + result = _require_registry().get(slug) + if result is None: + raise KeyError(f"unknown app: {slug}") + return result + +@mcp.tool(name="apps_create", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": False}) +def apps_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="apps_update", annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True}) +def apps_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) + +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=True) + 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=True) + 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")))