From 717ec9f14e664ad5fa0bba391b30d167ad9b91b9 Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Thu, 29 Oct 2026 00:43:51 +0000 Subject: [PATCH] providers_search is Oracle; tools_install/list/remove put capabilities on the belt Oracle (oracle.teahub.xyz, teahubxyz/oracle) is the capability index. Phast fetches its search.json projection (cached an hour in memory, last good copy on the volume) and searches it: every term must match name/provider/summary/id; results rank by evidence rung, then having a remote endpoint, then name matches, then Teahub's own declarations. Results carry the evidence state so a registry listing shows as what it is: a claim. tools_install puts a capability on a Phast app's belt as a reference (id, endpoint, evidence, sources), never a credential; tools_list and tools_remove round it out. The personal home and the Prefab app show the belt above the widgets. Not yet: proxying calls through the personal /mcp to installed endpoints, and the observation that write back to Oracle. That proxy is designed once, shared with the Tea MCP oracle doorway (teahubxyz/oracle#3). The belt stops being empty today. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01FAqjv1sLfUad2bMcsmPbRT --- server.py | 141 ++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 138 insertions(+), 3 deletions(-) diff --git a/server.py b/server.py index abb766f..6762df4 100644 --- a/server.py +++ b/server.py @@ -4,6 +4,7 @@ import asyncio import json import os import re +import time from pathlib import Path from typing import Any @@ -29,6 +30,71 @@ 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 = "" + try: + response = httpx.get(ORACLE + "/search.json", timeout=30, headers={"User-Agent": "phast-app/0.2"}) + if response.status_code == 200: + text = response.text + try: + ORACLE_CACHE.write_text(text) + except OSError: + pass + except httpx.HTTPError: + text = "" + if not text and not _oracle["rows"] and ORACLE_CACHE.exists(): + text = ORACLE_CACHE.read_text() + if text: + payload = json.loads(text) + columns = payload["columns"] + _oracle["rows"] = [dict(zip(columns, row)) for row in payload["rows"]] + _oracle["fresh"] = response.headers.get("last-modified", "") if text and "response" in locals() else "" + _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[0], item[1]["name"].lower()), reverse=True) + return [row for _, row in hits[: max(1, min(limit, 50))]] class V(IntrospectionTokenVerifier): @@ -158,6 +224,12 @@ def prefab(data: dict[str, Any]) -> PrefabApp: 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) @@ -171,6 +243,18 @@ def prefab(data: dict[str, Any]) -> PrefabApp: 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"]) @@ -189,12 +273,19 @@ def human(data: dict[str, Any]) -> str: + "b{font-size:28px}em{color:#ff4d00}h1{font-size:72px;letter-spacing:-5px;margin:80px 0 40px}" + "section{display:grid;grid-template-columns:repeat(3,1fr);gap:14px}" + "article{border:1px solid #d5d1c8;padding:18px;min-height:150px}p{color:#77746c}" + + "h2 small{font-size:11px;font-weight:400;color:#ff4d00;text-transform:uppercase;letter-spacing:.08em;margin-left:8px}" + + "h3{font-size:11px;text-transform:uppercase;letter-spacing:.1em;color:#77746c;margin:32px 0 12px}" + + "code{font-size:12px;word-break:break-all}" + "@media(max-width:700px){section{grid-template-columns:1fr}}" + "
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 + "
" ) @@ -254,8 +345,52 @@ def phasts_update(slug: str, name: str) -> dict: @m.tool() -def providers_search(query: str) -> dict: - return {"query": query, "source": "oracle", "status": "adapter-next"} +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()