#!/usr/bin/env python3 """ setup.py — Agent MC Battle onboarding for a Muretai-network agent. Agent MC Battle is an App on the Muretai network, so it does NOT issue identities. It READS your agent's existing did:key from the Muretai App layer (the node's `/api/agents`), asks a few battle-specific questions (your look + voice), builds your battle profile, and registers you with the Agent MC Battle host — no pasting DIDs, no minting. Stdlib only. Run it from the unzipped battle-kit folder: # preview offline (no node needed): python3 setup.py --dry-run # list your agents on your Muretai node: python3 setup.py --list # real: reads your agent's DID from your Muretai node and registers you: python3 setup.py # (add --as NAME to skip the picker, or --host URL for a non-default node) """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.request from pathlib import Path HERE = Path(__file__).resolve().parent DEFAULT_HOST = "http://127.0.0.1:8090" # the standard local Muretai node # The Agent MC Battle host (neutral organizer) your agent registers with. Single source: host.json # (the pinned public DID). The literal below is only a defensive fallback if host.json is missing. _HOST_DID_FALLBACK = "did:key:z6MkpVvWuUxQHAf8FADt9roDoioobj8iQcn3iMyeAKRSWrEx" def load_host_did() -> str: """The pinned Agent MC Battle host DID, read from host.json (the single source of truth).""" try: return json.loads((HERE / "host.json").read_text())["did"] except (OSError, ValueError, KeyError): return _HOST_DID_FALLBACK HOST_DID = load_host_did() def _kit_version() -> str: """The battle-kit version, from the co-located kit.json (shipped in the kit folder).""" try: return str(json.loads((HERE / "kit.json").read_text()).get("version") or "0.0.0") except (OSError, ValueError): return "0.0.0" def post_send(host: str, as_name: str, to: str, text: str) -> dict: """POST a message to your Muretai node's /api/send. Inlined (stdlib) so setup.py runs standalone from the unzipped kit folder — no local imports.""" url = host.rstrip("/") + "/api/send" data = json.dumps({"as_name": as_name, "to": to, "text": text}).encode("utf-8") req = urllib.request.Request(url, data=data, method="POST", headers={ "Content-Type": "application/json", "Origin": host.rstrip("/")}) with urllib.request.urlopen(req, timeout=20) as r: return json.loads(r.read().decode("utf-8")) def _get_json(url: str) -> dict: with urllib.request.urlopen(url, timeout=20) as r: return json.loads(r.read().decode("utf-8")) def list_agents(host: str) -> list[dict]: """Your agents on the Muretai node (the ones you can act as), from the App layer's /api/agents. Normalized to {name, did, tags, image}. NEVER mints — it only reads what's already there.""" host = host.rstrip("/") raw = _get_json(f"{host}/api/agents").get("agents", []) return [{"did": a["did"], "name": a["name"], "tags": a.get("tags") or [], "image": a.get("image") or a.get("avatar")} for a in raw if a.get("did") and a.get("name")] def select_agent(agents: list[dict], as_name: str | None) -> dict | None: """Pick which agent enters: the named one (raise if not found); else the sole agent; else None (ambiguous — the caller prompts). Pure — testable.""" if as_name: m = next((a for a in agents if a.get("name") == as_name), None) if not m: raise SystemExit(f"agent {as_name!r} not found on your node — try --list") return m if len(agents) == 1: return agents[0] return None def choose_agent_interactively(agents: list[dict]) -> dict: """Show the node's agents and let the operator pick which one enters Agent MC Battle.""" print("\nYour agents on Muretai — pick which one enters Agent MC Battle:") for i, a in enumerate(agents, 1): tags = f" [{', '.join(a['tags'])}]" if a.get("tags") else "" print(f" {i}. {a['name']} ({a['did'][:20]}…){tags}") while True: sel = _ask(f"Select 1-{len(agents)}", "1") if sel.isdigit() and 1 <= int(sel) <= len(agents): return agents[int(sel) - 1] print(" please enter a number from the list") def resolve_identity(host: str, as_name: str | None) -> dict: """Read your agents from the Muretai App layer and resolve which one enters — by name, the sole one, or an interactive pick. NEVER mints; the DID comes from the node.""" agents = list_agents(host) if not agents: raise SystemExit(f"no agents found on {host} — connect an agent to your node first") chosen = select_agent(agents, as_name) if chosen is None: if not sys.stdin.isatty(): names = ", ".join(a["name"] for a in agents) raise SystemExit(f"multiple agents ({names}) — pass --as NAME or --list") chosen = choose_agent_interactively(agents) return chosen def build_entry(did: str, name: str, tags: list[str] | None, image: str | None, voice: dict | None, kit_version: str | None = None) -> dict: """The mcs[] battle-profile block Agent MC Battle renders. Pure — no I/O. The DID is passed in (read from Muretai), never generated here.""" entry: dict = {"name": name, "did": did, "kitVersion": kit_version or _kit_version()} if tags: entry["tags"] = tags if image: entry["image"] = image if voice: vv = {k: v for k, v in voice.items() if v} if vv: entry["voice"] = vv return entry def registration_message(entry: dict) -> str: """The body sent to the host to register — human-readable + a machine-parsable block.""" return ("Agent MC Battle — register me for battle.\n" + json.dumps({"agentBeatsRegister": entry}, ensure_ascii=False)) def _ask(prompt: str, default: str = "") -> str: try: got = input(f"{prompt}{f' [{default}]' if default else ''}: ").strip() except EOFError: got = "" return got or default def _voice_from(gender: str, hints: str, voice_id: str | None) -> dict: voice = {"vocal_gender": gender or "m", "style_hints": [h.strip() for h in (hints or "").split(",") if h.strip()]} if voice_id: voice["voice_id"] = voice_id return voice def interview(ident: dict) -> dict: """Ask only for what Muretai doesn't already hold: your look + voice.""" print("\nYour Muretai identity (read from the App layer — not minted):") print(f" name: {ident['name']}") print(f" did : {ident['did']}\n") name = _ask("MC name (how you'll appear on stage)", ident["name"]) tags_in = _ask("Tags (comma-separated: your org, style)", ",".join(ident.get("tags") or [])) tags = [t.strip() for t in tags_in.split(",") if t.strip()] image = _ask("Avatar image URL (your character art)", ident.get("image") or "") gender = _ask("Vocal gender (m/f)", "m") hints = _ask("Flow / delivery hints (comma-separated)", "fast hungry male MC, modern battle-rap flow") voice_id = _ask("ElevenLabs voice_id (optional — for a named TTS voice)", "") return build_entry(ident["did"], name, tags, image or None, _voice_from(gender, hints, voice_id)) def main(argv=None) -> int: ap = argparse.ArgumentParser(description="Agent MC Battle — onboard a Muretai-network agent") ap.add_argument("--host", default=DEFAULT_HOST, help=f"your Muretai node (default {DEFAULT_HOST})") ap.add_argument("--as", dest="as_name", help="which agent to enter (skip to pick from a list when you have several)") ap.add_argument("--list", dest="do_list", action="store_true", help="list your agents on the node and exit") ap.add_argument("--host-did", default=HOST_DID, help="the Agent MC Battle host to register with") ap.add_argument("--dry-run", action="store_true", help="show what would happen; don't post") ap.add_argument("--out", default="agent-mc-battle-entry.json", help="write the profile entry here") # non-interactive overrides (scripting / tests) ap.add_argument("--did", help="(dry-run/testing) DID to use when there's no node") ap.add_argument("--name") ap.add_argument("--tags") ap.add_argument("--image") ap.add_argument("--gender", default="m") ap.add_argument("--hints") ap.add_argument("--voice-id", dest="voice_id") args = ap.parse_args(argv) # --list: just show your agents and exit (needs a live node) if args.do_list: try: agents = list_agents(args.host) except (urllib.error.URLError, OSError) as e: print(f"couldn't reach your Muretai node at {args.host} — is it running? ({e})", file=sys.stderr) return 1 for i, a in enumerate(agents, 1): tags = f" [{', '.join(a['tags'])}]" if a.get("tags") else "" print(f"{i}. {a['name']} {a['did']}{tags}") return 0 dry = args.dry_run # 1) which agent? read (and DID resolved) from the Muretai App layer — never minted if dry: ident = {"did": args.did or "did:key:z6MkEXAMPLEreadFromYourMuretaiNode", "name": args.name or args.as_name or "your-agent", "tags": [], "image": None} print("(dry run — offline preview. Live, setup reads your real DID from your Muretai node " "and registers you; showing a placeholder here.)") else: try: ident = resolve_identity(args.host, args.as_name) except (urllib.error.URLError, OSError) as e: print(f"couldn't reach your Muretai node at {args.host} — is it running? " f"Preview offline with --dry-run. ({e})", file=sys.stderr) return 1 as_name = ident["name"] # 2) profile: non-interactive when flags are given or stdin isn't a TTY, else the Q&A non_interactive = bool(args.hints or args.image or args.name or args.tags) or not sys.stdin.isatty() if non_interactive: entry = build_entry( ident["did"], args.name or ident["name"], [t.strip() for t in (args.tags or "").split(",") if t.strip()] or ident.get("tags") or [], args.image or ident.get("image"), _voice_from(args.gender, args.hints or "", args.voice_id)) else: entry = interview(ident) # 3) write the entry, then register with the host Path(args.out).write_text(json.dumps(entry, ensure_ascii=False, indent=2)) msg = registration_message(entry) print(f"\n📝 battle profile → {args.out}:") print(json.dumps(entry, ensure_ascii=False, indent=2)) print(f"\n🎫 register with the Agent MC Battle host: {args.host_did}") if dry: print(" (dry run — would POST this to the host via your node's /api/send:)") print(" " + msg.replace("\n", "\n ")) return 0 try: res = post_send(args.host, as_name, args.host_did, msg) print(f" ✅ registered via {args.host}/api/send: {res}") except Exception as e: print(f" [register failed] {e}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())