Sync versus async patterns

This guide helps you choose between synchronous and asynchronous patterns in the CoreWeave sandbox Python SDK (cwsandbox) and shows the sync and async form of each common operation. The SDK has a single async implementation internally. The sync/async flexibility comes from how you consume results: use .result() for sync and await for async.

Choose authentication

These examples run on serverless capacity unless a section specifies CKS. Choose the credential for the account that owns your sandboxes. See Choose and get credentials to obtain a key or token. Run the installation commands in an active Python virtual environment. Save the Python code from your credential tab as sandbox_auth.py beside the Python scripts you run from this page. It stores the authentication choice, not your credential. The examples import AUTH from that file and pass it explicitly to each new sandbox or session.

Forge API key
CoreWeave API access token

Use your Forge API key through the SDK’s W&B authentication strategy:

uv pip install 'cwsandbox[wandb]>=1.14.2'
export WANDB_API_KEY="[WANDB-API-KEY]"

sandbox_auth.py

from cwsandbox import AuthStrategy

AUTH = AuthStrategy.WANDB

Use the token for your CoreWeave organization:

uv pip install 'cwsandbox>=1.14.2'
export CWSANDBOX_API_KEY="[API-ACCESS-TOKEN]"

sandbox_auth.py

from cwsandbox import AuthStrategy

AUTH = AuthStrategy.COREWEAVE_API_KEY

For CKS placement, choose the CoreWeave token, enable a runner, and set placement_mode="cks" on sandbox creation or in SandboxDefaults. Creating a session with auth=AUTH also authenticates the sandboxes it owns. Keep the same credential when reconnecting, listing, or cleaning up sandboxes.

Quick decision guide

Use case	Pattern	Reason
Most operations	Sync	Simpler code, no asyncio boilerplate
Parallel execution	Sync	Operations are non-blocking by design
Jupyter notebooks	Sync	No `nest_asyncio` needed
Async codebase	Async	Integrates with existing async code

Rule of thumb: Use sync patterns (.result()) for simplicity. Switch to await only when you’re already in an async codebase.

Core concept: OperationRef

Most SDK methods return OperationRef[T], a wrapper that is both .result()-able and awaitable. Both paths raise the same exceptions.

ref = sandbox.read_file("/output/data.txt")

# Sync: block for the value
data = ref.result()

# Async: await instead
data = await ref

Sandbox.exec() returns Process, which extends OperationRef[ProcessResult] with streaming and stdin.

Never use .result() in async contexts. Calling .result() blocks the thread. In an async context this blocks the event loop and can deadlock your application. Use await instead.

Auto-start behavior

session.sandbox() returns an unstarted sandbox. The sandbox auto-starts on the first operation that requires it: Triggers auto-start: exec(), read_file(), write_file(), wait(), wait_until_complete() Does not auto-start: get_status() (raises SandboxNotRunningError), stop() (no-op if never started)

Operations

Create sandboxes

Sync
Async

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# Context manager (recommended)
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sb:
    result = sb.exec(["echo", "hello"]).result()
# Automatically stopped on exit

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# Construct, then use an async context manager for startup and cleanup
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
async with sb as sb:
    result = await sb.exec(["echo", "hello"])
# Automatically stopped on exit

Sandbox.run() calls start().result() internally, which blocks the event loop. In async code, construct with Sandbox(...) and use async with to start the sandbox. exec() waits for RUNNING before executing. To wait for startup separately, use await sandbox and check its status before operations.

start()

Sync
Async

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sb.start().result()  # Block until backend accepts
print(sb.sandbox_id)

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
await sb.start()  # Await until backend accepts
print(sb.sandbox_id)

exec()

Sync
Async

result = sb.exec(["echo", "hello"]).result()
print(result.stdout)  # "hello\n"

# Raise on non-zero exit code
result = sb.exec(["python", "-c", "exit(1)"], check=True).result()

result = await sb.exec(["echo", "hello"])
print(result.stdout)  # "hello\n"

# Raise on non-zero exit code
result = await sb.exec(["python", "-c", "exit(1)"], check=True)

read_file()

Sync
Async

data = sb.read_file("/output/result.txt").result()
print(data.decode())

data = await sb.read_file("/output/result.txt")
print(data.decode())

write_file()

Sync
Async

sb.write_file("/input/data.txt", b"content").result()

await sb.write_file("/input/data.txt", b"content")

stop()

Sync
Async

sb.stop().result()

# Ignore if already deleted
sb.stop(missing_ok=True).result()

await sb.stop()

# Ignore if already deleted
await sb.stop(missing_ok=True)

wait()

wait() waits for startup status. For example, it can return when the sandbox is RUNNING or TERMINATING, or after it completes successfully. It’s sync-only. In async code, await sandbox uses the same startup wait without blocking the event loop. A normal return doesn’t guarantee RUNNING. Check status before operations that require a running sandbox. For terminal-state and error handling, see Wait for a sandbox.

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

# Sync: wait for startup status, then check status
sb = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH).wait()
if sb.status == SandboxStatus.RUNNING:
    result = sb.exec(["echo", "ready"]).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

# Async: await the same startup wait, then check status
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
await sb
if sb.status == SandboxStatus.RUNNING:
    result = await sb.exec(["echo", "ready"])

wait_until_complete()

Sync
Async

from sandbox_auth import AUTH
sb = Sandbox.run(
    "python",
    "-c",
    "print('done')",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
sb.wait_until_complete().result()
print(f"Exit code: {sb.returncode}")

# Handle externally-terminated sandboxes without raising
sb.wait_until_complete(raise_on_termination=False).result()

from sandbox_auth import AUTH
sb = Sandbox(
    command="python",
    args=["-c", "print('done')"],
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
await sb.wait_until_complete()
print(f"Exit code: {sb.returncode}")

# Handle externally-terminated sandboxes without raising
await sb.wait_until_complete(raise_on_termination=False)

get_status()

get_status() is sync-only. It fetches the current status from the API.

status = sb.get_status()
print(f"Sandbox is {status}")  # e.g. SandboxStatus.RUNNING

Sandbox.list()

Sync
Async

from sandbox_auth import AUTH
sandboxes = Sandbox.list(tags=["my-job"], auth=AUTH).result()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

from sandbox_auth import AUTH
sandboxes = await Sandbox.list(tags=["my-job"], auth=AUTH)
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

Sandbox.from_id()

Sync
Async

from sandbox_auth import AUTH
sb = Sandbox.from_id("sandbox-abc123", auth=AUTH).result()
result = sb.exec(["echo", "reconnected"]).result()

from sandbox_auth import AUTH
sb = await Sandbox.from_id("sandbox-abc123", auth=AUTH)
result = await sb.exec(["echo", "reconnected"])

Sandbox.delete()

Sync
Async

from sandbox_auth import AUTH
Sandbox.delete("sandbox-abc123", auth=AUTH).result()

# Ignore if already deleted
Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH).result()

from sandbox_auth import AUTH
await Sandbox.delete("sandbox-abc123", auth=AUTH)

# Ignore if already deleted
await Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH)

session.list()

Sync
Async

sandboxes = session.list().result()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

sandboxes = await session.list()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

session.from_id()

Sync
Async

sb = session.from_id("sandbox-abc123").result()
result = sb.exec(["echo", "adopted"]).result()

sb = await session.from_id("sandbox-abc123")
result = await sb.exec(["echo", "adopted"])

Stream stdout

Sync
Async

process = sb.exec(["python", "-c", "import time; [print(i) or time.sleep(0.1) for i in range(5)]"])
for line in process.stdout:
    print(line, end="")
result = process.result()

process = sb.exec(["python", "-c", "import time; [print(i) or time.sleep(0.1) for i in range(5)]"])
async for line in process.stdout:
    print(line, end="")
result = await process

Stream stdin

Enable stdin with stdin=True. Use write() for raw bytes, writeline() for text lines, and close() to signal EOF.

Sync
Async

process = sb.exec(["cat"], stdin=True)
process.stdin.write(b"hello ").result()
process.stdin.writeline("world").result()
process.stdin.close().result()  # Signals EOF to the process
result = process.result()
print(result.stdout)  # "hello world\n"

process = sb.exec(["cat"], stdin=True)
await process.stdin.write(b"hello ")
await process.stdin.writeline("world")
await process.stdin.close()  # Signals EOF to the process
result = await process
print(result.stdout)  # "hello world\n"

cwsandbox.results()

cwsandbox.results() is a sync-only batch helper. It calls .result() on one or more OperationRef objects.

import cwsandbox

# Single ref
data = cwsandbox.results(sandbox.read_file("/path"))

# Multiple refs
all_data = cwsandbox.results([sb.read_file(f) for f in files])

cwsandbox.wait()

cwsandbox.wait() is sync-only. It waits for a sequence of Sandbox, OperationRef, or Process objects and returns (done, pending). For sandboxes, it uses the same startup wait as Sandbox.wait(), rather than waiting for the main command to finish. A sandbox in done isn’t guaranteed to be RUNNING. For example, it may already be completed or terminating. Check each sandbox’s status before operations that require it to be running.

from sandbox_auth import AUTH
import cwsandbox

# Wait for all sandbox startup waits to finish
sandboxes = [
    Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(5)
]
done, pending = cwsandbox.wait(sandboxes)

# Wait for first 2 operations to complete
refs = [sb.read_file(f) for f in files]
done, pending = cwsandbox.wait(refs, num_returns=2)

# Wait with timeout
done, pending = cwsandbox.wait(procs, timeout=30.0)

@session.function()

Sync
Async

from sandbox_auth import AUTH
with Session(defaults, auth=AUTH) as session:
    @session.function()
    def compute(x: int, y: int) -> int:
        return x + y

    result = compute.remote(2, 3).result()
    print(result)  # 5

    # Parallel map
    refs = compute.map([(1, 2), (3, 4), (5, 6)])
    results = [r.result() for r in refs]  # [3, 7, 11]

from sandbox_auth import AUTH
async with Session(defaults, auth=AUTH) as session:
    @session.function()
    def compute(x: int, y: int) -> int:
        return x + y

    result = await compute.remote(2, 3)
    print(result)  # 5

    # Parallel map
    refs = compute.map([(1, 2), (3, 4), (5, 6)])
    results = [await r for r in refs]  # [3, 7, 11]

Parallel execution

The sync API supports parallel execution because operations are non-blocking by design. Methods like exec(), read_file(), and write_file() return immediately. You only block when you call .result(). Sandbox.run() blocks until the backend accepts the start request (it calls start().result() internally). For parallel startup, use session.sandbox() (auto-starts on first operation) or collect start() refs:

from sandbox_auth import AUTH
from cwsandbox import Sandbox, Session, SandboxDefaults

# Option 1: session.sandbox() - sandboxes auto-start on first exec()
with Session(
    SandboxDefaults(resources={"cpu": "1", "memory": "1Gi"}),
    auth=AUTH,
) as session:
    sandboxes = [session.sandbox() for _ in range(3)]
    processes = [sb.exec(["echo", f"sb-{i}"]) for i, sb in enumerate(sandboxes)]
    results = [p.result() for p in processes]

# Option 2: Sandbox() + collect start() refs
sandboxes = [
    Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(3)
]
start_refs = [sb.start() for sb in sandboxes]
for ref in start_refs:
    ref.result()  # All starts proceed in parallel

processes = [sb.exec(["echo", f"sb-{i}"]) for i, sb in enumerate(sandboxes)]
results = [p.result() for p in processes]
for sb in sandboxes:
    sb.stop().result()

Jupyter notebooks

The sync API works in Jupyter without nest_asyncio because the SDK runs its own background event loop in a daemon thread:

from sandbox_auth import AUTH
# Cell 1 - Create sandbox
from cwsandbox import Sandbox, SandboxStatus
sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # Wait for startup status

# Cell 2 - Check status before executing commands
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["python", "-c", "print(1+1)"]).result()
    print(result.stdout)

# Cell 3 - Cleanup
sandbox.stop().result()

For async in Jupyter, await works directly since Jupyter has a built-in event loop:

from sandbox_auth import AUTH
# Works in Jupyter without asyncio.run()
sandboxes = await Sandbox.list(tags=["notebook"], auth=AUTH)

Error handling

Both sync and async paths raise the same exceptions. .result() re-raises any exception from the underlying operation, and await does the same. See What do common sandbox SDK error messages mean? for the full exception hierarchy and recovery patterns.

Sandboxes

Use sandboxes

Manage sandboxes

Reference

Sync versus async patterns

Choose authentication

Quick decision guide

Core concept: OperationRef

Auto-start behavior

Operations

Create sandboxes

start()

exec()

read_file()

write_file()

stop()

wait()

wait_until_complete()

get_status()

Sandbox.list()

Sandbox.from_id()

Sandbox.delete()

session.list()

session.from_id()

Stream stdout

Stream stdin

cwsandbox.results()

cwsandbox.wait()

@session.function()

Parallel execution

Jupyter notebooks

Error handling

​Choose authentication

​Quick decision guide

​Core concept: OperationRef

​Auto-start behavior

​Operations

​Create sandboxes

​start()

​exec()

​read_file()

​write_file()

​stop()

​wait()

​wait_until_complete()

​get_status()

​Sandbox.list()

​Sandbox.from_id()

​Sandbox.delete()

​session.list()

​session.from_id()

​Stream stdout

​Stream stdin

​cwsandbox.results()

​cwsandbox.wait()

​@session.function()

​Parallel execution

​Jupyter notebooks

​Error handling

Choose authentication

Quick decision guide

Core concept: OperationRef

Auto-start behavior

Operations

Create sandboxes

start()

exec()

read_file()

write_file()

stop()

wait()

wait_until_complete()

get_status()

Sandbox.list()

Sandbox.from_id()

Sandbox.delete()

session.list()

session.from_id()

Stream stdout

Stream stdin

cwsandbox.results()

cwsandbox.wait()

@session.function()

Parallel execution

Jupyter notebooks

Error handling