#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.11" # dependencies = ["anthropic>=1.17,<2", "exedev>=0.4,<0.5"] # /// """Run each Claude Managed Agents session on its own exe.dev VM. This claims work from a self-hosted environment's queue, as Anthropic's "Launch sandboxes from the SDK poller" describes. A session gets a work item whenever it needs a worker: when it starts, and when it resumes after its last worker left. For each one, this finds the exe.dev VM whose comment is the session's ID, or creates one, and starts `ant beta:worker run` there with the work item's secret, which is good for that session only. The worker runs the session's tools on the VM, and exits (stopping the work item) a minute after the session goes idle. The VM stays for the session's next turn. The dispatcher keeps no state: the VMs' comments are its memory. ANTHROPIC_ENVIRONMENT_ID the self-hosted environment, env_... ANTHROPIC_ENVIRONMENT_KEY its key, sk-ant-oat01-...; it reaches a VM only for a work item that comes without a secret EXE_TOKEN an exe.dev API key for new, ls, and ssh, ideally scoped to a tag (see the page this comes from) EXE_NEW_FLAGS optional flags for `new`, each as --name=value, such as --cpu=4 --memory=16GB https://exe.dev/deploy/claude-managed-agents """ import asyncio import collections import logging import os import secrets import shlex import sys import exedev # exe.dev's SDK from anthropic import APIStatusError, AsyncAnthropic ANT_VERSION = "1.42.0" # Anthropic's CLI, which has the worker log = logging.getLogger("dispatcher") # Runs on the session's VM, as exedev, with the work item's ID and the file # holding the worker's settings, secret included. systemd reads that file, # and deletes it when the worker stops. Running this twice for one work # item starts one worker. START_WORKER = """set -eo pipefail work=$1 env_file=$2 if systemctl is-active --quiet "claude-$work"; then rm -f "$env_file"; exit; fi ant=https://github.com/anthropics/anthropic-cli/releases/download/v%(v)s ant=$ant/ant_%(v)s_linux_$(dpkg --print-architecture).tar.gz [ -x /usr/local/bin/ant ] || curl -fsSL "$ant" | sudo tar -xz -C /usr/local/bin ant sudo install -d -o exedev -g exedev /workspace /mnt/memory sudo systemd-run --quiet --collect --unit="claude-$work" --uid=exedev \\ --working-directory=/workspace -p EnvironmentFile="$env_file" \\ -p ExecStopPost="/bin/rm -f $env_file" -p SyslogIdentifier=claude-worker \\ /usr/local/bin/ant beta:worker run --workdir /workspace || { rm -f "$env_file"; exit 1; } """ % {"v": ANT_VERSION} def new_options(): """EXE_NEW_FLAGS, such as "--cpu=4 --memory=16GB", as arguments to new().""" options = {} for flag in shlex.split(os.environ.get("EXE_NEW_FLAGS", "")): name, eq, value = flag.removeprefix("--").partition("=") if not eq: sys.exit(f"dispatcher: EXE_NEW_FLAGS: write {flag} as {flag}=...") options[name.replace("-", "_")] = value return options async def start_worker(exe, item, key): session = item.data.id vm = next((vm for vm in await exe.vms() if vm.comment == session), None) if vm is None: vm = await exe.new(comment=session, **new_options()) log.info("%s: new VM %s", session, vm.name) if item.secret: # good for this session only cred = f"ANTHROPIC_WORK_SECRET={item.secret}" else: log.warning("%s: no work secret; giving the VM the environment key", session) cred = f"ANTHROPIC_ENVIRONMENT_KEY={key}" env = [cred, f"ANTHROPIC_ENVIRONMENT_ID={item.environment_id}", f"ANTHROPIC_SESSION_ID={session}", f"ANTHROPIC_WORK_ID={item.id}"] if os.environ.get("ANTHROPIC_BASE_URL"): env.append(f"ANTHROPIC_BASE_URL={os.environ['ANTHROPIC_BASE_URL']}") env_file = f"/tmp/claude-{item.id}-{secrets.token_hex(4)}.env" await vm.write_file(env_file, "".join(f"{e}\n" for e in env), mode=0o600) await vm.run(["bash", "-c", START_WORKER, "start-worker", item.id, env_file], check=True) log.info("%s: worker started on %s; its log: ssh %s journalctl -t claude-worker", session, vm.name, vm.ssh_dest) async def dispatch(client, exe, item, key, lock): async with lock: # one work item per session at a time, so it gets one VM for attempt in range(3): try: return await start_worker(exe, item, key) except Exception as e: log.warning("%s: %s", item.data.id, e) await asyncio.sleep(5 * (attempt + 1)) log.error("%s: giving up; stopping work %s", item.data.id, item.id) try: await client.beta.environments.work.stop( item.id, environment_id=item.environment_id, force=True) except APIStatusError as e: log.error("%s: stop failed: %s", item.data.id, e) def need(name): if not os.environ.get(name): sys.exit(f"dispatcher: set {name}") return os.environ[name] async def main(): logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpx2").setLevel(logging.WARNING) env_id = need("ANTHROPIC_ENVIRONMENT_ID") key = need("ANTHROPIC_ENVIRONMENT_KEY") os.environ.pop("ANTHROPIC_API_KEY", None) # never needed, so never sent client = AsyncAnthropic(auth_token=key) exe = exedev.AsyncClient(need("EXE_TOKEN")) new_options() # bad flags fail now, not at the first session n = len(await exe.vms()) # a bad key fails here log.info("exe.dev key works; it can see %d VMs", n) stats = await client.beta.environments.work.stats(env_id) # and here log.info("environment key works; sessions waiting: %d", stats.depth) locks = collections.defaultdict(asyncio.Lock) tasks = set() async for item in client.beta.environments.work.poller( environment_id=env_id, environment_key=key, auto_stop=False): # the worker on the VM stops the work when done if item.data.type != "session": log.warning("skipping %s work %s", item.data.type, item.id) await client.beta.environments.work.stop(item.id, environment_id=env_id, force=True) continue task = asyncio.create_task(dispatch(client, exe, item, key, locks[item.data.id])) tasks.add(task) task.add_done_callback(tasks.discard) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: pass except (APIStatusError, exedev.ExeError) as e: sys.exit(f"dispatcher: {e}")