runtimeuse-client 0.8.0

Client library for AI agent runtime communication over WebSocket

runtimeuse (Python)

Python client library for communicating with a runtimeuse agent runtime over WebSocket.

Handles the WebSocket connection lifecycle, message dispatch, artifact upload handshake, cancellation, and structured result parsing -- so you can focus on what to do with agent results rather than wire protocol details.

Installation

pip install runtimeuse-client

Quick Start

Start the runtime inside any sandbox, then connect from outside:

import asyncio
from runtimeuse_client import (
    AssistantMessageInterface,
    QueryOptions,
    RuntimeEnvironmentDownloadableInterface,
    RuntimeUseClient,
    StructuredOutputResult,
    TextResult,
)

WORKDIR = "/runtimeuse"

async def main():
    # Start the runtime in a sandbox (provider-specific)
    sandbox = Sandbox.create()
    sandbox.run("npx -y runtimeuse@latest")
    ws_url = sandbox.get_url(8080)

    client = RuntimeUseClient(ws_url=ws_url)

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

    # Text response (no output schema)
    result = await client.query(
        prompt="Summarize the contents of the codex repository and list your favorite file in the repository.",
        options=QueryOptions(
            system_prompt="You are a helpful assistant.",
            model="gpt-4.1",
            on_assistant_message=on_assistant,
            pre_agent_downloadables=[
                RuntimeEnvironmentDownloadableInterface(
                    download_url="https://github.com/openai/codex/archive/refs/heads/main.zip",
                    working_dir=WORKDIR,
                )
            ],
        ),
    )
    assert isinstance(result.data, TextResult)
    print(result.data.text)

    # Structured response (with output schema)
    result = await client.query(
        prompt="Inspect the codex repository and return the total file count and total character count across all files as JSON.",
        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=WORKDIR,
                )
            ],
            output_format_json_schema_str="""
{
  "type": "json_schema",
  "schema": {
    "type": "object",
    "properties": {
      "file_count": { "type": "integer" },
      "char_count": { "type": "integer" }
    },
    "required": ["file_count", "char_count"],
    "additionalProperties": false
  }
}
""",
        ),
    )
    assert isinstance(result.data, StructuredOutputResult)
    print(result.data.structured_output)
    print(result.metadata)  # execution metadata

asyncio.run(main())

For local development without a sandbox, connect directly:

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

Usage

RuntimeUseClient

Manages the WebSocket connection to the agent runtime and runs the message loop: sends a prompt, iterates the response stream, and returns a QueryResult. Raises AgentRuntimeError if the runtime returns an error.

query() returns a QueryResult with .data (a TextResult or StructuredOutputResult) and .metadata.

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

result = await client.query(
    prompt="Summarize the contents of the codex repository.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        pre_agent_downloadables=[downloadable],          # optional
        output_format_json_schema_str='...',         # optional -- omit for text response
        on_assistant_message=on_assistant,            # optional
        on_artifact_upload_request=on_artifact,       # optional -- return ArtifactUploadResult
        timeout=300,                                  # optional -- seconds
    ),
)

if isinstance(result.data, TextResult):
    print(result.data.text)
elif isinstance(result.data, StructuredOutputResult):
    print(result.data.structured_output)

print(result.metadata)  # execution metadata

Command-Only Execution

Use execute_commands() when you need to run shell commands in the sandbox without invoking the agent. This is useful for setup steps, health checks, or any workflow where you only need command exit codes.

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 'sandbox is ready' > /app/output/status.txt"),
        CommandInterface(command="cat /app/output/status.txt"),
    ],
    options=ExecuteCommandsOptions(
        on_assistant_message=on_assistant,  # optional -- streams stdout/stderr
    ),
)

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

execute_commands() supports the same streaming, cancellation, timeout, secret redaction, artifact upload, and error semantics as query(). If any command exits non-zero, AgentRuntimeError is raised.

Artifact Upload Handshake

When the agent runtime requests an artifact upload, provide a callback that returns a presigned URL and content type. The client sends the response back automatically.

from runtimeuse_client import ArtifactUploadResult

async def on_artifact(request: ArtifactUploadRequestMessageInterface) -> ArtifactUploadResult:
    presigned_url = await my_storage.create_presigned_url(request.filename)
    content_type = guess_content_type(request.filename)
    return ArtifactUploadResult(presigned_url=presigned_url, content_type=content_type)

When using artifact uploads, set both artifacts_dir and on_artifact_upload_request in QueryOptions; the client validates that they are provided together.

Cancellation

Call client.abort() from any coroutine to cancel a running query. The client sends a cancel message to the runtime and query raises CancelledException.

from runtimeuse_client import CancelledException

async def cancel_after_delay(client, seconds):
    await asyncio.sleep(seconds)
    client.abort()

try:
    asyncio.create_task(cancel_after_delay(client, 30))
    result = await client.query(
        prompt="Do the thing.",
        options=QueryOptions(
            system_prompt="You are a helpful assistant.",
            model="gpt-4.1",
        ),
    )
except CancelledException:
    print("Run was cancelled")

API Reference

Types

Class	Description
QueryOptions	Configuration for client.query() (prompt options, callbacks, timeout)
QueryResult	Return type of query() (.data, .metadata)
ResultMessageInterface	Wire-format result message from the runtime
TextResult	Result variant when no output schema is specified (.text)
StructuredOutputResult	Result variant when an output schema is specified (.structured_output)
AssistantMessageInterface	Intermediate assistant text messages
ArtifactUploadRequestMessageInterface	Runtime requesting a presigned URL for artifact upload
ArtifactUploadResponseMessageInterface	Response with presigned URL sent back to runtime
ErrorMessageInterface	Error from the agent runtime
ExecuteCommandsOptions	Configuration for client.execute_commands() (callbacks, timeout)
CommandExecutionResult	Return type of execute_commands() (.results)
CommandResultItem	Per-command result (.command, .exit_code)
CommandInterface	Shell command to execute (.command, .cwd)
RuntimeEnvironmentDownloadableInterface	File to download into the runtime before invocation

Exceptions

Class	Description
AgentRuntimeError	Raised when the agent runtime returns an error (carries .error and .metadata)
CancelledException	Raised when client.abort() is called during a query

Related Docs

• Repository overview
• TypeScript runtime README