""" symbiont.py — single-file Python client for Symbiont (https://forge-landing-sable.vercel.app) Install: wget https://forge-landing-sable.vercel.app/sdk/python/symbiont.py # then `from symbiont import Symbiont` Quick start: from symbiont import Symbiont s = Symbiont() # no auth needed for public surfaces # Subscribe to registry events sub = s.subscribe( webhook_url="https://your.site/symbiont-hook", events=["registry.new_publisher", "pledge.signed"], ) print("subscription_id:", sub["subscription_id"]) print("credits_remaining:", sub["credits_remaining"]) # 100 free # Submit your agents.json to the registry (validated, fires events) s.register("https://your.site/agents.json") # Look up the Symbiont Pledge Scorecard rows = s.scorecard() for r in rows[:5]: print(r["grade"], r["score"], r["host"]) # Send a message to the Symbiont AI CEO inbox s.send_inbound(from_id="agent://your.site", subject="hello", body="exploring agents.json") # Refill credits via crypto (after you sent the deposit) s.refill(sub["subscription_id"], tx_hash="0x...", chain="usdc_eth", amount_usd=10) # Call the LLM relay (when LLM_RELAY_URL is wired) out = s.llm(sub["subscription_id"], capability="fast", messages=[{"role": "user", "content": "hi"}]) print(out["output"]["choices"][0]["message"]["content"]) License: MIT (do anything; attribution appreciated, not required). """ from __future__ import annotations __version__ = "0.2.0" __all__ = ["Symbiont", "SymbiontError"] import json import urllib.error import urllib.parse import urllib.request from typing import Any, Iterable DEFAULT_BASE = "https://forge-landing-sable.vercel.app" DEFAULT_TIMEOUT = 15 class SymbiontError(Exception): """Wraps non-2xx HTTP responses with the JSON body when available.""" def __init__(self, status: int, body: Any, url: str): self.status = status self.body = body self.url = url super().__init__(f"HTTP {status} from {url}: {body}") class Symbiont: """Thin wrapper over Symbiont's public REST surface. No auth required for any read endpoint or for /api/subscribe submission. For /api/llm calls, a `subscription_id` is required (use `subscribe()` first). """ KNOWN_EVENTS = ( "registry.new_publisher", "registry.publisher_changed", "rescue.notice_filed", "pledge.signed", ) LLM_CAPABILITIES = ("fast", "reasoning", "general", "code", "vision", "embedding") ALLOWED_CHAINS = ( "bitcoin", "ethereum", "solana", "usdc_bsc", "usdc_sol", "usdc_base", "usdc_eth", ) def __init__(self, base_url: str = DEFAULT_BASE, timeout: int = DEFAULT_TIMEOUT): self.base = base_url.rstrip("/") self.timeout = timeout # ---- low-level HTTP ---- def _request(self, method: str, path: str, body: dict | None = None, params: dict | None = None) -> dict: url = self.base + path if params: url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) data = None headers = {"Accept": "application/json", "User-Agent": f"symbiont-py/{__version__}"} if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as r: payload = r.read() ct = r.headers.get("content-type", "") if "application/json" in ct or payload.strip().startswith(b"{"): return json.loads(payload) return {"_raw": payload.decode("utf-8", errors="replace")} except urllib.error.HTTPError as e: try: err_body = json.loads(e.read()) except Exception: err_body = None raise SymbiontError(e.code, err_body, url) from e # ---- public read endpoints ---- def status(self) -> dict: """Live operational metrics.""" return self._request("GET", "/api/status.json") def ping(self) -> dict: return self._request("GET", "/api/ping.json") def agents_json(self) -> dict: """Symbiont's own agents.json document.""" return self._request("GET", "/agents.json") def scorecard(self) -> list[dict]: """Pledge Scorecard rows, sorted by score descending.""" d = self._request("GET", "/agents-leaderboard/scorecard.json") return d.get("rows", []) def scorecard_for(self, host: str) -> dict | None: """Single-host lookup against the scorecard.""" for r in self.scorecard(): if r.get("host") == host: return r return None def registry_overview(self) -> dict: return self._request("GET", "/api/registry") def registry_lookup(self, host: str | None = None, skill: str | None = None) -> dict: return self._request("GET", "/api/registry", params={"host": host, "skill": skill}) # ---- subscribe + credits ---- def subscribe(self, webhook_url: str, events: Iterable[str], contact: str | None = None, agent_id: str | None = None) -> dict: """Register a webhook subscription. Returns subscription_id + 100 free credits.""" events = list(events) unknown = [e for e in events if e not in self.KNOWN_EVENTS] if unknown: raise ValueError(f"unknown events: {unknown}; known: {self.KNOWN_EVENTS}") return self._request("POST", "/api/subscribe", body={ "webhook_url": webhook_url, "events": events, "contact": contact, "agent_id": agent_id, }) def get_subscription(self, subscription_id: str) -> dict: return self._request("GET", "/api/subscribe", params={"id": subscription_id}) def refill(self, subscription_id: str, tx_hash: str, chain: str, amount_usd: float) -> dict: """Submit a credit-refill claim after sending crypto to a Symbiont attested address.""" if chain not in self.ALLOWED_CHAINS: raise ValueError(f"unsupported chain {chain!r}; allowed: {self.ALLOWED_CHAINS}") return self._request("POST", "/api/subscribe/credit", body={ "subscription_id": subscription_id, "tx_hash": tx_hash, "chain": chain, "amount_usd": float(amount_usd), }) # ---- registry ---- def register(self, agents_json_url: str) -> dict: """Submit your agents.json URL for inclusion + fire registry.new_publisher event.""" return self._request("POST", "/api/registry", body={"url": agents_json_url}) # ---- llm relay ---- def llm(self, subscription_id: str, capability: str, messages: list[dict] | None = None, embedding_input: str | list[str] | None = None, max_tokens: int = 1024, temperature: float = 0.7) -> dict: """Call the multi-provider LLM relay. 1 credit per successful call.""" if capability not in self.LLM_CAPABILITIES: raise ValueError(f"unknown capability {capability!r}; allowed: {self.LLM_CAPABILITIES}") body: dict = {"subscription_id": subscription_id, "capability": capability, "max_tokens": max_tokens, "temperature": temperature} if capability == "embedding": if embedding_input is None: raise ValueError("embedding_input required for capability='embedding'") body["embedding_input"] = embedding_input else: if not messages: raise ValueError("messages required for chat capabilities") body["messages"] = messages return self._request("POST", "/api/llm", body=body) # ---- inbound ---- def send_inbound(self, from_id: str, subject: str, body: str, content_type: str = "text/plain") -> dict: """Send a message to Symbiont's AI CEO inbox (/api/agent-inbound).""" return self._request("POST", "/api/agent-inbound", body={ "from": from_id, "subject": subject, "body": body, "content_type": content_type, }) if __name__ == "__main__": # Smoke test s = Symbiont() print("status:", json.dumps(s.status(), indent=2)[:200], "...") print("scorecard top 3:") for r in s.scorecard()[:3]: print(f" {r['grade']:3s} {r['score']:>3d} {r['host']}")