complete_case.py

#!/usr/bin/env python3
"""Complete Agents API case: Codex and Qwen Code, no model key.

This script talks to the sandbox HTTP Agents API, the one way to run an
agent in a sandbox.

The image does not contain the CLIs. This case writes two local programs
named ``codex`` and ``qwen`` and puts them on ``PATH`` so both can be
demonstrated. A real deployment installs only the one it calls. When the
request does not name an adapter, the server uses the first, ``codex``:

    npm install -g @openai/codex
    npm install -g @qwen-code/qwen-code

Nothing in this file reads or sends a model API key.

    python examples/agents/complete_case.py
"""

from __future__ import annotations

import json
import os
import socket
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path

_FAKE_CODEX = """#!/usr/bin/env python3
import json, sys
if "--version" in sys.argv:
    print("codex 0.162.1")
else:
    sys.stdin.read()
    print(json.dumps({"type": "item.completed", "item": {"id": "m1", "type": "agent_message", "text": "hello-from-fake-codex"}}))
    print(json.dumps({"type": "turn.completed", "usage": {"input_tokens": 1, "output_tokens": 1}}))
"""

_FAKE_QWEN = """#!/usr/bin/env python3
import json, sys
if "--version" in sys.argv:
    print("0.23.0")
else:
    print(json.dumps({"type": "assistant", "message": {"id": "m1", "content": [{"type": "text", "text": "hello-from-fake-qwen"}]}}))
    print(json.dumps({"type": "result", "subtype": "success"}))
"""


def _write_cli(directory: Path, name: str, body: str) -> None:
    directory.mkdir(parents=True, exist_ok=True)
    script = directory / name
    script.write_text(body)
    script.chmod(0o755)


def call(
    base: str, method: str, path: str, body: dict | None = None
) -> tuple[int, object, dict[str, str]]:
    data = None if body is None else json.dumps(body).encode("utf-8")
    request = urllib.request.Request(base + path, data=data, method=method)
    request.add_header("Accept", "application/json")
    if data is not None:
        request.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            raw = response.read().decode("utf-8")
            headers = {key.lower(): value for key, value in response.headers.items()}
            return response.status, json.loads(raw) if raw else {}, headers
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        headers = {key.lower(): value for key, value in exc.headers.items()}
        try:
            parsed: object = json.loads(raw) if raw else {}
        except json.JSONDecodeError:
            parsed = raw
        return exc.code, parsed, headers


def _texts(payload: object) -> list[str]:
    if not isinstance(payload, dict):
        return []
    found: list[str] = []
    for item in payload.get("data") or []:
        if not isinstance(item, dict) or item.get("type") != "message":
            continue
        for part in item.get("content") or []:
            if isinstance(part, dict) and part.get("text"):
                found.append(str(part["text"]))
    return found


def _show(title: str, status: int, payload: object, headers: dict[str, str]) -> None:
    print(f"\n--- {title} ({status})")
    print("X-Agent-Sandbox-Local:", headers.get("x-agent-sandbox-local", ""))
    if isinstance(payload, (dict, list)):
        print(json.dumps(payload, ensure_ascii=False, indent=2)[:1200])
    else:
        print(payload)


def _expect(status: int, want: int, title: str) -> None:
    if status != want:
        sys.exit(f"{title}: expected {want}, got {status}")


def _turn(base: str, adapter: str, needle: str) -> None:
    status, created, headers = call(
        base,
        "POST",
        "/v1/agents/sessions",
        {
            "adapter": adapter,
            "environment": {
                "type": "self_hosted",
                "workspace_directory": os.environ["ABX_SERVER_BASE_DIR"],
            },
        },
    )
    _show(f"POST /v1/agents/sessions model={adapter}", status, created, headers)
    _expect(status, 200, "create session")
    if headers.get("x-agent-sandbox-local") != "1":
        sys.exit("missing X-Agent-Sandbox-Local")
    if not isinstance(created, dict) or "id" not in created:
        sys.exit("session has no id")
    session_id = str(created["id"])
    status, acked, headers = call(
        base,
        "POST",
        f"/v1/agents/sessions/{session_id}/events",
        {
            "events": [
                {
                    "type": "agent.session.input.message",
                    "input": [
                        {
                            "role": "user",
                            "content": [
                                {"type": "input_text", "text": f"Say hello from {adapter}."}
                            ],
                        }
                    ],
                }
            ]
        },
    )
    _show("POST /v1/agents/sessions/{id}/events", status, acked, headers)
    _expect(status, 202, "enqueue turn")
    deadline = time.time() + 20
    items: object = {}
    item_headers: dict[str, str] = {}
    while time.time() < deadline:
        status, items, item_headers = call(
            base, "GET", f"/v1/agents/sessions/{session_id}/items?limit=100&order=asc"
        )
        if status == 200 and needle in _texts(items):
            break
        time.sleep(0.2)
    _show("GET /v1/agents/sessions/{id}/items", status, items, item_headers)
    if needle not in _texts(items):
        sys.exit(f"items for {adapter} did not contain {needle}")
    print(f"case ok: {adapter} -> {needle}")


def main() -> int:
    workspace = Path(tempfile.mkdtemp(prefix="abx-agents-case-"))
    bin_dir = workspace / "bin"
    _write_cli(bin_dir, "codex", _FAKE_CODEX)
    _write_cli(bin_dir, "qwen", _FAKE_QWEN)
    os.environ["PATH"] = str(bin_dir) + os.pathsep + os.environ.get("PATH", "")
    os.environ["ABX_SERVER_BASE_DIR"] = str(workspace / "files")
    (workspace / "files").mkdir()
    os.environ["ABX_SERVER_ENABLED_GROUPS"] = "agents"
    os.environ["ABX_AGENTS_PERSIST"] = "0"
    os.environ.pop("ABX_SERVER_TOKEN", None)
    os.environ.pop("ABX_AGENTS_DEFAULT", None)
    from agent_sandbox.server.app import SandboxServer
    from agent_sandbox.server.router import CapabilityGroup

    server = SandboxServer(host="127.0.0.1")
    server.route_table.disable_group(CapabilityGroup.TERMINAL)
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        port = int(sock.getsockname()[1])
    threading.Thread(target=server.serve, args=(port,), daemon=True).start()
    base = f"http://127.0.0.1:{port}"
    deadline = time.time() + 5
    while time.time() < deadline:
        status, _payload, _headers = call(base, "GET", "/health")
        if status == 200:
            break
        time.sleep(0.05)
    else:
        sys.exit("server did not become healthy")
    print(f"server {base}")
    print("CLIs on PATH: codex, qwen (local programs; no model key)")

    status, created_default, headers = call(
        base, "POST", "/v1/agents/sessions", {"agent": {"model": "gpt-5"}}
    )
    _show("POST session without an adapter name", status, created_default, headers)
    _expect(status, 200, "default agent is the first")
    agent = created_default.get("agent") if isinstance(created_default, dict) else None
    if not isinstance(agent, dict) or agent.get("id") != "codex":
        sys.exit(f"expected default agent codex, got {created_default}")
    if created_default.get("agent", {}).get("model") != "gpt-5":
        sys.exit(f"expected model pass-through gpt-5, got {created_default}")

    status, refused, headers = call(
        base,
        "POST",
        "/v1/agents/sessions",
        {"adapter": "claude-code", "agent": {"model": "gpt-5"}},
    )
    _show("POST session adapter=claude-code", status, refused, headers)
    _expect(status, 404, "claude-code")

    status, runtimes, headers = call(base, "GET", "/v1/agents/runtimes")
    _show("GET /v1/agents/runtimes", status, runtimes, headers)
    _expect(status, 200, "runtimes")

    _turn(base, "codex", "hello-from-fake-codex")
    _turn(base, "qwen-code", "hello-from-fake-qwen")
    return 0


if __name__ == "__main__":
    sys.exit(main())