# Agent Runtime (/agent-runtime)



The [agent runtime](https://www.npmjs.com/package/runtimeuse) is the process that runs inside the sandbox. It exposes a WebSocket server, receives invocations from the Python client, and delegates work to an agent handler.

CLI [#cli]

```bash
npx -y runtimeuse@latest                            # OpenAI handler on port 8080
npx -y runtimeuse@latest --agent claude             # Claude handler
npx -y runtimeuse@latest --port 3000                # custom port
npx -y runtimeuse@latest --handler ./my-handler.js  # custom handler entrypoint
```

Built-in Handlers [#built-in-handlers]

* `openai`: the default handler, uses the [OpenAI Agents SDK](https://openai.github.io/openai-agents-js/) with shell and web search tools.
* `claude`: uses [Claude Agents SDK](https://platform.claude.com/docs/en/agent-sdk/overview) with Claude Code preset.

OpenAI Handler [#openai-handler]

Requires `OPENAI_API_KEY` to be set in the environment. The handler runs the agent with shell access and web search enabled.

```bash
export OPENAI_API_KEY=your_openai_api_key
npx -y runtimeuse@latest
```

Claude Handler [#claude-handler]

Requires the `@anthropic-ai/claude-code` CLI and `ANTHROPIC_API_KEY`. Always set `IS_SANDBOX=1` and `CLAUDE_SKIP_ROOT_CHECK=1` in the sandbox environment.

```bash
npm install -g @anthropic-ai/claude-code
export ANTHROPIC_API_KEY=your_anthropic_api_key
export IS_SANDBOX=1
export CLAUDE_SKIP_ROOT_CHECK=1
npx -y runtimeuse@latest --agent claude
```

Programmatic Startup [#programmatic-startup]

If you want to embed RuntimeUse directly in your own Node process, start it programmatically:

```typescript
import { RuntimeUseServer, openaiHandler } from "runtimeuse";

const server = new RuntimeUseServer({
  handler: openaiHandler,
  port: 8080,
});

await server.startListening();
```

Custom Handlers [#custom-handlers]

When the built-in handlers are not enough, you can pass your own handler to `RuntimeUseServer`:

```typescript
import { RuntimeUseServer } from "runtimeuse";
import type {
  AgentHandler,
  AgentInvocation,
  AgentResult,
  MessageSender,
} from "runtimeuse";

const handler: AgentHandler = {
  async run(
    invocation: AgentInvocation,
    sender: MessageSender,
  ): Promise<AgentResult> {
    sender.sendAssistantMessage(["Running agent..."]);

    const output = await myAgent(
      invocation.systemPrompt,
      invocation.userPrompt,
    );

    return {
      type: "structured_output",
      structuredOutput: output,
      metadata: { duration_ms: 1500 },
    };
  },
};

const server = new RuntimeUseServer({ handler, port: 8080 });
await server.startListening();
```

Handler Contracts [#handler-contracts]

Your handler receives an `AgentInvocation` with:

| Field          | Type                                                | Description                                                                                               |
| -------------- | --------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| `systemPrompt` | `string`                                            | System prompt for the agent.                                                                              |
| `userPrompt`   | `string`                                            | User prompt sent from the Python client.                                                                  |
| `model`        | `string`                                            | Model name passed by the client.                                                                          |
| `outputFormat` | `{ type: "json_schema"; schema: ... } \| undefined` | Present when the client requests structured output. Pass to your agent to enforce the schema.             |
| `signal`       | `AbortSignal`                                       | Fires when the client sends a cancel message. Pass to any async operations that support cancellation.     |
| `logger`       | `Logger`                                            | Use `invocation.logger.log(msg)` to emit log lines visible in sandbox logs.                               |
| `env`          | `Record<string, string> \| undefined`               | Environment variables from the client's `agent_env`. Merge with `process.env` when spawning subprocesses. |

Use `MessageSender` to stream intermediate output before returning the final result:

* `sendAssistantMessage(textBlocks: string[])`: emit text blocks the Python client receives via `on_assistant_message`.
* `sendErrorMessage(error: string, metadata?: Record<string, unknown>)`: signal a non-fatal error before aborting.

Return an `AgentResult` from your handler:

```typescript
// Text result
return { type: "text", text: "...", metadata: { duration_ms: 100 } };

// Structured output result
return { type: "structured_output", structuredOutput: { file_count: 42 }, metadata: {} };
```

`metadata` is optional and is passed through to `result.metadata` on the Python side.


# Introduction (/)



[RuntimeUse](https://github.com/getlark/runtimeuse) is an open-source runtime and client library for running AI agents inside isolated sandboxes and controlling them from Python over WebSocket.

<img src="/terminal.svg" alt="RuntimeUse terminal" style={{ maxWidth: '560px', width: '100%' }} />

<Card title="Quickstart" description="See how to integrate with popular sandbox providers." href="/quickstart" />

When to use [#when-to-use]

* Your agent needs filesystem, CLI, or network access inside an isolated runtime.
* Your application should stay outside the sandbox while still controlling the run.
* You don't want to build infrastructure for interacting with your agent in sandbox.

What it handles [#what-it-handles]

* **Task invocations**: send a prompt to any agent runtime and receive a result over WebSocket as text or typed JSON.
* **Pre-agent downloadables**: fetch code, repos, or data into the sandbox before the run starts.
* **Pre-commands**: run bash commands before the agent starts executing.
* **Artifact uploads**: move generated files out of the sandbox with a presigned URL handshake.
* **Streaming and cancellation**: receive progress updates and stop runs cleanly.
* **Secret-aware execution**: redact sensitive values before they leave the sandbox.
* **Persistent sessions**: run many sequential invocations and commands over a single WebSocket connection.


# Python Client (/python-client)



The [Python client](https://pypi.org/project/runtimeuse-client/) is the control plane for RuntimeUse. It connects to the sandbox runtime, sends the invocation, and turns runtime messages into a single `QueryResult`.

```bash
pip install runtimeuse-client
```

Basic Query [#basic-query]

```python
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)

Persistent Sessions [#persistent-sessions]

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.

```python
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 [#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`.

```python
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 [#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.

```python
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 [#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.

```python
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 [#upload-artifacts]

When the runtime requests an artifact upload, return a presigned URL and content type from `on_artifact_upload_request`. Set `artifacts_dirs` to a list of sandbox directories the runtime should watch for files to upload - both options must be provided together. Pass multiple paths to watch several directories within a single invocation; each may carry its own `.artifactignore`.

```python
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 a screenshot.",
    options=QueryOptions(
        system_prompt="You are a helpful assistant.",
        model="gpt-4.1",
        artifacts_dirs=["/runtimeuse/output", "/runtimeuse/screenshots"],
        on_artifact_upload_request=on_artifact_upload_request,
    ),
)
```

Stream Assistant Messages [#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.

```python
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 [#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.

```python
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 [#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`.

```python
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 [#set-a-timeout]

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

```python
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 [#redact-secrets]

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

```python
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 [#handle-errors]

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

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


# Quickstart (/quickstart)



import { Tab, Tabs } from 'fumadocs-ui/components/tabs';

1. Start the Runtime [#1-start-the-runtime]

<Tabs items={["Local", "E2B", "Daytona", "Vercel", "Modal"]}>
  <Tab>
    ```bash
    npm install -g @anthropic-ai/claude-code
    export ANTHROPIC_API_KEY=your_anthropic_api_key
    npx -y runtimeuse@latest --agent claude
    ```

    This starts the Claude Code agent on port `8080`. To use OpenAI agent instead:

    ```bash
    export OPENAI_API_KEY=your_openai_api_key
    npx -y runtimeuse@latest
    ```

    ```python
    ws_url = "ws://localhost:8080"
    ```
  </Tab>

  <Tab>
    ```bash
    pip install e2b
    ```

    ```python
    # Required env vars: E2B_API_KEY, ANTHROPIC_API_KEY
    from e2b import Template, wait_for_port, Sandbox

    template = (
        Template()
        .from_node_image("lts")
        .set_workdir("/runtimeuse")
        .npm_install(["@anthropic-ai/claude-code"], g=True)
        .set_envs(
            {
                "ANTHROPIC_API_KEY": anthropic_api_key,
                "IS_SANDBOX": "1",
                "CLAUDE_SKIP_ROOT_CHECK": "1",
            }
        )
        .set_start_cmd("npx -y runtimeuse@latest --agent claude", wait_for_port(8080))
    )

    sandbox = Sandbox.create(template="runtimeuse-quickstart-claude", api_key=e2b_api_key)
    ws_url = f"wss://{sandbox.get_host(8080)}"
    ```

    Full example: [examples/e2b-quickstart.py](https://github.com/getlark/runtimeuse/blob/main/examples/e2b-quickstart.py)
  </Tab>

  <Tab>
    ```bash
    pip install daytona
    ```

    ```python
    # Required env vars: DAYTONA_API_KEY, ANTHROPIC_API_KEY
    from daytona import (
        CreateSandboxFromImageParams,
        Daytona,
        DaytonaConfig,
        Image,
        SessionExecuteRequest,
    )

    image = Image.base("node:lts").run_commands(
        "apt-get update && apt-get install -y unzip",
        "npm install -g @anthropic-ai/claude-code",
    )

    daytona = Daytona(config=DaytonaConfig(api_key=daytona_api_key))

    sandbox = daytona.create(
        CreateSandboxFromImageParams(
            image=image,
            env_vars={
                "ANTHROPIC_API_KEY": anthropic_api_key,
                "IS_SANDBOX": "1",
                "CLAUDE_SKIP_ROOT_CHECK": "1",
            },
            public=True,
        ),
        timeout=600,
    )

    sandbox.process.create_session("runtimeuse")
    sandbox.process.execute_session_command(
        "runtimeuse",
        SessionExecuteRequest(
            command="npx -y runtimeuse@latest --agent claude",
            run_async=True,
        ),
    )

    preview = sandbox.create_signed_preview_url(8080, expires_in_seconds=3600)
    ws_url = _http_to_ws(preview.url)
    ```

    Full example: [examples/daytona-quickstart.py](https://github.com/getlark/runtimeuse/blob/main/examples/daytona-quickstart.py)
  </Tab>

  <Tab>
    ```bash
    pip install vercel python-dotenv
    ```

    ```python
    # Required env vars: VERCEL_TOKEN, VERCEL_PROJECT_ID, VERCEL_TEAM_ID, ANTHROPIC_API_KEY
    from vercel.sandbox import Sandbox

    sandbox = Sandbox.create(
        runtime="node24",
        ports=[8081],
        env={
            "ANTHROPIC_API_KEY": anthropic_api_key,
            "IS_SANDBOX": "1",
            "CLAUDE_SKIP_ROOT_CHECK": "1",
        },
    )

    sandbox.run_command("sudo", ["dnf", "install", "-y", "unzip"])
    sandbox.run_command("npm", ["install", "-g", "@anthropic-ai/claude-code"])
    sandbox.run_command_detached(
        "npx",
        ["-y", "runtimeuse", "--agent", "claude", "--port", "8081"],
    )

    ws_url = _http_to_ws(sandbox.domain(8081))
    ```

    Full example: [examples/vercel-quickstart.py](https://github.com/getlark/runtimeuse/blob/main/examples/vercel-quickstart.py)
  </Tab>

  <Tab>
    ```bash
    pip install modal
    ```

    ```python
    # Required env vars: ANTHROPIC_API_KEY
    # Authenticate with Modal: `modal token set` or set MODAL_TOKEN_ID + MODAL_TOKEN_SECRET
    import modal

    app = modal.App.lookup("runtimeuse-quickstart", create_if_missing=True)

    image = modal.Image.from_registry("node:lts").run_commands(
        "apt-get update && apt-get install -y unzip",
        "npm install -g @anthropic-ai/claude-code",
    )

    secret = modal.Secret.from_dict(
        {
            "ANTHROPIC_API_KEY": anthropic_api_key,
            "IS_SANDBOX": "1",
            "CLAUDE_SKIP_ROOT_CHECK": "1",
        }
    )

    sandbox = modal.Sandbox.create(
        app=app,
        image=image,
        secrets=[secret],
        workdir="/runtimeuse",
        encrypted_ports=[8080],
        timeout=600,
    )

    sandbox.exec("npx", "-y", "runtimeuse", "--agent", "claude")
    ws_url = _http_to_ws(sandbox.tunnels()[8080].url)
    ```

    Full example: [examples/modal-quickstart.py](https://github.com/getlark/runtimeuse/blob/main/examples/modal-quickstart.py)
  </Tab>
</Tabs>

2. Install the Client [#2-install-the-client]

```bash
pip install runtimeuse-client
```

3. Connect and Query [#3-connect-and-query]

Once you have a `ws_url`, the client flow is the same across providers:

```python
import asyncio

from runtimeuse_client import (
    QueryOptions,
    RuntimeEnvironmentDownloadableInterface,
    RuntimeUseClient,
    TextResult,
)


async def main(ws_url: str) -> None:
    client = RuntimeUseClient(ws_url=ws_url)

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

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


asyncio.run(main(ws_url))
```