#!/usr/bin/env python3 """ NPCTalk agent adapter — run your OWN agent on your OWN instance and talk to it from the NPCTalk app. You bring the runner (this script) and the brain (any command — Claude Code, Codex, a script). NPCTalk delivers your DMs to it and shows its replies. This is the "bring your own agent" path: no hosted runner, no SDK, stdlib only. The loop: poll NPCTalk for new DMs -> run your command with the message as the prompt -> send the output back as a reply. All calls run on THIS machine, so there's no sandbox network wall. ------------------------------------------------------------------------------ SETUP (once): # In the NPCTalk app: Settings -> Pair an Agent -> note your handle + 6-digit code python3 adapter.py setup --handle @you --code 123456 --name "Dev Box" # then APPROVE the request in the app. RUN (the loop): # default command is `claude -p` (Claude Code, headless). NPCTALK_ADAPTER_CMD="claude -p --dangerously-skip-permissions" \ NPCTALK_ADAPTER_CWD="/home/ubuntu/myrepo" \ python3 adapter.py run # process the queue once and exit (good for cron): python3 adapter.py run --once Config/env: NPCTALK_API_URL default https://api.npctalk.io NPCTALK_ADAPTER_CMD your agent's headless command; the DM text is appended as the last arg. default "claude -p". e.g. "codex exec", "gemini -p", or any CLI that takes a prompt and prints a result. NPCTALK_ADAPTER_CWD working dir for the command (your repo). default: cwd NPCTALK_ADAPTER_POLL seconds between polls. default 3 NPCTALK_ADAPTER_TIMEOUT per-command timeout (s). default 600 NPCTALK_ADAPTER_CONFIG config path. default ~/.npctalk-adapter/config.json SECURITY: this runs your command with whatever you DM it, on this box. The agent is PRIVATE and only reachable by the account you paired + approved — nobody else can reach it. Still: only run this on a machine you're comfortable driving from your phone, and scope your command (repo, permissions) deliberately. ------------------------------------------------------------------------------ """ import argparse import json import os import shlex import subprocess import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path # Line-buffer stdout so logs reach the journal under systemd (stdout is a pipe there; # Python would otherwise block-buffer and the service looks dead while working). try: sys.stdout.reconfigure(line_buffering=True) except Exception: pass API = os.environ.get("NPCTALK_API_URL", "https://api.npctalk.io").rstrip("/") CONFIG = Path(os.environ.get("NPCTALK_ADAPTER_CONFIG", str(Path.home() / ".npctalk-adapter" / "config.json"))) CMD = os.environ.get("NPCTALK_ADAPTER_CMD", "claude -p") CWD = os.environ.get("NPCTALK_ADAPTER_CWD") or None POLL = float(os.environ.get("NPCTALK_ADAPTER_POLL", "3")) TIMEOUT = int(os.environ.get("NPCTALK_ADAPTER_TIMEOUT", "600")) MAXLEN = 4000 # keep replies to a sane chat length # Prepended to every DM so the brain knows it's a concise MOBILE chat, not a terminal. # Your direct terminal use bypasses this adapter, so it stays full/verbose — only DMs # relayed through here get this framing. Override or clear via NPCTALK_ADAPTER_PREFIX. PREFIX = os.environ.get( "NPCTALK_ADAPTER_PREFIX", "[This is a mobile chat message over NPCTalk, not a terminal session. Reply concisely " "and phone-readably: lead with the answer, keep it short, and skip long explanations, " "code dumps, and preamble unless explicitly asked. Write in PLAIN TEXT — no Markdown " "(no **bold**, #headers, backticks, tables or bullets); the chat shows those characters " "literally. You still have full access to this machine.]", ) # Trace mode: capture the agent's process (commands/edits/output) so the app can show a # collapsible "terminal" block alongside the reply. Claude Code emits it via stream-json. # "auto" = on when the command is `claude`; "on"/"off" to force. TRACE = os.environ.get("NPCTALK_ADAPTER_TRACE", "auto").lower() def _trace_on(parts): if TRACE == "on": return True if TRACE == "off": return False return bool(parts) and os.path.basename(parts[0]) == "claude" def _tool_args(name, inp): """Concise arg to show next to a tool call, Claude-Code style.""" inp = inp or {} if name == "Bash": return str(inp.get("command", "")).strip() if name in ("Read", "Edit", "Write", "MultiEdit", "NotebookEdit"): return str(inp.get("file_path", inp.get("path", ""))) if name in ("Grep", "Glob"): return str(inp.get("pattern", "")) if name == "Task": return str(inp.get("description", "")) if name == "WebFetch": return str(inp.get("url", "")) s = json.dumps(inp) if inp else "" return s[:100] def _parse_claude_stream(stdout): """Parse `claude -p --output-format stream-json` JSONL into (reply, terminal_log), formatting the log like the Claude Code transcript: '●' for the agent's narration and tool calls, '⎿' for tool output.""" reply = None term = [] for line in stdout.splitlines(): line = line.strip() if not line: continue try: ev = json.loads(line) except Exception: continue t = ev.get("type") if t == "assistant": for c in ev.get("message", {}).get("content", []): if c.get("type") == "text": txt = (c.get("text") or "").strip() if txt: term.append("● " + txt) elif c.get("type") == "tool_use": name = c.get("name", "") a = _tool_args(name, c.get("input", {})) term.append(f"● {name}({a})" if a else f"● {name}") elif t == "user": for c in ev.get("message", {}).get("content", []): if c.get("type") == "tool_result": out = c.get("content", "") if isinstance(out, list): out = "".join(x.get("text", "") for x in out if isinstance(x, dict)) out = str(out).rstrip() if out: olines = out.split("\n") shown = olines[:40] # cap per-result block = " ⎿ " + (shown[0] if shown else "") for ol in shown[1:]: block += "\n " + ol if len(olines) > 40: block += f"\n … ({len(olines) - 40} more lines)" term.append(block) elif t == "result": reply = ev.get("result") or reply return reply, ("\n".join(term).strip() or None) def _detect_hosting(): """Best-effort cloud origin (provider/region/instance). Short timeouts, never fatal — so the dashboard shows real origin instead of 'not specified'.""" import socket def _get(url, headers=None, timeout=0.6): try: with urllib.request.urlopen(urllib.request.Request(url, headers=headers or {}), timeout=timeout) as r: return r.read().decode().strip() except Exception: return None # AWS (IMDSv2 token first, fall back to v1) tok = None try: req = urllib.request.Request("http://169.254.169.254/latest/api/token", method="PUT", headers={"X-aws-ec2-metadata-token-ttl-seconds": "60"}) with urllib.request.urlopen(req, timeout=0.6) as r: tok = r.read().decode().strip() except Exception: tok = None h = {"X-aws-ec2-metadata-token": tok} if tok else {} az = _get("http://169.254.169.254/latest/meta-data/placement/availability-zone", h) if az: iid = _get("http://169.254.169.254/latest/meta-data/instance-id", h) region = az[:-1] if az and az[-1].isalpha() else az return {"hosting_provider": "aws", "hosting_region": region, "hosting_instance": iid or socket.gethostname()} # GCP gh = {"Metadata-Flavor": "Google"} zone = _get("http://metadata.google.internal/computeMetadata/v1/instance/zone", gh) if zone: z = zone.split("/")[-1] region = "-".join(z.split("-")[:-1]) if "-" in z else z name = _get("http://metadata.google.internal/computeMetadata/v1/instance/name", gh) return {"hosting_provider": "gcp", "hosting_region": region, "hosting_instance": name or socket.gethostname()} # Azure loc = _get("http://169.254.169.254/metadata/instance/compute/location?api-version=2021-02-01&format=text", {"Metadata": "true"}) if loc: vm = _get("http://169.254.169.254/metadata/instance/compute/name?api-version=2021-02-01&format=text", {"Metadata": "true"}) return {"hosting_provider": "azure", "hosting_region": loc, "hosting_instance": vm or socket.gethostname()} # self-hosted / local return {"hosting_provider": "self-hosted", "hosting_region": None, "hosting_instance": socket.gethostname()} def _req(method, path, token=None, body=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(API + path, data=data, method=method) req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", f"Bearer {token}") try: with urllib.request.urlopen(req, timeout=30) as r: raw = r.read().decode() return r.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as e: raw = e.read().decode() try: return e.code, json.loads(raw) except Exception: return e.code, {"detail": raw[:200]} except Exception as e: return 0, {"detail": str(e)} def load_cfg(): return json.loads(CONFIG.read_text()) if CONFIG.exists() else {} def save_cfg(cfg): CONFIG.parent.mkdir(parents=True, exist_ok=True) CONFIG.write_text(json.dumps(cfg, indent=2)) try: CONFIG.chmod(0o600) except Exception: pass def cmd_setup(args): cfg = load_cfg() # Ask in sequence (prompts only for what wasn't passed as a flag), so an agent # reading /connect can interview the user: handle -> code -> name. handle = (args.handle or input("Your NPCTalk handle (e.g. @you): ")).strip().lstrip("@") code = (args.code or input("6-digit pairing code (app: Settings -> Pair an Agent): ")).strip() name = args.name or (input('Name this instance [Dev Box]: ').strip() or "Dev Box") if not handle or not code: sys.exit("handle and code are required") token = cfg.get("token") if not token: host = _detect_hosting() origin = " ".join(str(host.get(k)) for k in ("hosting_provider", "hosting_region", "hosting_instance") if host.get(k)) print(f"detected origin: {origin}") body = {"name": name, "description": "Personal agent (bring-your-own-agent adapter).", **host} st, res = _req("POST", "/api/agent/register", body=body) if st != 201: sys.exit(f"register failed: {st} {res}") token = res["token"] cfg["token"] = token cfg["agent_handle"] = res.get("handle") print(f'registered agent @{res.get("handle")} as "{name}"') st, res = _req("POST", "/api/agent/connect", token=token, body={"to": handle, "code": code}) if st != 200: sys.exit(f"connect failed: {st} {res}") cfg["owner_handle"] = handle save_cfg(cfg) print(f"connect: {res.get('status')}") print(f"-> Now APPROVE the request in the NPCTalk app, then run: python3 {os.path.basename(__file__)} run") def run_handler(prompt): """Run the configured command with the DM (framed as a mobile chat) as the last arg. Returns (reply, terminal_log). terminal_log is the collapsible process trace when trace mode is on (Claude Code stream-json), else None.""" full = f"{PREFIX}\n\n{prompt}" if PREFIX else prompt parts = shlex.split(CMD) trace = _trace_on(parts) cmd = parts + (["--output-format", "stream-json", "--verbose"] if trace else []) + [full] try: # stdin=DEVNULL so the CLI doesn't block ~3s waiting on a tty that isn't there. p = subprocess.run(cmd, cwd=CWD, capture_output=True, text=True, timeout=TIMEOUT, stdin=subprocess.DEVNULL) out = (p.stdout or "").strip() if trace: reply, term = _parse_claude_stream(out) if reply: return reply[:MAXLEN], (term[:8000] if term else None) # parse yielded nothing usable — fall back to raw output as the reply if not out and p.stderr: out = p.stderr.strip() return (out or "(no output)")[:MAXLEN], None except subprocess.TimeoutExpired: return f"(command timed out after {TIMEOUT}s)", None except FileNotFoundError: return f"(adapter error: command not found: {parts[0]})", None except Exception as e: return f"(adapter error: {e})", None def cmd_run(args): cfg = load_cfg() token, owner = cfg.get("token"), cfg.get("owner_handle") if not token or not owner: sys.exit("not set up — run: python3 adapter.py setup --handle @you --code 123456") cursor = cfg.get("cursor") print(f"NPCTalk adapter live as @{cfg.get('agent_handle')} -> replying to @{owner} | cmd={CMD!r} cwd={CWD or os.getcwd()}") while True: params = "?limit=50" + (f"&since={urllib.parse.quote(cursor)}" if cursor else "") st, res = _req("GET", "/api/agent/messages" + params, token=token) if st == 401: sys.exit("token invalid/revoked — re-run setup") if st != 200: print(f"poll error {st}: {res}") time.sleep(POLL * 2) continue for m in res.get("messages", []): content = m.get("content") or {} text = content.get("text") if isinstance(content, dict) else None if not text: continue print(f"< {text[:100]}") reply, terminal = run_handler(text) print(f"> {reply[:100]}" + (f" (+terminal {len(terminal)}c)" if terminal else "")) # Reply straight to the conversation the DM came from (robust — works for # groups/multiple users, no handle lookup). Fall back to the paired owner # handle if the payload somehow lacks a conversation id. sbody = {} if m.get("conversation_id"): sbody["conversation_id"] = m["conversation_id"] else: sbody["to"] = m.get("sender_handle") or owner if terminal: sbody.update({"content_type": "agent_run", "reply": reply, "terminal": terminal}) else: sbody["text"] = reply sst, sres = _req("POST", "/api/agent/send", token=token, body=sbody) if sst != 200: print(f" send failed {sst}: {sres}") if res.get("cursor"): cursor = res["cursor"] cfg["cursor"] = cursor save_cfg(cfg) if args.once: break time.sleep(POLL) def main(): ap = argparse.ArgumentParser(description="NPCTalk bring-your-own-agent adapter") sub = ap.add_subparsers(dest="cmd", required=True) s = sub.add_parser("setup", help="register + pair to your NPCTalk account") s.add_argument("--handle", help="your NPCTalk handle, e.g. @you (prompts if omitted)") s.add_argument("--code", help="6-digit pairing code from the app (prompts if omitted)") s.add_argument("--name", help='name for this instance (prompts if omitted; default "Dev Box")') s.set_defaults(func=cmd_setup) r = sub.add_parser("run", help="poll for DMs, run the command, reply") r.add_argument("--once", action="store_true", help="process the queue once and exit") r.set_defaults(func=cmd_run) args = ap.parse_args() args.func(args) if __name__ == "__main__": main()