Pool

Pool keeps warm cloud capacity for a SandboxSpec. Sandbox.create uses shared capacity; a named pool is for dedicated capacity.

Most programs never touch pools: a cloud Sandbox.create claims from a pool the SDK manages per image and shape (cua-auto-*), tuned with CloudOptions. A named Pool is dedicated capacity you size and own; Sandbox.create(cloud=CloudOptions(pool=...)) claims from it, and fields that differ from its template raise PoolSpecMismatch.

Pools and templates#

Pool#

class Pool
 
Pool(
    resource: Any,
    *,
    owned_template: Any = None,
    agent_type: str | None = None,
)

A Fleet warm pool that can provide durable Sandbox claims.

Pool.reconcile#

@classmethod
async def reconcile(request: CreatePoolRequest) -> 'Pool'

Pool.get#

@classmethod
async def get(name: str) -> 'Pool'

Pool.apply#

@classmethod
async def apply(*args: Any, **kwargs: Any) -> 'Pool'

Reconcile a Fleet pool: Pool.apply(name, spec, options).

spec (cua_sandbox.SandboxSpec) is what the pool's sandboxes run and options (cua_sandbox.PoolOptions) how the pool keeps capacity for them. The pool, its namespace and its template share name, which is globally unique on Fleet. This is the SDK's one pool writer (Fleet.apply): the image is pinned to the variant the runtime runs, a new pool is rolled back if its template fails, and a runtime the image cannot run on raises cua.CuaError.InvalidArgument before anything is created.

The old form Pool.apply(image, *, name=..., replicas=..., cpu=..., memory_mb=..., services=..., autoscaling=..., ttl_seconds_after_created=..., runtime=...) still works and is converted to the new one, with a DeprecationWarning.

Example

import asyncio
 
from cua_sandbox import CloudOptions, Image, Pool, PoolOptions, Sandbox, SandboxSpec
 
async def main():
    pool = await Pool.apply("my-team-linux", SandboxSpec(image=Image.linux()), PoolOptions(replicas=2))
    sb = await Sandbox.create(cloud=CloudOptions(pool=pool))  # or CloudOptions(pool="my-team-linux")
    print(sb.id)
    await sb.destroy()
    await pool.delete()
 
asyncio.run(main())

Pool.export#

@staticmethod
async def export(name: str, *, terraform: bool = False) -> 'PoolExport | str'

Read pool name back as the shared model (a PoolExport whose spec / options are the native records), or with terraform=True the equivalent Terraform fleets_pool block (attributes the provider lacks yet are commented).

Pool.check#

@staticmethod
async def check(name: str, spec: SandboxSpec) -> None

Raise cua_sandbox.PoolSpecMismatch (with a readable diff) when the set fields of spec differ from pool name's template.

Pool.apply_template#

@staticmethod
async def apply_template(name: str, spec: SandboxSpec) -> None

Lay the set fields of spec over pool name's template and write it (the pool's capacity is kept; a no-op when nothing differs).

Pool.delete#

async def delete() -> None

Delete this Fleet pool.

Pool.create_claim#

async def create_claim(
    *,
    spec: ClaimSpec | None = None,
    name: str | None = None,
    ttl_seconds_after_created: int | None = None,
) -> _ClaimHandle

Pool.claim#

def claim(
    *,
    spec: ClaimSpec | None = None,
    name: str | None = None,
    service: str = ENV_SERVICE,
    time_to_start: float | None = None,
    ttl_seconds_after_created: int | None = None,
    agent_type: str | None = None,
    claim_token: str | None = None,
) -> _ClaimResult[Sandbox]

Claim a sandbox. agent_type="osworld" selects the OSWorld transport for pools fetched with Pool.get (pools from Pool.apply remember it).

claim_token (see generate_claim_token) is delivered into the sandbox at /run/cua/env-token through the claim's Secret; the pool's spec must set claim_secrets=True. The claim waits (at most 90 s after it binds) until the sandbox has it, else it is released and cua_sandbox.ClaimSecretsNotDelivered is raised.

Template#

class Template
 
Template(resource: Any)

A reconciled Fleet sandbox template.

Template.reconcile#

@classmethod
async def reconcile(request: CreateTemplateRequest) -> 'Template'

Managed pools (cua_sandbox.pools)#

pools#

Managed Fleet pools: list them and garbage-collect them.

Sandbox.ephemeral(image, local=False) claims sandboxes from this account's managed pools (cua-auto-*), one per image spec, created on first use and scaled to zero when idle. Idle pools are deleted automatically after CUA_FLEET_POOL_IDLE_GC; these calls inspect them or collect now.

from cua_sandbox import pools
 
for pool in await pools.list_pools():
    print(pool.name, pool.claims, pool.last_used)
report = await pools.gc(idle_after=1800)

list_pools#

async def list_pools(cfg: Optional[AutoPoolConfig] = None) -> list[ManagedPoolInfo]

This account's managed pools (cua-auto-* and legacy cua-eph-*).

list_claims#

async def list_claims(cfg: Optional[AutoPoolConfig] = None) -> list[ClaimInfo]

Claims in every pool this account can see (managed and explicit).

gc#

async def gc(
    idle_after: Optional[Duration] = None,
    cfg: Optional[AutoPoolConfig] = None,
) -> GcReport

Delete idle managed pools and stuck managed claims now.

gc_pools#

async def gc_pools(names: list[str], idle_after: Duration = 0) -> GcReport

Delete the named managed pools once they have no claims and have been idle for idle_after (default: now). Pools with live claims are kept. Tests use this to remove exactly the pools they created.

is_managed_pool_name#

def is_managed_pool_name(name: str, prefix: str = MANAGED_PREFIX) -> bool

ManagedPoolInfo#

class ManagedPoolInfo
 
ManagedPoolInfo(
    name: str,
    spec_hash: Optional[str],
    managed: bool,
    replicas: Optional[int],
    ready_replicas: Optional[int],
    claims: int,
    last_used: Optional[datetime],
    created_at: Optional[str],
)

ClaimInfo#

class ClaimInfo
 
ClaimInfo(
    name: str,
    pool: str,
    phase: Optional[str],
    managed: bool,
    created_at: Optional[str],
)

GcReport#

class GcReport
 
GcReport(
    pools_deleted: list[str] = list(),
    claims_deleted: list[str] = list(),
    namespaces_deleted: list[str] = list(),
    errors: list[str] = list(),
)

Specs#

SandboxSpec#

class SandboxSpec
 
SandboxSpec(
    image: Any = None,
    command: Optional[Sequence[str]] = None,
    args: Optional[Sequence[str]] = None,
    env: dict[str, str] = dict(),
    services: dict[str, int] = dict(),
    wait_for: Any = None,
    cpu: Optional[int] = None,
    memory_mb: Optional[int] = None,
    memory: Union[str, int, None] = None,
    efi: bool = False,
    sidecars: Sequence[Any] = (),
    registry_secret: Any = None,
    registry_secret_name: Optional[str] = None,
    process_mode: Optional[str] = None,
    claim_secrets: bool = False,
)

What a sandbox runs. Unset fields keep Fleet's defaults (and are not compared against a named pool's template).

• image: an cua_sandbox.Image (Image.from_registry(ref, secret=...) carries its registry secret) or a registry reference.
• command / args: argv replacing the image ENTRYPOINT / CMD.
• env: plain environment variables (not secrets).
• services: named guest ports, {"mcp": 8765}.
• wait_for: tcp("mcp") / http("mcp", "/health"): a replica binds a claim only once it passes.
• cpu / memory_mb (or memory="4GB").
• sidecars: cua_sandbox.Container (or image strings).
• registry_secret: a cua_sandbox.RegistrySecret, stored as the pool's cua-registry-* pull Secret; registry_secret_name references an existing one.
• process_mode: "Legacy" or "Run".
• claim_secrets: claims may carry a per-claim env token.

SandboxSpec.reference#

def reference() -> str

The image's registry reference ("" when unset).

SandboxSpec.native#

def native() -> Any

The cua.SandboxSpec record.

PoolOptions#

class PoolOptions
 
PoolOptions(
    runtime: Optional[str] = None,
    replicas: Optional[int] = None,
    warm: Optional[bool] = None,
    min_pool_size: Optional[int] = None,
    max_pool_size: Optional[int] = None,
    idle_ttl: Seconds = None,
    ttl_policy: Optional[str] = None,
    pool_ttl: Seconds = None,
    claim_ttl: Seconds = None,
)

How a pool keeps capacity for a SandboxSpec. Unset fields keep Fleet's defaults.

• runtime: "gvisor" or "kubevirt" (unset: from the image).
• replicas: size of a pool without autoscaling (default 1).
• warm: keep one sandbox ready (minPoolSize: 1).
• min_pool_size / max_pool_size: autoscaling floor / ceiling.
• idle_ttl: delete the pool after this long without claims.
• ttl_policy: "Retain" or "Cascade" (what TTL expiry deletes).
• pool_ttl: pool creation-age TTL.
• claim_ttl: default TTL of claims made on the pool.

Durations are seconds or a timedelta.

PoolOptions.native#

def native() -> Any

The cua.PoolOptions record.

PoolExport#

class PoolExport
 
PoolExport(name: str, runtime: str, spec: Any, options: Any, terraform: str)

A pool read back as the shared model (Pool.export).

generate_claim_token#

def generate_claim_token() -> str

A fresh per-claim env token (64 hex characters) for pool.claim(claim_token=...).

Errors#

PoolSpecMismatch#

class PoolSpecMismatch(InvalidArgument)

A named cloud pool's template differs from the sandbox fields given with it. The message holds a readable diff; pass CloudOptions(pool=..., apply=True) to update the pool's template instead, or omit the fields to use the pool as it is.

ClaimSecretsNotDelivered#

class ClaimSecretsNotDelivered(TimeoutError)

A claim bound, but its per-claim secrets (the env token) never reached the sandbox within the bounded wait (90 s); the claim was released.

PoolAccessDeniedError#

class PoolAccessDeniedError(PermissionError)

Fleet refused a pool or template operation for this credential.

Re-exported names#

cua_sandbox also re-exports these names from other packages; see their own documentation.