Documentation Index
Fetch the complete documentation index at: https://mintlify.com/trycua/cua/llms.txt
Use this file to discover all available pages before exploring further.
Because every Sandbox method is async, Python’s asyncio primitives let you run many sandboxes concurrently without threads. You can fan out a fixed list of tasks with asyncio.gather, cap peak concurrency with asyncio.Semaphore, or drain a dynamic work queue with asyncio.Queue. The right pattern depends on whether you know all jobs upfront and how much you want to constrain resource use.
Every local sandbox consumes host CPU, memory, and disk. Start with two or three concurrent sandboxes, observe resource usage, and scale up gradually. Cloud sandboxes are limited by your cua.ai plan quota rather than host hardware.
Run a fixed set concurrently
Use asyncio.gather when you have the complete list of jobs before any sandbox starts. All sandboxes launch at once and the call returns when every job finishes.
import asyncio
from cua import Sandbox, Image
async def run_task(task: str) -> str:
async with Sandbox.ephemeral(Image.linux(), local=True) as sb:
result = await sb.shell.run(f"echo '{task}'")
return result.stdout.strip()
async def main():
tasks = ["task-1", "task-2", "task-3", "task-4"]
results = await asyncio.gather(*[run_task(t) for t in tasks])
for task, output in zip(tasks, results):
print(f"{task}: {output}")
asyncio.run(main())
Cap the number of concurrent sandboxes
Use asyncio.Semaphore to set a hard ceiling on how many sandboxes run at any given moment. New tasks wait at the semaphore boundary until a running task completes and releases its slot.
import asyncio
from cua import Sandbox, Image
MAX_CONCURRENT = 5
async def process_item(sem: asyncio.Semaphore, item: str) -> dict:
async with sem: # blocks if MAX_CONCURRENT are already running
async with Sandbox.ephemeral(Image.linux(), local=True) as sb:
result = await sb.shell.run(f"python /app/process.py '{item}'")
return {
"item": item,
"output": result.stdout,
"ok": result.success,
}
async def main():
items = [f"item-{i}" for i in range(20)]
sem = asyncio.Semaphore(MAX_CONCURRENT)
results = await asyncio.gather(*[process_item(sem, item) for item in items])
passed = sum(r["ok"] for r in results)
print(f"{passed}/{len(results)} succeeded")
asyncio.run(main())
A Semaphore value of 5 is a reasonable starting point for local Docker containers. For local QEMU VMs, start at 2–3 because each VM reserves a fixed memory allocation.
Process a dynamic queue of tasks
Use asyncio.Queue when tasks arrive at runtime — for example from a message broker, a generator, or user input. A fixed pool of workers drains the queue; each worker runs one sandbox per task.
import asyncio
from cua import Sandbox, Image
async def worker(queue: asyncio.Queue, worker_id: int):
while True:
task = await queue.get()
if task is None: # sentinel: shut down this worker
break
async with Sandbox.ephemeral(Image.linux(), local=True) as sb:
result = await sb.shell.run(task)
print(f"[worker-{worker_id}] {result.stdout.strip()}")
queue.task_done()
async def main():
queue: asyncio.Queue = asyncio.Queue()
num_workers = 4
# Start worker coroutines
workers = [asyncio.create_task(worker(queue, i)) for i in range(num_workers)]
# Enqueue work
for i in range(10):
await queue.put(f"echo 'job {i}'")
# Send one sentinel per worker to signal shutdown
for _ in range(num_workers):
await queue.put(None)
await asyncio.gather(*workers)
asyncio.run(main())
Handle failures without aborting the whole batch
By default asyncio.gather re-raises the first exception and cancels pending tasks. Pass return_exceptions=True to collect failures alongside successes and decide what to do after all tasks finish.
import asyncio
from cua import Sandbox, Image
async def run_task(task: str) -> str:
async with Sandbox.ephemeral(Image.linux(), local=True) as sb:
result = await sb.shell.run(f"python /app/{task}.py")
if not result.success:
raise RuntimeError(f"task {task!r} failed: {result.stderr}")
return result.stdout.strip()
async def main():
tasks = ["preprocess", "train", "evaluate", "export"]
results = await asyncio.gather(
*[run_task(t) for t in tasks],
return_exceptions=True,
)
for task, result in zip(tasks, results):
if isinstance(result, Exception):
print(f"{task}: FAILED — {result}")
else:
print(f"{task}: {result}")
asyncio.run(main())
Cloud vs local parallelism
Cloud sandboxes are independent of host hardware. Parallelism is limited only by your cua.ai plan quota and the region parameter. Use asyncio.Semaphore to stay within quota limits.import os, asyncio
from cua import Sandbox, Image
MAX_CONCURRENT = 20 # match your cua.ai plan quota
async def cloud_task(sem, job_id):
async with sem:
async with Sandbox.ephemeral(
Image.linux(),
api_key=os.environ["CUA_API_KEY"],
region="us-east-1",
) as sb:
result = await sb.shell.run(f"echo job-{job_id}")
return result.stdout.strip()
async def main():
sem = asyncio.Semaphore(MAX_CONCURRENT)
results = await asyncio.gather(*[cloud_task(sem, i) for i in range(50)])
print(results)
asyncio.run(main())
Local sandboxes compete for host CPU and memory. The recommended limit depends on image kind:| Image kind | Suggested MAX_CONCURRENT |
|---|
| Linux container | 8–16 (depends on RAM) |
| Linux VM (QEMU) | 2–4 |
| macOS VM (Lume) | 1–2 |
| Windows VM | 1–2 |
import asyncio
from cua import Sandbox, Image
MAX_CONCURRENT = 4 # for local QEMU VMs
async def local_task(sem, job_id):
async with sem:
async with Sandbox.ephemeral(Image.linux(), local=True) as sb:
result = await sb.shell.run(f"echo job-{job_id}")
return result.stdout.strip()
async def main():
sem = asyncio.Semaphore(MAX_CONCURRENT)
results = await asyncio.gather(*[local_task(sem, i) for i in range(16)])
print(results)
asyncio.run(main())
Practical example: N tasks across N sandboxes
The following end-to-end example runs a web-scraping job across eight sandboxes in parallel, collects results, and reports which tasks succeeded.
import asyncio
from cua import Sandbox, Image
URLS = [
"https://example.com",
"https://example.org",
"https://iana.org",
"https://httpbin.org",
"https://httpstat.us/200",
"https://google.com",
"https://github.com",
"https://python.org",
]
async def fetch_title(url: str) -> dict:
img = Image.linux().apt_install("curl")
async with Sandbox.ephemeral(img, local=True) as sb:
result = await sb.shell.run(
f"curl -sL --max-time 10 '{url}' | grep -oP '(?<=<title>)[^<]+'"
)
return {
"url": url,
"title": result.stdout.strip(),
"ok": result.success,
}
async def main():
sem = asyncio.Semaphore(4)
async def bounded(url):
async with sem:
return await fetch_title(url)
results = await asyncio.gather(
*[bounded(url) for url in URLS],
return_exceptions=True,
)
for r in results:
if isinstance(r, Exception):
print(f"ERROR: {r}")
else:
status = "✅" if r["ok"] else "❌"
print(f"{status} {r['url']}: {r['title'] or '(no title)'}")
asyncio.run(main())
