> ## Documentation Index
> Fetch the complete documentation index at: https://docs.declaw.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# AsyncSandbox

> Async-native sandbox class for concurrent workloads, multiple sandboxes, and async frameworks such as FastAPI and LangGraph.

```python theme={null}
from declaw import AsyncSandbox
```

`AsyncSandbox` is the async counterpart of `Sandbox`. Every method is a coroutine (`async def`) and must be awaited. It is the preferred choice when:

* You need to create or manage multiple sandboxes concurrently with `asyncio.gather`.
* Your application is built on an async framework (FastAPI, LangGraph, LiteLLM, etc.).
* You want to stream command output using `async for` or `asyncio.Queue`.

## When to use async

```python theme={null}
import asyncio
from declaw import AsyncSandbox

async def run_parallel():
    # Create three sandboxes concurrently
    sandboxes = await asyncio.gather(
        AsyncSandbox.create(api_key="key", domain="host:8080"),
        AsyncSandbox.create(api_key="key", domain="host:8080"),
        AsyncSandbox.create(api_key="key", domain="host:8080"),
    )
    # Run commands in all three simultaneously
    results = await asyncio.gather(
        *[sbx.run_command("hostname") for sbx in sandboxes]
    )
    for sbx, result in zip(sandboxes, results):
        print(result.stdout.strip())
        await sbx.kill()

asyncio.run(run_parallel())
```

## Class methods

### `AsyncSandbox.create()`

Create a new sandbox and return a connected `AsyncSandbox` instance.

```python theme={null}
sbx = await AsyncSandbox.create(
    template="base",
    timeout=300,
    envs={"MY_VAR": "hello"},
    api_key="your-api-key",
    domain="104.198.24.180:8080",
)
```

All parameters are identical to [`Sandbox.create()`](/sdks/python/sandbox#sandboxcreate).

**Returns** `AsyncSandbox`

***

### `AsyncSandbox.connect()`

Connect to an existing sandbox by ID.

```python theme={null}
sbx = await AsyncSandbox.connect(
    sandbox_id="abc123",
    api_key="your-api-key",
    domain="104.198.24.180:8080",
)
```

All parameters are identical to [`Sandbox.connect()`](/sdks/python/sandbox#sandboxconnect).

**Returns** `AsyncSandbox`

***

## Instance methods

### `await sbx.kill()`

Kill and destroy the sandbox.

```python theme={null}
await sbx.kill()
```

**Returns** `bool`

***

### `await sbx.is_running()`

Check whether the sandbox is in the `running` state.

```python theme={null}
running = await sbx.is_running()
```

**Returns** `bool`

***

### `await sbx.set_timeout()`

Update the sandbox timeout.

```python theme={null}
await sbx.set_timeout(600)
```

<ParamField body="timeout" type="int" required>
  New timeout in seconds.
</ParamField>

**Returns** `None`

***

### `await sbx.get_info()`

Fetch the current metadata and state.

```python theme={null}
info = await sbx.get_info()
print(info.state)
```

**Returns** `SandboxInfo`

***

### `await sbx.get_metrics()`

Retrieve resource usage metrics for a time range.

```python theme={null}
import datetime
metrics = await sbx.get_metrics(
    start=datetime.datetime.utcnow() - datetime.timedelta(minutes=5),
)
```

**Returns** `list[SandboxMetrics]`

***

### `await sbx.pause()`

Pause the sandbox.

```python theme={null}
await sbx.pause()
```

**Returns** `None`

***

### `await sbx.create_snapshot()`

Create a sandbox snapshot.

```python theme={null}
snap = await sbx.create_snapshot()
print(snap.snapshot_id)
```

**Returns** `SnapshotInfo`

***

### `await sbx.snapshot()`

Create a manual snapshot of this sandbox. Manual snapshots accumulate — every
call creates a new persistent checkpoint that survives `sbx.kill()`. Use
[`AsyncSandbox.restore()`](#asyncsandbox-restore) or
[`sbx.list_snapshots()`](#await-sbx-list_snapshots) to retrieve and fork from
them.

```python theme={null}
snap = await sbx.snapshot()
print(snap.snapshot_id)
```

<ParamField body="request_timeout" type="float | None" default="None">
  Per-request HTTP timeout in seconds.
</ParamField>

**Returns** `Snapshot`

***

### `await sbx.list_snapshots()`

List all snapshots (periodic, pause, and manual) for this sandbox, newest first.

```python theme={null}
for snap in await sbx.list_snapshots():
    print(snap.snapshot_id, snap.created_at)
```

<ParamField body="request_timeout" type="float | None" default="None">
  Per-request HTTP timeout in seconds.
</ParamField>

**Returns** `list[Snapshot]`

***

### `await AsyncSandbox.restore()`

Restore a sandbox from a snapshot. The restored sandbox may run on a different
worker than the original. Returns a usable `AsyncSandbox` instance already
connected to the restored sandbox.

```python theme={null}
sbx = await AsyncSandbox.restore("sbx-a1b2c3d4", snapshot_id="snap-xyz")
```

<ParamField body="sandbox_id" type="str" required>
  The sandbox to restore.
</ParamField>

<ParamField body="snapshot_id" type="str | None" default="None">
  Specific snapshot to restore from. If omitted, the most recent snapshot is
  used (preference order: pause > periodic > manual).
</ParamField>

<ParamField body="api_key" type="str | None" default="$DECLAW_API_KEY">
  API key override.
</ParamField>

<ParamField body="domain" type="str | None" default="$DECLAW_DOMAIN">
  Domain override.
</ParamField>

<ParamField body="request_timeout" type="float | None" default="None">
  Per-request HTTP timeout in seconds.
</ParamField>

**Returns** `AsyncSandbox`

***

## Command methods (inline on AsyncSandbox)

Unlike the synchronous `Sandbox`, `AsyncSandbox` exposes command operations directly as methods rather than through a sub-module:

### `await sbx.run_command()`

Run a command and return its result (or a handle if `background=True`).

```python theme={null}
from declaw import CommandResult, AsyncCommandHandle

result: CommandResult = await sbx.run_command("echo hello")

# Background command
handle: AsyncCommandHandle = await sbx.run_command("sleep 10", background=True)
```

<ParamField body="cmd" type="str" required>
  Shell command to execute.
</ParamField>

<ParamField body="background" type="bool" default="False">
  When `True`, returns an `AsyncCommandHandle` immediately without waiting for
  the command to finish.
</ParamField>

<ParamField body="envs" type="dict[str, str] | None" default="None">
  Environment variables for the command.
</ParamField>

<ParamField body="user" type="str" default="'user'">
  Unix user to run the command as.
</ParamField>

<ParamField body="cwd" type="str | None" default="None">
  Working directory for the command.
</ParamField>

<ParamField body="on_stdout" type="Callable[[str], None] | None" default="None">
  Callback invoked for each stdout line after completion (foreground only).
</ParamField>

<ParamField body="on_stderr" type="Callable[[str], None] | None" default="None">
  Callback invoked for each stderr line after completion (foreground only).
</ParamField>

<ParamField body="timeout" type="float | None" default="60">
  Command execution timeout in seconds.
</ParamField>

**Returns** `CommandResult | AsyncCommandHandle`

***

### `await sbx.list_commands()`

List all running processes in the sandbox.

```python theme={null}
processes = await sbx.list_commands()
```

**Returns** `list[ProcessInfo]`

***

### `await sbx.kill_command()`

Kill a running command by PID.

```python theme={null}
killed = await sbx.kill_command(pid=1234)
```

<ParamField body="pid" type="int" required>
  Process ID to kill.
</ParamField>

**Returns** `bool`

***

## Filesystem methods (inline on AsyncSandbox)

### `await sbx.read_file()`

Read a file's content.

```python theme={null}
content: str = await sbx.read_file("/home/user/script.py")
raw: bytearray = await sbx.read_file("/data/image.png", format="bytes")
```

<ParamField body="path" type="str" required>
  Absolute path inside the sandbox.
</ParamField>

<ParamField body="format" type="str" default="'text'">
  `"text"` returns a `str`; `"bytes"` returns a `bytearray`.
</ParamField>

<ParamField body="user" type="str" default="'user'">
  Unix user context.
</ParamField>

**Returns** `str | bytearray`

***

### `await sbx.write_file()`

Write content to a file.

```python theme={null}
info = await sbx.write_file("/home/user/hello.py", "print('hello')")
```

<ParamField body="path" type="str" required>
  Absolute path inside the sandbox.
</ParamField>

<ParamField body="data" type="str | bytes" required>
  Content to write.
</ParamField>

**Returns** `WriteInfo`

***

### `await sbx.list_files()`

List directory entries.

```python theme={null}
entries = await sbx.list_files("/home/user")
```

<ParamField body="path" type="str" required>
  Directory path.
</ParamField>

<ParamField body="depth" type="int" default="1">
  Recursion depth. `1` lists only the immediate directory.
</ParamField>

**Returns** `list[EntryInfo]`

***

## Async context manager

```python theme={null}
async with await AsyncSandbox.create(api_key="key", domain="host:8080") as sbx:
    result = await sbx.run_command("python3 --version")
    print(result.stdout)
    await sbx.kill()
```

`__aexit__` calls `await sbx.close()` which releases the HTTP client. It does **not** kill the sandbox — call `sbx.kill()` explicitly.

***

## AsyncCommandHandle

Returned by `sbx.run_command(background=True)`. Allows you to wait for or kill the background process.

```python theme={null}
handle = await sbx.run_command("python3 long_script.py", background=True)
print("PID:", handle.pid)
result = await handle.wait()
```

| Method                                    | Returns         | Description                                                          |
| ----------------------------------------- | --------------- | -------------------------------------------------------------------- |
| `await handle.wait(on_stdout, on_stderr)` | `CommandResult` | Wait for completion. Raises `CommandExitException` on non-zero exit. |
| `await handle.kill()`                     | `bool`          | Send SIGKILL to the process.                                         |
| `handle.pid`                              | `int`           | Process ID.                                                          |