Skip to main content
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

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. 
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(). Returns AsyncSandbox

​
AsyncSandbox.connect()

Connect to an existing sandbox by ID. 
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(). Returns AsyncSandbox

​
Instance methods

​
await sbx.kill()

Kill and destroy the sandbox. 
await sbx.kill()
Returns bool

​
await sbx.is_running()

Check whether the sandbox is in the running state. 
running = await sbx.is_running()
Returns bool

​
await sbx.set_timeout()

Update the sandbox timeout. 
await sbx.set_timeout(600)
​
timeout
int
required
New timeout in seconds.
Returns None

​
await sbx.get_info()

Fetch the current metadata and state. 
info = await sbx.get_info()
print(info.state)
Returns SandboxInfo

​
await sbx.get_metrics()

Retrieve resource usage metrics for a time range. 
import datetime
metrics = await sbx.get_metrics(
    start=datetime.datetime.utcnow() - datetime.timedelta(minutes=5),
)
Returns list[SandboxMetrics]

​
await sbx.pause()

Pause the sandbox. 
await sbx.pause()
Returns None

​
await sbx.create_snapshot()

Create a sandbox snapshot. 
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() or sbx.list_snapshots() to retrieve and fork from them. 
snap = await sbx.snapshot()
print(snap.snapshot_id)
​
request_timeout
float | None
default:"None"
Per-request HTTP timeout in seconds.
Returns Snapshot

​
await sbx.list_snapshots()

List all snapshots (periodic, pause, and manual) for this sandbox, newest first. 
for snap in await sbx.list_snapshots():
    print(snap.snapshot_id, snap.created_at)
​
request_timeout
float | None
default:"None"
Per-request HTTP timeout in seconds.
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. 
sbx = await AsyncSandbox.restore("sbx-a1b2c3d4", snapshot_id="snap-xyz")
​
sandbox_id
str
required
The sandbox to restore.
​
snapshot_id
str | None
default:"None"
Specific snapshot to restore from. If omitted, the most recent snapshot is used (preference order: pause > periodic > manual).
​
api_key
str | None
default:"$DECLAW_API_KEY"
API key override.
​
domain
str | None
default:"$DECLAW_DOMAIN"
Domain override.
​
request_timeout
float | None
default:"None"
Per-request HTTP timeout in seconds.
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). 
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)
​
cmd
str
required
Shell command to execute.
​
background
bool
default:"False"
When True, returns an AsyncCommandHandle immediately without waiting for the command to finish.
​
envs
dict[str, str] | None
default:"None"
Environment variables for the command.
​
user
str
default:"'user'"
Unix user to run the command as.
​
cwd
str | None
default:"None"
Working directory for the command.
​
on_stdout
Callable[[str], None] | None
default:"None"
Callback invoked for each stdout line after completion (foreground only).
​
on_stderr
Callable[[str], None] | None
default:"None"
Callback invoked for each stderr line after completion (foreground only).
​
timeout
float | None
default:"60"
Command execution timeout in seconds.
Returns CommandResult | AsyncCommandHandle

​
await sbx.list_commands()

List all running processes in the sandbox. 
processes = await sbx.list_commands()
Returns list[ProcessInfo]

​
await sbx.kill_command()

Kill a running command by PID. 
killed = await sbx.kill_command(pid=1234)
​
pid
int
required
Process ID to kill.
Returns bool

​
Filesystem methods (inline on AsyncSandbox)

​
await sbx.read_file()

Read a file’s content. 
content: str = await sbx.read_file("/home/user/script.py")
raw: bytearray = await sbx.read_file("/data/image.png", format="bytes")
​
path
str
required
Absolute path inside the sandbox.
​
format
str
default:"'text'"
"text" returns a str; "bytes" returns a bytearray.
​
user
str
default:"'user'"
Unix user context.
Returns str | bytearray

​
await sbx.write_file()

Write content to a file. 
info = await sbx.write_file("/home/user/hello.py", "print('hello')")
​
path
str
required
Absolute path inside the sandbox.
​
data
str | bytes
required
Content to write.
Returns WriteInfo

​
await sbx.list_files()

List directory entries. 
entries = await sbx.list_files("/home/user")
​
path
str
required
Directory path.
​
depth
int
default:"1"
Recursion depth. 1 lists only the immediate directory.
Returns list[EntryInfo]

​
Async context manager

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. 
handle = await sbx.run_command("python3 long_script.py", background=True)
print("PID:", handle.pid)
result = await handle.wait()
MethodReturnsDescription
await handle.wait(on_stdout, on_stderr)CommandResultWait for completion. Raises CommandExitException on non-zero exit.
await handle.kill()boolSend SIGKILL to the process.
handle.pidintProcess ID.