Metadata-Version: 2.5
Name: tangle-sandbox
Version: 0.1.0
Summary: Python SDK for the Tangle Sandbox platform
Project-URL: Homepage, https://tangle.tools
Project-URL: Repository, https://github.com/tangle-network/agent-container/tree/develop/products/sandbox/sdk-python
Author-email: Tangle Technologies <eng@tangle.tools>
License-Expression: MIT
Keywords: ai,containers,development,sandbox,tangle
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.25.0
Requires-Dist: pydantic<3,>=2.11
Provides-Extra: all
Requires-Dist: eth-account>=0.10; extra == 'all'
Requires-Dist: web3>=6.0; extra == 'all'
Requires-Dist: websockets>=12.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: eth-account>=0.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest-httpx>=0.30; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Requires-Dist: web3>=6.0; extra == 'dev'
Requires-Dist: websockets>=12.0; extra == 'dev'
Provides-Extra: gateway
Requires-Dist: websockets>=12.0; extra == 'gateway'
Provides-Extra: tangle
Requires-Dist: eth-account>=0.10; extra == 'tangle'
Requires-Dist: web3>=6.0; extra == 'tangle'
Description-Content-Type: text/markdown

# tangle-sandbox

Python SDK for the Tangle Sandbox platform. Create isolated dev containers, run AI agents, and build automation workflows.

## Installation

Install [`tangle-sandbox` from PyPI](https://pypi.org/project/tangle-sandbox/):

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

## Quick Start

The production journey runs these calls against each cell every five minutes (`scripts/production-journey.mjs`, step `python`).

```python
import os

from tangle_sandbox import Sandbox

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

box = client.create(name="my-project")
box.wait_for("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

```python
from tangle_sandbox import AsyncSandbox

async with AsyncSandbox(api_key="...", base_url="...") as client:
    box = await client.create(name="my-project")
    await box.wait_for("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.

```python
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.

```python
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.

```python
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.

```python
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()`.

```python
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`](src/tangle_sandbox/_http.py) 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

```python
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)`

```python
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)`

```python
from tangle_sandbox import ListSandboxOptions, SandboxStatus

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

#### `client.get(sandbox_id)`

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

### Sandbox Instance

#### `box.exec(command, options)`

```python
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)`

```python
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)`

```python
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

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

#### Code Search

```python
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

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

#### Snapshots

```python
from tangle_sandbox import SnapshotOptions

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

snapshots = box.list_snapshots()
```

#### SSH

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

#### Egress

```python
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

```python
from tangle_sandbox import SandboxStatus, WaitForOptions

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

### Image Builder

```python
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.

```python
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
