Import canonical Phast control-plane source
This commit is contained in:
matcha committed
1 parent
0c4874a087
commit
eccfaaa7c3
1 file changed
+148
@@ -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")))
|
||||
Reference in new issue
Block a user