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)

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)

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.

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 when you want intermediate progress while the run is still happening.

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


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,
    ),
)

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}")