Python Client

The Python client is the control plane for RuntimeUse. It connects to the sandbox runtime, sends the invocation, and turns runtime messages into a single QueryResult.

pip install runtimeuse-client

Basic Query

import asyncio

from runtimeuse_client import QueryOptions, RuntimeUseClient, TextResult


async def main() -> None:
    client = RuntimeUseClient(ws_url="ws://localhost:8080")

    result = await client.query(
        prompt="What is 2 + 2",
        options=QueryOptions(
            system_prompt="You are a helpful assistant.",
            model="gpt-4.1",
        ),
    )

    assert isinstance(result.data, TextResult)
    print(result.data.text)
    print(result.metadata)


asyncio.run(main())

query() returns a QueryResult with:

data: either TextResult (.text) or StructuredOutputResult (.structured_output)
metadata: execution metadata returned by the runtime (includes token usage when available)

Use client.session() when you want to run several sequential query() or execute_commands() calls against the same sandbox without paying the connection cost each time. The session keeps a single WebSocket open until the context exits.

async with client.session() as session:
    summary = await session.query(
        prompt="Summarize the repository.",
        options=QueryOptions(
            system_prompt="You are a helpful assistant.",
            model="gpt-4.1",
        ),
    )

    commands = await session.execute_commands(
        commands=[CommandInterface(command="ls /runtimeuse")],
        options=ExecuteCommandsOptions(),
    )

Calls inside a session are serialized: each one runs to completion (or cancellation) before the next is sent.

session.abort() cancels only the in-flight request and leaves the session open for the next call. One-shot client.abort() is the equivalent for the convenience wrappers below.

When the context exits, the client sends end_session_message and waits for the runtime to drain any artifact uploads triggered by post-agent commands before closing the socket — files written after the last call still make it out.

The default WebSocketTransport supports persistent sessions. A custom transport must implement PersistentTransport for client.session() to work.

Return Structured JSON

Pass output_format_json_schema_str when your application needs machine-readable output instead of free-form text. The result will be a StructuredOutputResult.

import json

from pydantic import BaseModel
from runtimeuse_client import StructuredOutputResult


class RepoStats(BaseModel):
    file_count: int
    char_count: int


result = await client.query(
    prompt="Inspect the repository and return the total file count and character count as JSON.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        output_format_json_schema_str=json.dumps(
            {
                "type": "json_schema",
                "schema": RepoStats.model_json_schema(),
            }
        ),
    ),
)

assert isinstance(result.data, StructuredOutputResult)
stats = RepoStats.model_validate(result.data.structured_output)
print(stats)

Set Agent Environment Variables

Use agent_env to inject environment variables into the agent process. These are merged on top of the sandbox's existing environment.

result = await client.query(
    prompt="Run the build and report any failures.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-5.4",
        agent_env={
            "NODE_ENV": "production",
            "DATABASE_URL": "postgres://localhost/mydb",
        },
    ),
)

Download Files into the Sandbox

Use pre_agent_downloadables to fetch a repository, zip archive, or any URL into the sandbox before the agent runs. This is the primary way to give the agent access to a codebase or dataset.

from runtimeuse_client import RuntimeEnvironmentDownloadableInterface

result = await client.query(
    prompt="Summarize the contents of this repository and list your favorite file.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        pre_agent_downloadables=[
            RuntimeEnvironmentDownloadableInterface(
                download_url="https://github.com/openai/codex/archive/refs/heads/main.zip",
                working_dir="/runtimeuse",
            )
        ],
    ),
)

The runtime downloads and extracts the file before handing control to the agent.

Upload Artifacts

When the runtime requests an artifact upload, return a presigned URL and content type from on_artifact_upload_request. Set artifacts_dir to tell the runtime which sandbox directory contains the files to upload - both options must be provided together.

from runtimeuse_client import ArtifactUploadResult


async def on_artifact_upload_request(request) -> ArtifactUploadResult:
    presigned_url = await create_presigned_url(request.filename)
    return ArtifactUploadResult(
        presigned_url=presigned_url,
        content_type="application/octet-stream",
    )


result = await client.query(
    prompt="Generate a report and save it as report.txt.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        artifacts_dir="/runtimeuse/output",
        on_artifact_upload_request=on_artifact_upload_request,
    ),
)

Stream Assistant Messages

Use on_assistant_message to receive the agent's intermediate text output while the run is still happening, and on_command_output for stdout/stderr from any commands the runtime executes.

async def on_assistant_message(msg) -> None:
    for block in msg.text_blocks:
        print(f"[assistant] {block}")


async def on_command_output(msg) -> None:
    print(f"[{msg.stream}] {msg.command}: {msg.text}", end="")


result = await client.query(
    prompt="Inspect this repository.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        on_assistant_message=on_assistant_message,
        on_command_output=on_command_output,
    ),
)

Each command_output_message carries the stream ("stdout" or "stderr"), the chunk of text, and the original command string.

Run Commands Without the Agent

Use execute_commands() when you only need to run shell commands in the sandbox -- no agent invocation, no prompt. The method returns per-command exit codes and raises AgentRuntimeError if any command fails.

from runtimeuse_client import (
    CommandInterface,
    ExecuteCommandsOptions,
    RuntimeUseClient,
)

client = RuntimeUseClient(ws_url="ws://localhost:8080")

result = await client.execute_commands(
    commands=[
        CommandInterface(command="mkdir -p /app/output"),
        CommandInterface(
            command="echo $GREETING > /app/output/status.txt",
            env={"GREETING": "sandbox is ready"},
        ),
        CommandInterface(command="cat /app/output/status.txt"),
    ],
    options=ExecuteCommandsOptions(
        on_command_output=on_command_output,  # streams stdout/stderr
    ),
)

for item in result.results:
    print(f"{item.command} -> exit {item.exit_code}")
    print(item.stdout)

Each item in result.results carries exit_code and the buffered stdout from that command, alongside the real-time output delivered via on_assistant_message.

execute_commands() supports the same callbacks and options as query(): streaming via on_command_output, artifact uploads, cancellation, timeout, and secrets_to_redact. Use pre_execution_downloadables to fetch files into the sandbox before the commands run.

Each CommandInterface accepts an optional env dict that is merged on top of the sandbox's process.env for that command.

Cancel a Run

Call client.abort() from another coroutine to cancel an in-flight query. The client sends a cancel message to the runtime and query() raises CancelledException.

import asyncio
from runtimeuse_client import CancelledException


async def cancel_soon(client: RuntimeUseClient) -> None:
    await asyncio.sleep(5)
    client.abort()


try:
    asyncio.create_task(cancel_soon(client))
    await client.query(prompt="Do the thing.", options=options)
except CancelledException:
    print("Run was cancelled")

Set a Timeout

Use timeout (in seconds) to limit how long a query can run. If the limit is exceeded, query() raises TimeoutError.

result = await client.query(
    prompt="Do the thing.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        timeout=120,
    ),
)

Redact Secrets

Pass secrets_to_redact to strip sensitive strings from any output or logs that leave the sandbox.

result = await client.query(
    prompt="Check the API status.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        secrets_to_redact=["sk-live-abc123", "my_db_password"],
    ),
)

Handle Errors

query() raises AgentRuntimeError if the runtime sends back an error. The exception carries .error (the error message) and .metadata.

from runtimeuse_client import AgentRuntimeError

try:
    result = await client.query(prompt="Do the thing.", options=options)
except AgentRuntimeError as e:
    print(f"Runtime error: {e.error}")
    print(f"Metadata: {e.metadata}")