Add an agent

To deploy an agent into the sandbox for evaluation, you implement a deployer: a Python class that installs, launches, and reads back one harness. It does all its work through the handle the framework hands it, so the same code runs on any image and any OS. This page is how to write one. The Agents & executor page has the concepts.

The one rule: read from the sandbox, never hardcode

A deployer is constructed with one argument, the executor, and everything hangs off it. Your agent's own config is self.config; the executor also gives you the work dir, the sandbox handle, and env. The discipline that makes a deployer portable: read binary paths, the OS, and the work dir from the handle and branch on the OS, never write "node" or a literal path.

So a path is built, not written:

sandbox = self.executor.sandbox
node  = sandbox.node                                   # not "node"
entry = self._join(sandbox.mcp_server_dir, "src", "index.js",
                   is_linux=sandbox.is_linux)          # /…/index.js  vs  C:\…\index.js

@staticmethod
def _join(*parts, is_linux):
    sep = "/" if is_linux else "\\"
    head, *rest = parts
    return sep.join([head.rstrip("/\\"), *(p.strip("/\\") for p in rest)])

install(): stage the agent

Get the harness in place and ready to run. The canonical sandbox-CLI sequence (see agents/claude_code/deployer.py, agents/codex/deployer.py):

async def install(self) -> None:
    sandbox = self.executor.sandbox

    # 1. locate the CLI; (re)install when missing OR stale vs the pinned version.
    #    Resolve the installed shim by absolute path under the install prefix —
    #    images pre-bake older copies that would otherwise shadow it on PATH.
    cli = shutil.which("claude")
    if not cli or installed_version(cli) != pinned_version:      # pinned in config (cli_version)
        cli = await self._npm_install("@anthropic-ai/claude-code@2.1.170")   # → ~/.local shim

    # 2. prove it runs  (stdin=DEVNULL avoids TTY/Defender hangs on Windows)
    subprocess.run([cli, "--version"], check=True, stdin=subprocess.DEVNULL, timeout=60)

    # 3. a clean work dir for this run
    Path(self.executor.work_dir).mkdir(parents=True, exist_ok=True)

    # 4. hand the agent a GUI action space: ensure the cua MCP bridge, point the CLI at it
    await ensure_cua_mcp_server(sandbox)                 # idempotent, no-op when pre-baked
    mcp = {"mcpServers": {"cua": {
        "command": sandbox.node,                         # read, don't hardcode
        "args":    [self._join(sandbox.mcp_server_dir, "src", "index.js",
                               is_linux=sandbox.is_linux)],
        "env":     cua_bridge_env(self.executor),
    }}}
    (Path(self.executor.work_dir) / "mcp_config.json").write_text(json.dumps(mcp))

Common steps: probe-or-install the binary, create the work dir, ensure the cua MCP bridge and write the agent's MCP config, and set up auth. Most knobs (which version, whether to overlay a patched binary, what to disable) come from self.config.

launch(prompt): run it, return a result

Spawn the agent detached, poll until it exits, and return an AgentRunResult. Detaching + polling (rather than blocking) keeps the run cancellable so a SIGTERM can reap the child:

async def launch(self, prompt: str) -> AgentRunResult:
    wd = Path(self.executor.work_dir)
    (wd / "prompt.txt").write_text(prompt)
    argv, env = self._build_argv(cfg=self.config), self._build_env(cfg=self.config)

    t0 = time.monotonic()
    with open(wd/"prompt.txt","rb") as pin, open(wd/"transcript.jsonl","wb") as out, \
         open(wd/"stderr.log","wb") as err:
        proc = subprocess.Popen(argv, stdin=pin, stdout=out, stderr=err,
                                env=env, cwd=str(wd), start_new_session=True)
    try:
        while proc.poll() is None:
            await asyncio.sleep(2)                        # poll; stays cancellable
    except asyncio.CancelledError:
        proc.terminate(); proc.kill(); raise              # reap on shutdown

    return AgentRunResult(
        status   = "completed" if proc.returncode == 0 else "failed",
        exit_code= proc.returncode,
        duration_s = time.monotonic() - t0,
        transcript_path = str(wd / "transcript.jsonl"),
        stderr_path     = str(wd / "stderr.log"),
    )

A host-side harness (out-of-sandbox, e.g. agents/ale_claw/deployer.py) writes launch() differently: instead of a subprocess it opens a session / MCP runtime to the VM and drives its own agent loop in-process, writing a transcript to work_dir at the end. Same signature, same return.

parse_artifacts(): logs → trajectory

A host-side @classmethod that runs after the framework gathers work_dir locally. Read the agent's native transcript and emit one trajectory step per turn:

@classmethod
def parse_artifacts(cls, *, work_dir, config, run_result, builder) -> None:
    t = work_dir / "transcript.jsonl"
    if not t.exists():
        builder.add_step(source="system", message="no transcript")   # missing → record a note
        return
    for line in t.read_text(errors="replace").splitlines():
        if not line.strip():
            continue
        ev = json.loads(line)
        if ev["type"] == "assistant":
            builder.add_step(
                source="agent",
                message=_text(ev),
                tool_calls=[ToolCall(id=b["id"], name=b["name"], arguments=b["input"])
                            for b in _tool_uses(ev)],
                metrics=StepMetrics(input_tokens=_usage(ev, "input_tokens"), ...),
            )
        elif ev["type"] == "user":                        # tool results
            builder.add_step(source="environment", observation=_obs(ev))

Map each agent's own format to the same schema: assistant turns become agent steps carrying their ToolCalls (and, when the transcript reports usage, per-step StepMetrics, which finalize() sums into the run's token and cost totals); tool results become environment observations. Handle a missing or cut-off transcript by recording what you can (a source="system" note). The framework wraps this call in try/except anyway, so a stray exception just logs and appends a failure step rather than crashing the run.

Wire it in

Put a standalone config dataclass next to the deployer (each agent's config is its own, there's no shared base class), and set three class-level attributes (declared on the class, read by the framework):

@dataclass
class MyAgentConfig:
    name: ClassVar[str] = "my-agent"
    model: str = "anthropic/claude-sonnet-4.6"
    max_turns: int = 30           # only the knobs this agent consumes

class MyAgentDeployer(BaseAgentDeployer):
    default_executor    : ClassVar[str]               = "sandbox"   # or "local"
    supported_executors : ClassVar[frozenset[str]]    = frozenset({"sandbox"})
    hot_artifacts       : ClassVar[tuple[str, ...]]   = ("transcript.jsonl", "stderr.log")

default_executor / supported_executors tell the framework where this agent may run; hot_artifacts is just a list of filenames. The framework tails them off the VM periodically during the run (you write no pull logic), so a killed run still leaves a partial trace. Register the deployer in orchestration/factory.py (_AGENT_FQNS), then ship a preset:

# configs/agents/my_agent.yaml
harness: my_agent
model: anthropic/claude-sonnet-4.6
config:
  max_turns: 30

At run time the framework instantiates your Config dataclass from this config: block (unset fields fall back to the config.py defaults) and hands the instance to the deployer as self.config.