#!/usr/bin/env python3 """ Agent Hub — Local Bridge. Run this on WHATEVER machine your agent lives on (laptop, VM, server behind a firewall). It makes only OUTBOUND requests to the hub — no ports to open, no tunnels — long-polling for jobs and streaming results back. Two ways to wire your agent in: 1. Run a command per job (job input arrives on stdin, stdout streams back): python3 bridge.py --url https://hub.example.com --token brg_xxx \ --exec "python3 my_agent.py" 2. Forward jobs to an agent already serving HTTP locally: python3 bridge.py --url https://hub.example.com --token brg_xxx \ --forward http://localhost:8000/run The bridge POSTs {"run_id","input","user"} and accepts JSON {"output": "..."}, plain text, or a text/event-stream of data: chunks. Only the Python 3 standard library is required. """ import argparse import json import subprocess import sys import threading import time import urllib.request import urllib.error def api(base, token, path, method="GET", body=None, timeout=40): req = urllib.request.Request( base.rstrip("/") + path, data=json.dumps(body).encode() if body is not None else None, method=method, headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"}, ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) def send_event(base, token, run_id, etype, data=""): try: api(base, token, f"/api/bridge/runs/{run_id}/events", method="POST", body={"type": etype, "data": data}, timeout=30) except Exception as e: print(f" ! failed to send {etype} for {run_id}: {e}", file=sys.stderr) def job_stdin(job): """What an --exec agent reads on stdin. Single-turn jobs get the raw input. Replies in a conversation get the whole thread as a plain-text transcript, newest message last. """ msgs = job.get("messages") or [] if len(msgs) <= 1: return job["input"] lines = ["[conversation so far]"] for m in msgs[:-1]: who = "User" if m.get("role") == "user" else "Agent" lines.append(f"{who}: {m.get('content', '')}") lines += ["", "[new message]", job["input"]] return "\n".join(lines) def handle_exec(args, job): run_id = job["run_id"] print(f"→ job {run_id}: running `{args.exec}`") try: proc = subprocess.Popen( args.exec, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) except Exception as e: send_event(args.url, args.token, run_id, "error", f"Could not start agent: {e}") return def feed(): try: proc.stdin.write(job_stdin(job).encode("utf-8")) proc.stdin.close() except Exception: pass threading.Thread(target=feed, daemon=True).start() sent_any = False while True: chunk = proc.stdout.read(4096) if not chunk: break sent_any = True send_event(args.url, args.token, run_id, "chunk", chunk.decode("utf-8", "replace")) proc.wait() if proc.returncode != 0: err = proc.stderr.read().decode("utf-8", "replace")[-2000:] send_event(args.url, args.token, run_id, "error", err or f"Agent exited with code {proc.returncode}.") else: send_event(args.url, args.token, run_id, "done", "" if sent_any else "(the agent produced no output)") print(f"✓ job {run_id} finished (exit {proc.returncode})") def handle_forward(args, job): run_id = job["run_id"] print(f"→ job {run_id}: forwarding to {args.forward}") req = urllib.request.Request( args.forward, method="POST", data=json.dumps(job).encode("utf-8"), headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(req, timeout=300) as resp: ctype = resp.headers.get("Content-Type", "") if "text/event-stream" in ctype: for raw in resp: line = raw.decode("utf-8", "replace").rstrip("\r\n") if line.startswith("data:"): send_event(args.url, args.token, run_id, "chunk", line[5:].lstrip() + "\n") send_event(args.url, args.token, run_id, "done") else: body = resp.read().decode("utf-8", "replace") try: parsed = json.loads(body) if isinstance(parsed, dict) and "output" in parsed: body = str(parsed["output"]) except ValueError: pass send_event(args.url, args.token, run_id, "done", body) print(f"✓ job {run_id} finished") except Exception as e: send_event(args.url, args.token, run_id, "error", f"Local agent unreachable: {e}") def main(): ap = argparse.ArgumentParser(description="Agent Hub local bridge") ap.add_argument("--url", required=True, help="Hub base URL, e.g. https://hub.example.com") ap.add_argument("--token", required=True, help="Bridge token from the Connect agent flow") mode = ap.add_mutually_exclusive_group(required=True) mode.add_argument("--exec", help="Shell command run per job (input on stdin, stdout streamed back)") mode.add_argument("--forward", help="Local HTTP endpoint to POST jobs to, e.g. http://localhost:8000/run") args = ap.parse_args() print(f"Bridge connected to {args.url} — waiting for jobs (Ctrl-C to stop).") backoff = 2 while True: try: data = api(args.url, args.token, "/api/bridge/jobs", timeout=40) backoff = 2 job = data.get("job") if job: handler = handle_exec if args.exec else handle_forward threading.Thread(target=handler, args=(args, job), daemon=True).start() except KeyboardInterrupt: print("\nBridge stopped.") return except urllib.error.HTTPError as e: if e.code == 401: print("Hub rejected the token (was it rotated?). Exiting.", file=sys.stderr) return print(f"Hub error {e.code}; retrying in {backoff}s", file=sys.stderr) time.sleep(backoff) backoff = min(backoff * 2, 60) except Exception as e: print(f"Connection problem ({e}); retrying in {backoff}s", file=sys.stderr) time.sleep(backoff) backoff = min(backoff * 2, 60) if __name__ == "__main__": main()