from __future__ import annotations import asyncio import json import os import re import time from pathlib import Path from typing import Any import httpx import uvicorn from fastmcp import FastMCP from fastmcp.server.auth.oauth_proxy import OAuthProxy from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier from fastmcp.server.dependencies import get_access_token from prefab_ui.app import PrefabApp from prefab_ui.components import Badge, Card, Column, Heading, Row, Text from starlette.responses import FileResponse, HTMLResponse, JSONResponse, Response DATA = Path(os.getenv("PHAST_DATA_DIR", "/data/phast")) # The public pages are served by this app too: one source, no copies to drift. HERE = Path(__file__).resolve().parent PAGES = {"phast.app": HERE / "index.html", "my.phast.app": HERE / "my" / "index.html"} DATA.mkdir(parents=True, exist_ok=True) GITEA = os.getenv("GITEA_URL", "https://teahub.xyz").rstrip("/") CID = os.environ["GITEA_OAUTH_CLIENT_ID"] SECRET = os.environ["GITEA_OAUTH_CLIENT_SECRET"] JWT = os.environ["JWT_SIGNING_KEY"] GITEA_TOKEN = os.getenv("GITEA_TOKEN", "") USERS = {x.strip() for x in os.getenv("ALLOWED_GITEA_USERS", "matcha,chai").split(",") if x.strip()} RX = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$") # Oracle is where providers come from: the capability index's search projection, fetched and cached. # Phast never holds a copy of the graph, only the one file it needs to find something to connect to. ORACLE = os.getenv("ORACLE_URL", "https://oracle.teahub.xyz").rstrip("/") ORACLE_TTL = int(os.getenv("ORACLE_TTL_SECONDS", "3600")) ORACLE_CACHE = DATA / "_oracle-search.json" LADDER = ["claimed", "reachable", "tested", "verified", "transacted", "repeated"] _oracle: dict[str, Any] = {"rows": [], "at": 0.0, "fresh": ""} def oracle_rows() -> list[dict[str, Any]]: """The search index as dicts, refreshed at most every ORACLE_TTL seconds; the last good copy (memory, then the volume) is used when Oracle can't be reached.""" now = time.time() if _oracle["rows"] and now - _oracle["at"] < ORACLE_TTL: return _oracle["rows"] text, fresh = "", "" try: response = httpx.get(ORACLE + "/search.json", timeout=30, headers={"User-Agent": "phast-app/0.2"}) if response.status_code == 200: text, fresh = response.text, response.headers.get("last-modified", "") try: ORACLE_CACHE.write_text(text) except OSError: pass except httpx.HTTPError: pass if not text and not _oracle["rows"] and ORACLE_CACHE.exists(): text, fresh = ORACLE_CACHE.read_text(), "cache" if text: payload = json.loads(text) columns = payload["columns"] _oracle["rows"] = [dict(zip(columns, row)) for row in payload["rows"]] _oracle["fresh"] = fresh _oracle["at"] = now return _oracle["rows"] def oracle_find(capability_id: str) -> dict[str, Any]: for row in oracle_rows(): if row["id"] == capability_id: return row raise KeyError("unknown capability") def oracle_search(query: str, limit: int = 10, connectable_only: bool = True) -> list[dict[str, Any]]: terms = [t for t in re.split(r"[^a-z0-9]+", query.lower()) if t] if not terms: return [] hits = [] for row in oracle_rows(): if connectable_only and not row["remote_url"]: continue hay = " ".join((row["name"], row["provider"], row["summary"], row["id"])).lower() if not all(t in hay for t in terms): continue name = row["name"].lower() score = ( LADDER.index(row["evidence_state"]) if row["evidence_state"] in LADDER else 0, 1 if row["remote_url"] else 0, sum(1 for t in terms if t in name), 1 if "teahub" in row["sources"] else 0, ) hits.append((score, row)) hits.sort(key=lambda item: item[1]["name"].lower()) hits.sort(key=lambda item: item[0], reverse=True) return [row for _, row in hits[: max(1, min(limit, 50))]] class V(IntrospectionTokenVerifier): async def verify_token(self, token): access = await super().verify_token(token) if not access: return None async with httpx.AsyncClient(timeout=10) as client: response = await client.get( GITEA + "/login/oauth/userinfo", headers={"Authorization": "Bearer " + token}, ) if response.status_code != 200: return None login = str(response.json().get("preferred_username") or "") if login not in USERS: return None access.claims = {**(access.claims or {}), "login": login} return access def auth(base: str, resource: str | None = None) -> OAuthProxy: verifier = V( introspection_url=GITEA + "/login/oauth/introspect", client_id=CID, client_secret=SECRET, client_auth_method="client_secret_basic", required_scopes=["openid"], ) return OAuthProxy( upstream_authorization_endpoint=GITEA + "/login/oauth/authorize", upstream_token_endpoint=GITEA + "/login/oauth/access_token", upstream_client_id=CID, upstream_client_secret=SECRET, token_verifier=verifier, base_url=base, resource_base_url=resource or base, valid_scopes=["openid", "profile", "email"], forward_pkce=True, forward_resource=False, token_endpoint_auth_method="client_secret_basic", jwt_signing_key=JWT, require_authorization_consent="remember", enable_cimd=True, ) def actor() -> str: access = get_access_token() login = str(((access.claims if access else {}) or {}).get("login") or "") if login not in USERS: raise RuntimeError("actor not enabled") return login def slug(value: str) -> str: value = value.strip().lower() if not RX.fullmatch(value) or value in {"my", "www", "mcp"}: raise ValueError("invalid slug") return value def path(value: str) -> Path: return DATA / slug(value) / "phast.json" def read(value: str) -> dict[str, Any]: p = path(value) if not p.exists(): raise KeyError("unknown Phast app") return json.loads(p.read_text()) def write(value: str, data: dict[str, Any]) -> dict[str, Any]: p = path(value) p.parent.mkdir(parents=True, exist_ok=True) try: p.parent.chmod(0o700) except OSError: pass tmp = p.with_suffix(".tmp") tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n") tmp.chmod(0o600) tmp.replace(p) try: p.chmod(0o600) except OSError: pass return data def mine(value: str) -> dict[str, Any]: data = read(value) if data.get("owner") != actor(): raise PermissionError("not your Phast app") return data def create(value: str, name: str = "") -> dict[str, Any]: value = slug(value) if path(value).exists(): raise ValueError("already exists") # Provision auth before durable state so a failed OAuth mutation cannot leave # a half-created personal MCP behind. return write( value, { "schema_version": 1, "kind": "phast", "slug": value, "name": name.strip() or value, "owner": actor(), "host": value + ".phast.app", "surfaces": {"human": "/", "mcp": "/mcp"}, "tools": {"installed": []}, "connections": [], "prefab": {"version": 1, "home": {"mode": "grid", "edit": True, "widgets": []}}, }, ) def prefab(data: dict[str, Any]) -> PrefabApp: widgets = data["prefab"]["home"]["widgets"] with PrefabApp(title=data["name"], css_class="max-w-4xl mx-auto p-6", mode="light") as app: with Column(gap=5): with Row(justify="between", align="center"): Heading(data["name"], level=1) Badge(str(len(data["tools"]["installed"])) + " tools", variant="secondary") Text("https://" + data["host"] + "/mcp", code=True) for tool in data["tools"]["installed"]: with Card(css_class="p-5"): with Row(justify="between", align="center"): Heading(str(tool["name"]), level=2) Badge(str(tool.get("evidence_state", "claimed")), variant="secondary") Text(str(tool.get("provider", "")) + " · " + str(tool.get("remote_url", ""))) if not widgets: with Card(css_class="p-5"): Heading("Blank home", level=2) Text("Add a provider or widget from Chat or my.phast.app.") for widget in widgets: with Card(css_class="p-5"): Heading(str(widget.get("title") or widget["id"]), level=2) Text(str(widget.get("description", ""))) return app def human(data: dict[str, Any]) -> str: widgets = data["prefab"]["home"]["widgets"] belt = "".join( "

" + str(tool["name"]) + " " + str(tool.get("evidence_state", "claimed")) + "

" + str(tool.get("provider", "")) + "

" + str(tool.get("remote_url", "")) + "

" for tool in data["tools"]["installed"] ) cards = "".join( "

" + str(widget.get("title") or widget["id"]) + "

" + str(widget.get("description", "")) + "

" for widget in widgets ) or "

Blank home

Add a widget from Chat or my.phast.app.

" return ( '' + "" + data["name"] + " - Phast app" + "" + "
Phast app." + data["host"] + "/mcp

" + data["name"] + "

Tool belt · " + str(len(data["tools"]["installed"])) + "

" + (belt or "

Empty belt

Ask your agent: search providers, then install one.

") + "

Home

" + cards + "
" ) m = FastMCP( "My Phast app", version="0.2.0", instructions="Create and manage portable Phast app MCP tool belts and Prefab homes.", auth=auth("https://my.phast.app"), mask_error_details=True, ) @m.tool() def phast_status() -> dict: return {"ok": True, "actor": actor(), "version": "0.2.1", "oracle": ORACLE, "oracle_capabilities": len(_oracle["rows"]), "oracle_fresh": _oracle["fresh"]} @m.tool() def phasts_list() -> dict: current = actor() rows = [] for p in DATA.glob("*/phast.json"): try: data = json.loads(p.read_text()) except Exception: continue if data.get("owner") == current: rows.append(data) return {"phasts": sorted(rows, key=lambda item: item["slug"])} @m.tool() def phasts_get(slug: str) -> dict: return mine(slug) @m.tool() def phasts_create(slug: str, name: str = "") -> dict: data = create(slug, name) return { "phast": data, "human_url": "https://" + data["host"] + "/", "mcp_url": "https://" + data["host"] + "/mcp", } @m.tool() def phasts_update(slug: str, name: str) -> dict: data = mine(slug) clean = name.strip() if not clean: raise ValueError("name is required") data["name"] = clean return write(slug, data) @m.tool() def providers_search(query: str, limit: int = 10, connectable_only: bool = True) -> dict: """Find providers in Oracle, the capability index (oracle.teahub.xyz). Results carry the evidence state on Oracle's ladder (claimed < reachable < tested < verified < transacted < repeated); a registry listing is only a claim. connectable_only keeps results with a remote MCP endpoint.""" actor() rows = oracle_search(query, limit, connectable_only) return {"query": query, "source": ORACLE, "count": len(rows), "results": rows} @m.tool() def tools_list(slug: str) -> dict: return {"tools": list(mine(slug)["tools"]["installed"])} @m.tool() def tools_install(slug: str, capability_id: str) -> dict: """Put an Oracle capability on this Phast app's belt. The belt holds a reference to the capability and its endpoint, never a credential; connecting through it is the next step.""" data = mine(slug) installed = data["tools"]["installed"] if any(tool["id"] == capability_id for tool in installed): raise ValueError("already on the belt") row = oracle_find(capability_id) if not row["remote_url"]: raise ValueError("no remote endpoint to connect to") tool = { "id": row["id"], "name": row["name"], "provider": row["provider"], "remote_url": row["remote_url"], "remote_type": row["remote_type"], "evidence_state": row["evidence_state"], "sources": row["sources"], "source": "oracle", "installed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } installed.append(tool) write(slug, data) return tool @m.tool() def tools_remove(slug: str, capability_id: str) -> dict: data = mine(slug) installed = data["tools"]["installed"] kept = [tool for tool in installed if tool["id"] != capability_id] if len(kept) == len(installed): raise KeyError("not on the belt") data["tools"]["installed"] = kept write(slug, data) return {"removed": capability_id} @m.tool() def widgets_list(slug: str) -> dict: return {"widgets": list(mine(slug)["prefab"]["home"]["widgets"])} @m.tool() def widgets_add(slug: str, widget_id: str, title: str, description: str = "") -> dict: data = mine(slug) widgets = data["prefab"]["home"]["widgets"] if any(widget["id"] == widget_id for widget in widgets): raise ValueError("widget exists") widget = {"id": widget_id, "title": title, "kind": "prefab", "description": description} widgets.append(widget) write(slug, data) return widget @m.tool() def widgets_update(slug: str, widget_id: str, title: str = "", description: str = "") -> dict: data = mine(slug) for widget in data["prefab"]["home"]["widgets"]: if widget["id"] == widget_id: if title.strip(): widget["title"] = title.strip() if description: widget["description"] = description write(slug, data) return widget raise KeyError("unknown widget") @m.tool() def widgets_remove(slug: str, widget_id: str) -> dict: data = mine(slug) widgets = data["prefab"]["home"]["widgets"] kept = [widget for widget in widgets if widget["id"] != widget_id] if len(kept) == len(widgets): raise KeyError("unknown widget") data["prefab"]["home"]["widgets"] = kept write(slug, data) return {"removed": widget_id} @m.tool(app=True) def phast_preview(slug: str) -> PrefabApp: return prefab(mine(slug)) mgmt = m.http_app(path="/mcp", transport="streamable-http", stateless_http=True) cache: dict[str, Any] = {} started: dict[str, Any] = {} locks: dict[str, asyncio.Lock] = {} def inst(value: str): value = slug(value) if value in cache: return cache[value] data = read(value) personal = FastMCP( data["name"], version="0.2.0", instructions="Portable personal Phast app MCP.", auth=auth("https://my.phast.app", "https://" + data["host"]), mask_error_details=True, ) @personal.tool() def phast_status() -> dict: owned = mine(value) return { "ok": True, "slug": value, "owner": owned["owner"], "tools": len(owned["tools"]["installed"]), "widgets": len(owned["prefab"]["home"]["widgets"]), } @personal.tool() def phast_manifest() -> dict: return mine(value) @personal.tool(app=True) def home_app() -> PrefabApp: return prefab(mine(value)) cache[value] = personal.http_app(path="/mcp", transport="streamable-http", stateless_http=True) return cache[value] async def ready_inst(value: str): value = slug(value) app = inst(value) if value in started: return app lock = locks.setdefault(value, asyncio.Lock()) async with lock: if value not in started: cm = app.router.lifespan_context(app) await cm.__aenter__() started[value] = cm return app class Router: async def __call__(self, scope, receive, send): if scope.get("type") == "lifespan": try: await mgmt(scope, receive, send) finally: for cm in reversed(list(started.values())): try: await cm.__aexit__(None, None, None) except Exception: pass return host = dict(scope.get("headers", [])).get(b"host", b"").decode().split(":")[0].lower() request_path = scope.get("path", "/") method = scope.get("method", "GET") if request_path == "/healthz": await JSONResponse({"ok": True, "service": "phast", "source_sha": os.getenv("SOURCE_SHA", "local")})(scope, receive, send) return page = PAGES.get(host) if page and request_path == "/" and method in {"GET", "HEAD"}: await FileResponse(page, media_type="text/html", headers={"Cache-Control": "public, max-age=60"})(scope, receive, send) return if host == "phast.app": await JSONResponse({"error": "not found"}, status_code=404)(scope, receive, send) return if host == "my.phast.app": await mgmt(scope, receive, send) return if host.endswith(".phast.app"): value = host[:-10] try: data = read(value) except Exception: await JSONResponse({"error": "unknown Phast app"}, status_code=404)(scope, receive, send) return if request_path == "/" and method in {"GET", "HEAD"}: body = human(data) if method == "HEAD": await Response(content=b"", media_type="text/html", status_code=200)(scope, receive, send) else: await HTMLResponse(body)(scope, receive, send) return personal = await ready_inst(value) await personal(scope, receive, send) return await JSONResponse({"error": "unknown host"}, status_code=404)(scope, receive, send) app = Router() if __name__ == "__main__": uvicorn.run(app, host=os.getenv("PHAST_HOST", "::"), port=int(os.getenv("PORT", "8081")))