tangle-sandbox

Python SDK for Tangle Sandbox. Create cloud sandboxes, run commands and coding agents in them, and delete them when done.

Installation

Install tangle-sandbox from PyPI:

pip install tangle-sandbox  # extras: [gateway] session gateway, [tangle] on-chain client, [all]

Quick Start

import os

from tangle_sandbox import Sandbox, SandboxStatus

client = Sandbox(
    api_key=os.environ["TANGLE_API_KEY"],
    base_url="https://sandbox.tangle.tools",
)

box = client.create(name="my-project")
box.wait_for(SandboxStatus.RUNNING)
try:
    # Execute commands
    result = box.exec("python3 -c 'print(sum(range(1_000_000)))'")
    print(result.stdout)

    # Prompt the sandbox agent
    result = box.prompt("Reply with exactly the word OK")
    print(result.response if result.success else result.error)
finally:
    # Clean up
    box.delete()
    client.close()

Async Usage

from tangle_sandbox import AsyncSandbox, SandboxStatus

async with AsyncSandbox(api_key="...", base_url="...") as client:
    box = await client.create(name="my-project")
    await box.wait_for(SandboxStatus.RUNNING)

    result = await box.exec("npm test")
    print(result.stdout)

    await box.delete()

Batch Execution

Each task runs once against each backend. Backend IDs distinguish configurations even when they use the same runtime type.

from tangle_sandbox import (
    BatchBackend,
    BatchBackendModel,
    BatchRunRequest,
    BatchTask,
    Sandbox,
)

request = BatchRunRequest(
    tasks=[
        BatchTask(id="review", message="Review the repository and report the highest-risk bug"),
        BatchTask(id="fix", message="Fix the failing tests"),
    ],
    backends=[
        BatchBackend(id="builder", type="opencode"),
        BatchBackend(
            id="reviewer",
            type="claude-code",
            model=BatchBackendModel(
                provider="anthropic",
                model="claude-sonnet-5-5",
            ),
        ),
    ],
    persistent=True,
    reuse_key="repository-review-workspace",
)

with Sandbox(api_key="...", base_url="...") as client:
    result = client.run_batch(request, idempotency_key="review-and-fix-run-001")
    print(result.total_success, result.total_executions, result.success_rate)
    print(result.cleanup.mode)

Use stream_batch() for progress events. It yields every valid server event, including batch.failed. After cleanup events finish, both methods raise BatchFailedError for batch.failed. The error is a SandboxError with code OPERATION_FAILED whose cleanup field reports confirmed releases, pending releases, and completeness. run_batch() also rejects incomplete or inconsistent results. Malformed or inconsistent stream data raises NetworkError. For immediate cleanup, result.cleanup lists confirmed and pending sidecar IDs and reports whether deletion is complete. Persistent and grace-period runs return their retention details instead.

for event in client.stream_batch(request):
    if event.type in {"task.completed", "task.failed"}:
        print(event.data["taskId"], event.data["backendId"])

The async client exposes the same API through async for client.stream_batch(...) and await client.run_batch(...). BatchRunRequest requires tasks and explicitly named backends, and also supports execution class, persistence, lifecycle timeouts, cleanup grace time, and Git workspace setup. Billing identity, scaling, and capacity are assigned by the service. persistent=True requires reuse_key, which selects the same caller-owned workspace across separate batch requests. lifecycle and git also require persistent=True; grace_ms applies only to non-persistent batches and is omitted by default. Batch Git repositories must use a public DNS host over HTTPS with no query, fragment, or custom port. Supply credentials through git.auth rather than embedding them in the URL.

Closing a sync stream or cancelling its async consumer disconnects the HTTP stream immediately. Without idempotency_key, that disconnect cancels the server run. With idempotency_key, the server continues the exact request so another call with the same key and body can join or replay it. Use a key only for retries of one unchanged request body. reuse_key controls persistent workspace selection, while idempotency_key controls retry identity for one batch run.

Sandbox Fleets

Use fleets when one workload should run across several sandboxes. A researcher can create a coordinator plus workers, run the same command across the workers, compare outputs, then delete the whole fleet.

from tangle_sandbox import (
    CreateSandboxFleetWithCoordinatorOptions,
    Sandbox,
    SandboxFleetMachineSpec,
)

with Sandbox(api_key="...", base_url="...") as client:
    fleet = client.fleets.create_with_coordinator(
        CreateSandboxFleetWithCoordinatorOptions(
            workers=[
                SandboxFleetMachineSpec(machine_id="worker-1"),
                SandboxFleetMachineSpec(machine_id="worker-2"),
            ],
        )
    )

    try:
        run = fleet.dispatch_exec_with_insights("python run_experiment.py")
        for result in run["results"]:
            print(result["machineId"], result["ok"])
        print(run["intelligence"]["recommendedActions"])
    finally:
        fleet.delete()

Trace Intelligence

Sandboxes and fleet runs expose the full raw trace by default. The compact intelligence envelope is opt-in unless you call the explicit intelligence helper. The raw trace is the evidence: lifecycle events, usage, timing metrics, runtime state, and, for fleets, machine events, dispatch results, critical path, and workspace state. The intelligence envelope is the current platform analysis: reliability, parallelism efficiency for fleets, failure classes, resource attribution, bottlenecks, and recommended actions.

box_bundle = box.trace()          # {"trace": ...}
box_insights = box.intelligence() # intelligence only
box_bundle_with_insights = box.trace(include_intelligence=True)

run = fleet.dispatch_exec_with_insights("pytest -q")
print(run["trace"]["events"])
print(run["intelligence"]["signals"])
print(run["intelligence"]["recommendedActions"])

bundle = fleet.trace()          # {"trace": ...}
insights = fleet.intelligence() # intelligence only
bundle_with_insights = fleet.trace(include_intelligence=True)

Forward the native bundle to your own warehouse or convert it to OTEL JSON for OpenTelemetry-style collectors.

The built-in intelligence is deterministic platform analysis. It reports billing.billable: false and billing.costUsd: 0; customers are not charged for generating these envelopes. Products opt into generated intelligence per trace request with include_intelligence=True or by calling intelligence().

box.export_trace(
    "https://collector.example.com/traces",
    headers={"Authorization": f"Bearer {OBSERVABILITY_TOKEN}"},
    format="otel-json",
    service_name="research-agent",
)

fleet.export_trace(
    "https://collector.example.com/traces",
    headers={"Authorization": f"Bearer {OBSERVABILITY_TOKEN}"},
    format="otel-json",
    service_name="research-agent",
)

For Braintrust, Lemma, Raindrop, Langfuse, Datadog, or internal tools, keep the vendor credentials on your side. Fetch box.trace() or fleet.trace() and transform the raw bundle through that platform's SDK/API, or point export_trace() at your own ingest endpoint.

Timeouts and retries

_http.TIMEOUTS owns request defaults in seconds. Every sync and async request, including .direct(), uses the client's timeout_ms unless the call supplies a longer operation timeout. Creation shares that client budget across attempts. Exec waits its execution limit plus margin; fleet dispatch waits the larger of its requested duration and fleet_dispatch, plus margin. Snapshot restore and image builds retain their bounded operation waits in _instance_ops.py and image.py; collaboration, trace export and Tangle operator requests retain their own bounded defaults.

Streams use TIMEOUTS["stream"] as the maximum read gap; PromptOptions.timeout_ms replaces it for a prompt. HTTPX deadlines bound network phases and read gaps, rather than a whole streamed operation's lifetime. A request timeout does not cancel work already accepted by the server. Retries require a catalog-retryable error and an idempotent method, or a connection failure that proves no request was delivered. Request attempts and retry pacing share one budget; a Retry-After beyond it returns the original refusal. Creation reuses its idempotency key through its separate recovery budget.

API Reference

Client

from tangle_sandbox import Sandbox, CreateSandboxOptions, GitConfig

client = Sandbox(
    api_key="sk_sandbox_...",
    base_url="https://your-sandbox-api.example.com",
    timeout_ms=30000,
)

client.create(options)

from tangle_sandbox import CreateSandboxOptions, GitConfig

box = client.create(CreateSandboxOptions(
    name="my-project",
    runtime="firecracker",
    git=GitConfig(url="https://github.com/user/repo.git", ref="main"),
    ssh_enabled=True,
    env={"NODE_ENV": "development"},
))

runtime and environment mean what the TypeScript SDK README's "Runtime and image" table says; client.environments.list() shows each environment's runtimes.

client.list(options)

from tangle_sandbox import ListSandboxOptions, SandboxStatus

boxes = client.list(ListSandboxOptions(status=SandboxStatus.RUNNING, limit=10))

client.get(sandbox_id)

box = client.get("sandbox_abc123")
if box:
    print(box.status)

Sandbox Instance

box.exec(command, options)

from tangle_sandbox import ExecOptions

result = box.exec("npm install", ExecOptions(cwd="/workspace", timeout_ms=60000))
print(result.exit_code)
print(result.stdout)

box.prompt(message, options)

from tangle_sandbox import BackendConfig, PromptOptions

result = box.prompt("What files are in this project?", PromptOptions(
    session_id="session_123",
    backend=BackendConfig(
        type="claude-code",
        model={
            "provider": "anthropic",
            "model": "claude-sonnet-5-5",
        },
    ),
))
print(result.response)

box.stream_prompt(message, options)

for event in box.stream_prompt("Explain this codebase"):
    if event.type == "message.part.updated":
        part = event.data.get("part", {})
        if part.get("type") == "text" and event.data.get("delta"):
            print(event.data["delta"], end="")

File Operations

content = box.read("src/index.ts")
box.write("src/fix.ts", "export const fix = () => {}")

Code Search

from tangle_sandbox import SearchOptions

matches = box.search("TODO:", SearchOptions(glob="**/*.ts"))
for match in matches:
    print(f"{match.path}:{match.line}: {match.text}")

Git

status = box.git.status()
if status.is_dirty:
    box.git.add(["."])
    box.git.commit("Update files")
    box.git.push()

Snapshots

from tangle_sandbox import SnapshotOptions

snapshot = box.snapshot(SnapshotOptions(tags=["before-deploy"]))
print(snapshot.snapshot_id)

snapshots = box.list_snapshots()

SSH

creds = box.ssh()
print(f"ssh {creds.username}@{creds.host} -p {creds.port}")

Egress

from tangle_sandbox import EgressPolicy

# strict reaches these plus the sandbox's own model endpoint, nothing else.
policy, source = box.egress_update(EgressPolicy(
    mode="strict",
    allow_domains=["api.github.com", "pypi.org"],
))

Lifecycle

from tangle_sandbox import SandboxStatus, WaitForOptions

box.wait_for(SandboxStatus.RUNNING, WaitForOptions(timeout_ms=60000))
box.stop()
box.resume()
box.delete()

Image Builder

from tangle_sandbox import Image

image = (
    Image.create("ubuntu:22.04")
    .python("3.11")
    .add_packages(["numpy", "pandas", "scikit-learn"])
    .add_apt_packages(["ffmpeg"])
    .with_env({"PYTHONUNBUFFERED": "1"})
    .build(client=client)
)

box = client.create(environment=image.id)

Error Handling

Every SDK error is one SandboxError. code is the platform's error code; status, retryable and next_step come from the platform error catalog (tangle_sandbox/_error_catalog.py, generated from packages/runtime-contracts/src/error-catalog.ts). A code's own fields (timeout_ms, current_state, quota_type, failures) read as attributes.

from tangle_sandbox import SandboxError

try:
    result = box.exec("npm test")
except SandboxError as e:
    print(f"{e.code} (HTTP {e.status}): {e.message}")
    print(f"Next: {e.next_step}")
    if e.retryable:
        ...  # a later attempt can succeed

AuthError, NotFoundError, QuotaError, CapabilityError, PartialFailureError, ValidationError, StateError, TimeoutError, NetworkError, ServerError and BatchFailedError remain for one minor release as aliases of SandboxError, so existing except clauses keep working; catch SandboxError and read code instead.

Requirements

• Python 3.10+
• httpx >= 0.25.0

License

MIT