phast/app
1
0 Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

providers_search is Oracle; tools_install/list/remove put capabilities on the belt

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FAqjv1sLfUad2bMcsmPbRT
This commit is contained in:
teahubxyzandClaude Fable 5.1 committed 2026-10-29 00:43:51 +00:00
1 parent cbd4a28e75
commit 717ec9f14e
1 file changed
+138 -3
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+138 -3
View File
@@ -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(
"<article><h2>"
+ str(tool["name"])
+ " <small>"
+ str(tool.get("evidence_state", "claimed"))
+ "</small></h2><p>"
+ str(tool.get("provider", ""))
+ "</p><p><code>"
+ str(tool.get("remote_url", ""))
+ "</code></p></article>"
for tool in data["tools"]["installed"]
)
cards = "".join(
"<article><h2>"
+ 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}}</style>"
+ "<main><header><b>Phast app<em>.</em></b><code>"
+ data["host"]
+ "/mcp</code></header><h1>"
+ data["name"]
+ "</h1><section>"
+ "</h1><h3>Tool belt · "
+ str(len(data["tools"]["installed"]))
+ "</h3><section>"
+ (belt or "<article><h2>Empty belt</h2><p>Ask your agent: search providers, then install one.</p></article>")
+ "</section><h3>Home</h3><section>"
+ cards
+ "</section></main>"
)
@@ -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()