---
name: upstash-box-py
description: Work with the upstash-box Python SDK for sandboxed cloud containers with AI agents, shell, filesystem, git, cron schedules, snapshots, and a headless browser. Use when building with Upstash Box in Python, creating a sandbox or isolated environment to run untrusted or agent-generated code, running AI coding agents in containers, giving an agent a cloud dev environment with a shell and repository, browser automation from a box, scheduling recurring jobs inside a box, saving and restoring snapshots, or orchestrating parallel boxes.
license: MIT
metadata:
  author: Upstash
  homepage: https://upstash.com
---

# upstash-box Python SDK

Sandboxed cloud containers with built-in AI agents, shell, filesystem, git, cron schedules, and an optional headless browser.

Mirrors the `@upstash/box` TypeScript SDK (`upstash-box-js` skill) with
snake_case names; the intentional differences are listed under Gotchas.

## Install & Setup

```bash
pip install upstash-box
```

Set `UPSTASH_BOX_API_KEY` env var or pass `api_key` to constructors.

The SDK ships both a synchronous `Box` (used in the examples below) and an
asynchronous `AsyncBox` (`box = await AsyncBox.create(...)`, `await box.agent.run(...)`).
The async surface is identical with `await` and `async for`.

Anonymous telemetry headers are sent with every request; opt out with the
`UPSTASH_DISABLE_TELEMETRY` env var.

## Box Lifecycle

```python
import os
from upstash_box import Box, Agent, ClaudeCode, BoxApiKey

# Create with agent + git + env vars
box = Box.create(
    name="my-box",
    runtime="node",  # "node" | "python" | "golang" | "ruby" | "rust" (+ "-alpine" variants)
    size="small",  # "small" (2 CPU/4GB) | "medium" (4/8) | "large" (8/16)
    labels=["beta", "x-team"],  # max 5, <=20 chars each
    keep_alive=True,  # don't idle-pause the box
    init_command="npm install && npm run dev",  # keep-alive boxes only
    browser=True,  # provision headless Chromium for box.browser
    agent={
        "harness": Agent.CLAUDE_CODE,  # Agent.CODEX | Agent.OPEN_CODE | Agent.CURSOR | Agent.CUSTOM
        "model": ClaudeCode.SONNET_4_5,  # or a plain string "anthropic/claude-sonnet-4-5"
        # api_key options:
        #   omit                    → server decides which key to use
        #   BoxApiKey.UPSTASH_KEY   → use Upstash-provided LLM key
        #   BoxApiKey.STORED_KEY    → use key previously stored via Upstash Console
        #   "sk-..."                → direct API key string
        "api_key": BoxApiKey.UPSTASH_KEY,
    },
    git={  # all fields optional
        "token": os.environ["GITHUB_TOKEN"],  # or link your GitHub account via Upstash Console
        "user_name": "Bot",
        "user_email": "bot@example.com",
    },
    env={"DATABASE_URL": "..."},
    skills=["upstash/qstash-js/qstash-js"],  # owner/repo/skill-name
    timeout=600_000,  # request timeout in ms
    debug=False,
)

# Reconnect, list, delete, pause/resume
# Box.get / Box.get_by_name take api_key, base_url, git_token, timeout, debug
same = Box.get(box.id, git_token="ghp_...")  # git_token, not git={...}, when reconnecting
by_name = Box.get_by_name("my-box")
all_boxes = Box.list()
beta = Box.list(label="beta")  # filter by label
box.pause()  # raises on keep-alive boxes — they are never idle-paused
box.resume()
box.delete()  # irreversible
status = box.get_status()["status"]

box.id, box.size, box.keep_alive, box.cwd, box.network_policy

# Init command (keep-alive boxes only — raises otherwise)
box.set_init_command("npm run dev")
script = box.get_init_command()
box.delete_init_command()

# Bulk delete (classmethods, by ID)
Box.delete_boxes(box_ids=["box_1", "box_2"])  # JS static `delete` is `delete_boxes` here
Box.delete_snapshots(snapshot_ids=["snap_1"])  # omit ids → delete all
```

### Account-level env vars

Injected into every box you create.

```python
Box.set_env("API_TOKEN", "secret")
env = Box.list_env()  # values are masked
Box.set_all_env({"A": "1", "B": "2"})  # full replace — unlisted keys are removed
Box.delete_env("API_TOKEN")
```

## Agent Runs

```python
from pydantic import BaseModel

# Structured output with a Pydantic model (or a raw JSON-schema dict)
class Finding(BaseModel):
    severity: str  # "high" | "medium" | "low"
    file: str
    issue: str

class Review(BaseModel):
    verdict: str  # "approved" | "changes_requested"
    findings: list[Finding]

run = box.agent.run(
    prompt="Review the code for security issues",
    response_schema=Review,
    timeout=120_000,
    max_retries=2,
    options={"max_turns": 20, "max_budget_usd": 1.0, "effort": "high"},  # harness-specific
    on_tool_use=lambda tool: print(tool["name"], tool["input"]),
    on_tool_result=lambda result: print(result["tool_call_id"], result["output"]),
)

run.status   # "running" | "completed" | "failed" | "cancelled" | "detached"
run.result   # typed from schema (a Review instance)
run.cost     # RunCost(input_tokens, output_tokens, cached_input_tokens, compute_ms, total_usd)

# Attach files to a prompt (max 10 files, 10 MB each)
box.agent.run(prompt="Describe this", files=["./screenshot.png"])
box.agent.run(
    prompt="Describe this",
    files=[{"data": b64, "media_type": "image/png", "filename": "shot.png"}],
)

# Streaming — chunks are typed dataclasses discriminated on `.type`
stream = box.agent.stream(prompt="Build a REST API")
for chunk in stream:
    if chunk.type == "text-delta":
        print(chunk.text, end="")
    elif chunk.type == "reasoning":
        print(chunk.text, end="")
    elif chunk.type == "tool-call":
        print(chunk.tool_name, chunk.input)
    elif chunk.type == "tool-result":
        print(chunk.output)
    elif chunk.type == "finish":
        print(chunk.usage.input_tokens, chunk.usage.cached_input_tokens, chunk.session_id)
    # also: StartChunk(run_id) | StatsChunk(cpu_ns, memory_peak_bytes) | UnknownChunk(event, data)
    # FinishChunk also carries .output (the final text)

# stream() takes the same prompt/files/options/timeout/on_tool_use/on_tool_result as run().
# It has no response_schema, max_retries, or webhook — use run() for those.

# Fire-and-forget with webhook
box.agent.run(
    prompt="Run tests",
    webhook={"url": "https://example.com/hook", "headers": {"Authorization": "Bearer ..."}},
)
```

### Agent options (per harness)

`options` is forwarded to the harness — the accepted keys depend on which one
the box runs. Keys are snake_case in Python; the SDK converts **top-level** keys
to each harness's backend casing (Claude Code / OpenCode → camelCase, Codex →
snake_case). Keys inside nested dicts are sent verbatim.

```python
# Agent.CLAUDE_CODE → ClaudeCodeAgentOptions
{
    "max_turns": 20,
    "max_budget_usd": 1.0,
    "effort": "high",  # "low" | "medium" | "high" | "max"
    # nested dicts are forwarded verbatim — keep `budgetTokens` camelCase here
    "thinking": {"type": "adaptive"},  # or {"type": "enabled", "budgetTokens": 8000} / {"type": "disabled"}
    "disallowed_tools": ["Bash"],
    "agents": {"reviewer": {...}},  # custom subagent definitions
    "prompt_suggestions": False,
    "fallback_model": "anthropic/claude-sonnet-4-5",
    "system_prompt": "You are a release engineer.",
}

# Agent.CODEX → CodexAgentOptions
{
    "model_reasoning_effort": "high",  # "none" | "minimal" | "low" | "medium" | "high" | "xhigh"
    "model_reasoning_summary": "concise",  # "auto" | "concise" | "detailed" | "none"
    "personality": "pragmatic",  # "friendly" | "pragmatic" | "none"
    "web_search": "live",  # or True / False
}

# Agent.OPEN_CODE → OpenCodeAgentOptions
{
    "reasoning_effort": "high",  # "low" | "medium" | "high"
    "text_verbosity": "low",  # "low" | "medium" | "high"
    "reasoning_summary": "auto",  # "auto" | "concise" | "detailed" | "none"
    "thinking": {"type": "enabled", "budgetTokens": 8000},  # Anthropic-backed models
}

# Agent.CURSOR → free-form dict
```

Unlike the JS generic `AgentOptions<TProvider>`, Python does not narrow
`options` by harness — the type is the union of all shapes plus a raw dict.

### Harness & model

`harness` is required. Model enums: `ClaudeCode`, `OpenAICodex`, `OpenCodeModel`,
`CursorModel`, `OpenRouterModel`, `VercelModel` — or any provider-prefixed string.

```python
from upstash_box import ClaudeCode, OpenAICodex, OpenCodeModel, CursorModel, OpenRouterModel, VercelModel

ClaudeCode.FABLE_5_1  # "anthropic/claude-fable-5-1"
ClaudeCode.OPUS_5  # "anthropic/claude-opus-5"
ClaudeCode.SONNET_5  # "anthropic/claude-sonnet-5"
OpenAICodex.GPT_6_ASTRA  # "openai/gpt-6-astra"
OpenAICodex.GPT_5_6  # "openai/gpt-5.6"
OpenCodeModel.CLAUDE_OPUS_5  # "opencode/claude-opus-5"
CursorModel.COMPOSER_2_5  # "cursor/composer-2.5"
OpenRouterModel.CLAUDE_OPUS_5  # "openrouter/anthropic/claude-opus-5"
VercelModel.GPT_5_5  # "vercel/openai/gpt-5.5"

# Read / change the box's harness + model at runtime
box.model_config  # {"harness": ..., "model": ...}
box.configure_model("anthropic/claude-opus-4-8")

# Which harness a bare model string implies (prefix-based)
from upstash_box import infer_default_provider

infer_default_provider("openai/gpt-5.6")  # Agent.CODEX
infer_default_provider("cursor/default")  # Agent.CURSOR
```

### Custom harness

Run your own agent process inside the box instead of a managed harness.

```python
import asyncio
from upstash_box import Agent, Box, CustomHarnessDone, run_custom_harness

box = Box.create(
    agent={
        "harness": Agent.CUSTOM,
        "model": "my-agent",  # label forwarded to the process
        # command: name on PATH, or an absolute path under /workspace/home or /home/boxuser
        "custom_harness": {
            "command": "python",
            "args": ["/workspace/home/agent.py"],
            "protocol": "box-sse-v1",  # default
        },
    },
)
box.configure_custom_harness({"command": "python", "args": ["/workspace/home/agent2.py"]})

# Inside the box, agent.py emits box-sse-v1 events. The backend appends
# `-p <prompt> --model <model> --stream` (+ `--session <id>` when resuming).
# ctx: CustomHarnessContext(prompt, model, stream, args, session_id)
async def handler(ctx, emit):
    emit.text("working...")
    emit.reasoning("thinking out loud")  # -> `thinking` event
    emit.tool({"tool_call_id": "1", "name": "Bash", "input": {"command": "ls"}})
    emit.tool_result({"tool_call_id": "1", "output": "file.txt"})
    emit.emit("custom-event", {"any": "payload"})  # raw escape hatch
    # emit.error("boom") to fail the run
    return CustomHarnessDone(
        output="done",
        input_tokens=10,
        output_tokens=5,
        cached_input_tokens=0,
        total_cost_usd=0.01,
        session_id=ctx.session_id,
    )  # returning a plain string is shorthand for CustomHarnessDone(output=...)

asyncio.run(run_custom_harness(handler))  # run_custom_harness is async; handler may be sync or async
```

## Run Fields

Every `run` (agent, command, or code) returns a `Run`:

```python
run = box.exec.command("npm test")
run.id         # run ID
run.status     # "completed" | "failed" | ...
run.result     # stdout on success, stderr on failure (or typed result with response_schema)
run.stdout     # raw stdout (command/code runs)
run.stderr     # raw stderr (command/code runs)
run.exit_code  # int | None (None for agent runs)
run.cost       # RunCost(input_tokens, output_tokens, cached_input_tokens, compute_ms, total_usd)

run.cancel()          # cancel a running run
logs = run.logs()     # [RunLog(timestamp, level, message)]

# Box-level history
entries = box.logs(limit=100, offset=0)  # [LogEntry(timestamp, level, source, message)]
runs = box.list_runs()         # backend run records, newest first
```

## Shell Execution

```python
# Run commands
run = box.exec.command("echo hello && ls -la")

# Run code snippets — lang: "js" | "ts" | "python"
run2 = box.exec.code(code="print(1 + 1)", lang="python", timeout=10_000)

# Streaming shell / code
stream = box.exec.stream("npm run build")
stream2 = box.exec.stream_code(code="print('hi')", lang="python")
for chunk in stream:
    # chunk: ExecOutputChunk(type="output", data) | ExecExitChunk(type="exit", exit_code, cpu_ns)
    ...
```

### Live sessions

`exec.session()` opens a WebSocket to a process that is *still running* — stdin,
streamed stdout/stderr, PTY resize, and signals. It is the one feature not carried
by `httpx`; the `websockets` dependency is imported lazily, only when a session
opens. Available on `Box`, `AsyncBox`, and both ephemeral clients.

```python
session = box.exec.session(
    cmd="sort",               # run via `bash -lc`; argv=["sort"] runs the program with
                              # no shell and takes precedence over cmd
    cwd="/workspace/home",    # defaults to the box's tracked cwd
    env=["LOG_LEVEL=debug"],  # KEY=VALUE entries overlaid on the box environment
    on_stdout=lambda b: print(b.decode(), end=""),  # bytes
    on_stderr=lambda b: print(b.decode(), end=""),  # separate stream unless tty
)

session.pid       # in-box PID, always non-zero
session.exec_id   # server-side exec id
session.write("banana\napple\n")
session.end_stdin()             # EOF — a command that reads to EOF now exits by itself
exit_code = session.wait()      # -1 if torn down while still running; wait(timeout=5) raises TimeoutError
session.close()                 # hang up; also kills the process

# Interactive programs / TUIs — tty allocates a real PTY, merging stderr into stdout
with box.exec.session(argv=["bash", "-i"], tty=True, rows=40, cols=120) as shell:
    shell.resize(50, 160)
    shell.kill("INT")      # allowlist: TERM KILL INT HUP TSTP QUIT USR1 USR2 (default TERM)
    shell.terminate(5000)  # server-side SIGTERM, then SIGKILL after the grace (first call wins)
```

The session owns the process: `close()` (or leaving the `with` block), a dropped
connection, or the program exiting all kill the command, and a session cannot be
reattached. Use `wait()` to run something to completion.

On `AsyncBox` every handle method is a coroutine (`await session.write(...)`,
`await session.wait()`, `async with await box.exec.session(...) as s:`) and an
`async` callback is awaited; `wait()` there takes no timeout. In the sync client
the callbacks run on a background reader
thread — keep them short, and never call `wait()` from inside one, since the exit
frame it waits for arrives on the very thread it is blocking.

## Filesystem

```python
box.files.write(path="/workspace/home/app.py", content="print('hi')")
content = box.files.read("/workspace/home/app.py")
entries = box.files.list("/workspace/home")  # [FileEntry(name, path, size, is_dir, mod_time)]

# Binary files — use encoding="base64" for read and write
box.files.write(path="/workspace/home/image.png", content=base64_string, encoding="base64")
b64 = box.files.read("/workspace/home/image.png", encoding="base64")

# Bounded byte-range read — the *presence* of `length` selects the range, so
# length=0 reads zero bytes rather than the whole file. Server caps it at 8 MiB.
head = box.files.read("/workspace/home/big.log", length=64 * 1024)
chunk = box.files.read("/workspace/home/big.log", offset=1024, length=512)

# Metadata — defaults to lstat, so a symlink reports type "symlink"
stat = box.files.stat("/workspace/home/app.py")
# stat: FileStat(type="file" | "directory" | "symlink" | "other", size, mod_time, inode, version)
# `version` is an opaque freshness token (inode + mtime + size) for optimistic-concurrency
# guards — compare it for equality, never parse it.
target = box.files.stat("/workspace/home/link", follow=True)  # dereference

# Directories, moves, deletes
box.files.mkdir("build/cache", parents=True)  # parents mirrors `mkdir -p`
box.files.rename("draft.md", "docs/final.md")  # positional (from_path, to_path)
box.files.remove("build/cache", recursive=True)  # recursive required for a directory

# Upload local files
box.files.upload([{"path": "./local/file.txt", "destination": "/workspace/home/file.txt"}])

# Download — `folder` is a path INSIDE the box; files land in ./<basename>
box.files.download(folder="src")  # → ./src
box.files.download()  # whole cwd → ./workspace
```

## cd / Working Directory

The SDK tracks `cwd` client-side. All operations (exec, files, git, agent) run relative to it.

```python
box.cwd  # current working directory (starts at /workspace/home)
box.cd("my-repo")                  # relative to current cwd
box.cd("/workspace/home/other")    # absolute path
```

## Git

Clones land inside the box's isolated container, never on the caller's machine. Cloned
code is data until something runs it — treat an untrusted repo as untrusted input, and
pair it with a restrictive `network_policy` (see below) before running its build or tests.

Every git call except `clone` runs in the box's current directory, so `cd` into the
clone first. At the workspace root there is no repository, and `status` comes back
empty, which reads as a clean tree.

```python
box.git.clone(repo="github.com/org/repo", branch="main")
box.git.clone(repo="github.com/org/repo", depth=1)  # shallow clone
box.git.clone(repo="github.com/org/repo", folder="my-app")  # destination
box.cd("repo")  # the clone lands in a directory named after the repo

status = box.git.status()
diff = box.git.diff()
result = box.git.commit(  # GitCommitResult(sha, message)
    message="fix: resolve bug",
    author_name="Jane Doe",  # optional per-commit override
    author_email="jane@example.com",
)
box.git.push(branch="feature/fix")

box.git.checkout(branch="release/v2")
pr = box.git.create_pr(title="Fix bug", body="...", base="main")
# pr: PullRequest(url, number, title, base)

# Update the box-wide git identity
cfg = box.git.update_config(user_name="Bot", user_email="bot@example.com")
# cfg: GitConfigResult(git_user_name, git_user_email)

# Arbitrary git commands — returns the output string only. Unlike the JS SDK, which
# returns { output, exit_code }, Python drops the status, so a failure (exit 128 when
# the cwd is not a repository) is indistinguishable from success — check the output.
output = box.git.exec(args=["log", "--oneline", "-5"])
```

## Schedules

Cron tasks on a box — shell commands or agent prompts. Available on `Box` and `EphemeralBox`. Cron is UTC.

```python
exec_schedule = box.schedule.exec(
    cron="* * * * *",
    command=["bash", "-c", "date >> /workspace/home/cron.log"],
    folder="/workspace/home",  # optional cwd override
    webhook_url="https://example.com/hook",
    webhook_headers={"Authorization": "Bearer ..."},
)

agent_schedule = box.schedule.agent(
    cron="0 9 * * *",
    prompt="Run the test suite and fix any failures",
    folder="/workspace/home/repo",  # optional cwd override
    model="anthropic/claude-sonnet-5",  # optional override
    options={"max_budget_usd": 1.0, "effort": "high"},
    timeout=300_000,
    webhook_url="https://example.com/hook",
    webhook_headers={"Authorization": "Bearer ..."},
)

schedules = box.schedule.list()
one = box.schedule.get(agent_schedule.id)

# Partial update — omitted args keep their value, "" / [] / {} clear a field,
# options=None clears agent options. The schedule's type cannot change.
# Updatable: cron, command, prompt, folder, model, options, timeout,
#            webhook_url, webhook_headers
box.schedule.update(agent_schedule.id, cron="0 18 * * *", webhook_url="")

box.schedule.pause(agent_schedule.id)
box.schedule.resume(agent_schedule.id)
box.schedule.delete(agent_schedule.id)
```

## Snapshots

```python
# Snapshot — checkpoint workspace state
snap = box.snapshot(name="after-setup")
# snap: Snapshot(id, name, box_id, size_bytes, status, created_at)

# from_snapshot takes the same BoxConfig kwargs as create (shared request body):
# name, labels, size, keep_alive, init_command, runtime, browser, agent, git, env,
# attach_headers, network_policy, skills, mcp_servers. Note the JS SDK's
# Box.fromSnapshot() drops browser / skills / mcpServers — Python forwards them