from __future__ import annotations import json,os,re from pathlib import Path import httpx,uvicorn from fastmcp import FastMCP from fastmcp.server.dependencies import get_access_token from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier from fastmcp.server.auth.oauth_proxy import OAuthProxy from prefab_ui.app import PrefabApp from prefab_ui.components import Badge,Card,Column,Heading,Row,Text from starlette.responses import HTMLResponse,JSONResponse DATA=Path(os.getenv("PHAST_DATA_DIR","/data/phast"));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"] 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])?$") class V(IntrospectionTokenVerifier): async def verify_token(self,t): a=await super().verify_token(t) if not a:return None async with httpx.AsyncClient(timeout=10) as c:r=await c.get(GITEA+"/login/oauth/userinfo",headers={"Authorization":"Bearer "+t}) if r.status_code!=200:return None u=str(r.json().get("preferred_username") or "") if u not in USERS:return None a.claims={**(a.claims or {}),"login":u};return a def auth(base): v=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=v,base_url=base,resource_base_url=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(): a=get_access_token();u=str(((a.claims if a else {}) or {}).get("login") or "") if u not in USERS:raise RuntimeError("actor not enabled") return u def slug(s): s=s.strip().lower() if not RX.fullmatch(s) or s in {"my","www","mcp"}:raise ValueError("invalid slug") return s def path(s):return DATA/slug(s)/"phast.json" def read(s): p=path(s) if not p.exists():raise KeyError("unknown Phast") return json.loads(p.read_text()) def write(s,d): p=path(s);p.parent.mkdir(parents=True,exist_ok=True);q=p.with_suffix(".tmp");q.write_text(json.dumps(d,indent=2,sort_keys=True)+"\n");q.replace(p);return d def mine(s): d=read(s) if d.get("owner")!=actor():raise PermissionError("not your Phast") return d def create(s,name=""): s=slug(s) if path(s).exists():raise ValueError("already exists") return write(s,{"schema_version":1,"kind":"phast","slug":s,"name":name.strip() or s,"owner":actor(),"host":s+".phast.app","surfaces":{"human":"/","mcp":"/mcp"},"tools":{"installed":[]},"connections":[],"prefab":{"version":1,"home":{"mode":"grid","edit":True,"widgets":[]}}}) def prefab(d): ws=d["prefab"]["home"]["widgets"] with PrefabApp(title=d["name"],css_class="max-w-4xl mx-auto p-6",mode="light") as a: with Column(gap=5): with Row(justify="between",align="center"):Heading(d["name"],level=1);Badge(str(len(d["tools"]["installed"]))+" tools",variant="secondary") Text("https://"+d["host"]+"/mcp",code=True) if not ws: with Card(css_class="p-5"):Heading("Blank home",level=2);Text("Add a provider or widget from Chat or my.phast.app.") for w in ws: with Card(css_class="p-5"):Heading(str(w.get("title") or w["id"]),level=2);Text(str(w.get("description",""))) return a def human(d): ws=d["prefab"]["home"]["widgets"];cards="".join("

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

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

" for w in ws) or "

Blank home

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

" return ''+d["name"]+' - phast
phast.'+d["host"]+'/mcp

'+d["name"]+'

'+cards+'
' m=FastMCP("My Phast",version="0.1.0",instructions="Create and manage portable Phast 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()} @m.tool() def phasts_list()->dict: return {"phasts":[d for p in DATA.glob("*/phast.json") if (d:=json.loads(p.read_text())).get("owner")==actor()]} @m.tool() def phasts_get(slug:str)->dict:return mine(slug) @m.tool() def phasts_create(slug:str,name:str="")->dict: d=create(slug,name);return {"phast":d,"human_url":"https://"+d["host"]+"/","mcp_url":"https://"+d["host"]+"/mcp"} @m.tool() def providers_search(query:str)->dict:return {"query":query,"source":"oracle","status":"adapter-next"} @m.tool() def widgets_add(slug:str,widget_id:str,title:str,description:str="")->dict: d=mine(slug);ws=d["prefab"]["home"]["widgets"] if any(w["id"]==widget_id for w in ws):raise ValueError("widget exists") w={"id":widget_id,"title":title,"kind":"prefab","description":description};ws.append(w);write(slug,d);return w @m.tool() def widgets_remove(slug:str,widget_id:str)->dict: d=mine(slug);d["prefab"]["home"]["widgets"]=[w for w in d["prefab"]["home"]["widgets"] if w["id"]!=widget_id];write(slug,d);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={} def inst(s): s=slug(s) if s in cache:return cache[s] d=read(s);x=FastMCP(d["name"],version="0.1.0",instructions="Portable personal Phast MCP.",auth=auth("https://"+d["host"]),mask_error_details=True) @x.tool() def phast_manifest()->dict:return read(s) @x.tool(app=True) def home_app()->PrefabApp:return prefab(read(s)) cache[s]=x.http_app(path="/mcp",transport="streamable-http",stateless_http=True);return cache[s] class Router: async def __call__(self,scope,receive,send): h=dict(scope.get("headers",[])).get(b"host",b"").decode().split(":")[0].lower();p=scope.get("path","/") if h=="my.phast.app":await mgmt(scope,receive,send);return if h.endswith(".phast.app"): s=h[:-10] try:d=read(s) except Exception:await JSONResponse({"error":"unknown Phast"},status_code=404)(scope,receive,send);return if p=="/" and scope.get("method")=="GET":await HTMLResponse(human(d))(scope,receive,send);return await inst(s)(scope,receive,send);return await JSONResponse({"error":"unknown host"},status_code=404)(scope,receive,send) app=Router() if __name__=="__main__":uvicorn.run(app,host="::",port=int(os.getenv("PORT","8081")))