Drive a sandbox from your agent loop

# box_tool.py — the sandbox tool, provider-independent
import boxlite

# The schema body is plain JSON Schema, so every provider accepts it unchanged.
SANDBOX_EXEC_SCHEMA = {
    "type": "object",
    "properties": {
        "argv": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Command and arguments, e.g. ['python', '-c', 'print(1)']",
        }
    },
    "required": ["argv"],
}

SANDBOX_EXEC_DESCRIPTION = (
    "Run a command inside an isolated sandbox and return its stdout, stderr and exit code."
)


async def sandbox_exec(box, argv) -> dict:
    """The single door between the model and the environment.

    Always returns a dict — never raises — so one bad tool call cannot end the loop.
    """
    if not isinstance(argv, list) or not argv or not all(isinstance(a, str) for a in argv):
        return {"stdout": "", "stderr": "argv must be a non-empty list of strings", "exit_code": 2}
    try:
        result = await box.exec(*argv)
    except RuntimeError as exc:
        # A command the image does not have (a common model hallucination) raises here.
        # Report it as a tool result so the model can correct itself.
        return {"stdout": "", "stderr": f"command could not be started: {exc}", "exit_code": -1}
    # A non-zero exit does NOT raise — pass the exit code back and let the model react.
    return {"stdout": result.stdout, "stderr": result.stderr, "exit_code": result.exit_code}

# provider_openai.py
import json
import os
from openai import AsyncOpenAI

from box_tool import SANDBOX_EXEC_SCHEMA, SANDBOX_EXEC_DESCRIPTION, sandbox_exec

TOOLS = [{
    "type": "function",
    "name": "sandbox_exec",
    "description": SANDBOX_EXEC_DESCRIPTION,
    "parameters": SANDBOX_EXEC_SCHEMA,        # <-- the shared schema, unchanged
}]

def build_client():
    # base_url is what selects the endpoint. Omit it for OpenAI itself.
    return AsyncOpenAI(
        api_key=os.environ["<YOUR_API_KEY_ENV_VAR>"],   # e.g. OPENAI_API_KEY
        base_url=os.getenv("<YOUR_BASE_URL_ENV_VAR>"),  # e.g. https://api.minimax.io/v1
    )

async def run_turn(client, box, goal, model, max_rounds=12):
    response = await client.responses.create(
        model=model,                                    # e.g. "<YOUR_MODEL>"
        instructions="Use sandbox_exec to interact with the environment. Summarize when done.",
        input=[{"role": "user", "content": goal}],
        tools=TOOLS,
        tool_choice="auto",
    )
    for _ in range(max_rounds):
        calls = [item for item in response.output if item.type == "function_call"]
        if not calls:
            return response
        outputs = []
        for call in calls:
            args = json.loads(call.arguments or "{}")
            result = await sandbox_exec(box, args.get("argv", []))
            outputs.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result),
            })
        response = await client.responses.create(
            model=model,
            previous_response_id=response.id,
            input=outputs,
            tools=TOOLS,
            tool_choice="auto",
        )
    return response

# provider_anthropic.py
import os
from anthropic import AsyncAnthropic

from box_tool import SANDBOX_EXEC_SCHEMA, SANDBOX_EXEC_DESCRIPTION, sandbox_exec

TOOLS = [{
    "name": "sandbox_exec",
    "description": SANDBOX_EXEC_DESCRIPTION,
    "input_schema": SANDBOX_EXEC_SCHEMA,      # <-- the same shared schema
}]

def build_client():
    return AsyncAnthropic(api_key=os.environ["<YOUR_API_KEY_ENV_VAR>"])  # e.g. ANTHROPIC_API_KEY

async def run_turn(client, box, goal, model, max_rounds=12):
    messages = [{"role": "user", "content": goal}]
    for _ in range(max_rounds):
        response = await client.messages.create(
            model=model,                                # e.g. "<YOUR_MODEL>"
            max_tokens=4096,
            system="Use sandbox_exec to interact with the environment. Summarize when done.",
            messages=messages,
            tools=TOOLS,
        )
        calls = [b for b in response.content if b.type == "tool_use"]
        if not calls:
            return response
        messages.append({"role": "assistant", "content": response.content})
        results = []
        for call in calls:
            result = await sandbox_exec(box, call.input.get("argv", []))
            results.append({
                "type": "tool_result",
                "tool_use_id": call.id,
                "content": str(result),
            })
        messages.append({"role": "user", "content": results})
    return response

# agent.py — run with: python agent.py
import asyncio
import os

import boxlite

from box_tool import sandbox_exec  # noqa: F401  (used by the provider module)
from provider_openai import build_client, run_turn   # <-- swap this line to change provider


async def main() -> None:
    goal = "Check the Python version, then compute the first 10 primes and print them."
    try:
        client = build_client()
        # One Box per task: created on entry, destroyed on exit.
        async with boxlite.SimpleBox(image="python:slim") as box:
            response = await run_turn(client, box, goal, model=os.environ["<YOUR_MODEL_ENV_VAR>"])
            print(response)
    except RuntimeError as exc:
        # No hardware virtualization, or the image could not be pulled.
        print(f"sandbox unavailable: {exc}")


if __name__ == "__main__":
    asyncio.run(main())

pip install "boxlite[orchestration]"