Update to Upstash Redis

{
  "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.",
  "included_files": [],
  "skill_md_contents": "---\nname: upstash-box-py\ndescription: 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.\nlicense: MIT\nmetadata:\n  author: Upstash\n  homepage: https://upstash.com\n---\n\n# upstash-box Python SDK\n\nSandboxed cloud containers with built-in AI agents, shell, filesystem, git, cron schedules, and an optional headless browser.\n\nMirrors the `@upstash/box` TypeScript SDK (`upstash-box-js` skill) with\nsnake_case names; the intentional differences are listed under Gotchas.\n\n## Install & Setup\n\n```bash\npip install upstash-box\n```\n\nSet `UPSTASH_BOX_API_KEY` env var or pass `api_key` to constructors.\n\nThe SDK ships both a synchronous `Box` (used in the examples below) and an\nasynchronous `AsyncBox` (`box = await AsyncBox.create(...)`, `await box.agent.run(...)`).\nThe async surface is identical with `await` and `async for`.\n\nAnonymous telemetry headers are sent with every request; opt out with the\n`UPSTASH_DISABLE_TELEMETRY` env var.\n\n## Box Lifecycle\n\n```python\nimport os\nfrom upstash_box import Box, Agent, ClaudeCode, BoxApiKey\n\n# Create with agent + git + env vars\nbox = Box.create(\n    name=\"my-box\",\n    runtime=\"node\",  # \"node\" | \"python\" | \"golang\" | \"ruby\" | \"rust\" (+ \"-alpine\" variants)\n    size=\"small\",  # \"small\" (2 CPU/4GB) | \"medium\" (4/8) | \"large\" (8/16)\n    labels=[\"beta\", \"x-team\"],  # max 5, <=20 chars each\n    keep_alive=True,  # don't idle-pause the box\n    init_command=\"npm install && npm run dev\",  # keep-alive boxes only\n    browser=True,  # provision headless Chromium for box.browser\n    agent={\n        \"harness\": Agent.CLAUDE_CODE,  # Agent.CODEX | Agent.OPEN_CODE | Agent.CURSOR | Agent.CUSTOM\n        \"model\": ClaudeCode.SONNET_4_5,  # or a plain string \"anthropic/claude-sonnet-4-5\"\n        # api_key options:\n        #   omit                    → server decides which key to use\n        #   BoxApiKey.UPSTASH_KEY   → use Upstash-provided LLM key\n        #   BoxApiKey.STORED_KEY    → use key previously stored via Upstash Console\n        #   \"sk-...\"                → direct API key string\n        \"api_key\": BoxApiKey.UPSTASH_KEY,\n    },\n    git={  # all fields optional\n        \"token\": os.environ[\"GITHUB_TOKEN\"],  # or link your GitHub account via Upstash Console\n        \"user_name\": \"Bot\",\n        \"user_email\": \"bot@example.com\",\n    },\n    env={\"DATABASE_URL\": \"...\"},\n    skills=[\"upstash/qstash-js/qstash-js\"],  # owner/repo/skill-name\n    timeout=600_000,  # request timeout in ms\n    debug=False,\n)\n\n# Reconnect, list, delete, pause/resume\n# Box.get / Box.get_by_name take api_key, base_url, git_token, timeout, debug\nsame = Box.get(box.id, git_token=\"ghp_...\")  # git_token, not git={...}, when reconnecting\nby_name = Box.get_by_name(\"my-box\")\nall_boxes = Box.list()\nbeta = Box.list(label=\"beta\")  # filter by label\nbox.pause()  # raises on keep-alive boxes — they are never idle-paused\nbox.resume()\nbox.delete()  # irreversible\nstatus = box.get_status()[\"status\"]\n\nbox.id, box.size, box.keep_alive, box.cwd, box.network_policy\n\n# Init command (keep-alive boxes only — raises otherwise)\nbox.set_init_command(\"npm run dev\")\nscript = box.get_init_command()\nbox.delete_init_command()\n\n# Bulk delete (classmethods, by ID)\nBox.delete_boxes(box_ids=[\"box_1\", \"box_2\"])  # JS static `delete` is `delete_boxes` here\nBox.delete_snapshots(snapshot_ids=[\"snap_1\"])  # omit ids → delete all\n```\n\n### Account-level env vars\n\nInjected into every box you create.\n\n```python\nBox.set_env(\"API_TOKEN\", \"secret\")\nenv = Box.list_env()  # values are masked\nBox.set_all_env({\"A\": \"1\", \"B\": \"2\"})  # full replace — unlisted keys are removed\nBox.delete_env(\"API_TOKEN\")\n```\n\n## Agent Runs\n\n```python\nfrom pydantic import BaseModel\n\n# Structured output with a Pydantic model (or a raw JSON-schema dict)\nclass Finding(BaseModel):\n    severity: str  # \"high\" | \"medium\" | \"low\"\n    file: str\n    issue: str\n\nclass Review(BaseModel):\n    verdict: str  # \"approved\" | \"changes_requested\"\n    findings: list[Finding]\n\nrun = box.agent.run(\n    prompt=\"Review the code for security issues\",\n    response_schema=Review,\n    timeout=120_000,\n    max_retries=2,\n    options={\"max_turns\": 20, \"max_budget_usd\": 1.0, \"effort\": \"high\"},  # harness-specific\n    on_tool_use=lambda tool: print(tool[\"name\"], tool[\"input\"]),\n    on_tool_result=lambda result: print(result[\"tool_call_id\"], result[\"output\"]),\n)\n\nrun.status   # \"running\" | \"completed\" | \"failed\" | \"cancelled\" | \"detached\"\nrun.result   # typed from schema (a Review instance)\nrun.cost     # RunCost(input_tokens, output_tokens, cached_input_tokens, compute_ms, total_usd)\n\n# Attach files to a prompt (max 10 files, 10 MB each)\nbox.agent.run(prompt=\"Describe this\", files=[\"./screenshot.png\"])\nbox.agent.run(\n    prompt=\"Describe this\",\n    files=[{\"data\": b64, \"media_type\": \"image/png\", \"filename\": \"shot.png\"}],\n)\n\n# Streaming — chunks are typed dataclasses discriminated on `.type`\nstream = box.agent.stream(prompt=\"Build a REST API\")\nfor chunk in stream:\n    if chunk.type == \"text-delta\":\n        print(chunk.text, end=\"\")\n    elif chunk.type == \"reasoning\":\n        print(chunk.text, end=\"\")\n    elif chunk.type == \"tool-call\":\n        print(chunk.tool_name, chunk.input)\n    elif chunk.type == \"tool-result\":\n        print(chunk.output)\n    elif chunk.type == \"finish\":\n        print(chunk.usage.input_tokens, chunk.usage.cached_input_tokens, chunk.session_id)\n    # also: StartChunk(run_id) | StatsChunk(cpu_ns, memory_peak_bytes) | UnknownChunk(event, data)\n    # FinishChunk also carries .output (the final text)\n\n# stream() takes the same prompt/files/options/timeout/on_tool_use/on_tool_result as run().\n# It has no response_schema, max_retries, or webhook — use run() for those.\n\n# Fire-and-forget with webhook\nbox.agent.run(\n    prompt=\"Run tests\",\n    webhook={\"url\": \"https://example.com/hook\", \"headers\": {\"Authorization\": \"Bearer ...\"}},\n)\n```\n\n### Agent options (per harness)\n\n`options` is forwarded to the harness — the accepted keys depend on which one\nthe box runs. Keys are snake_case in Python; the SDK converts **top-level** keys\nto each harness's backend casing (Claude Code / OpenCode → camelCase, Codex →\nsnake_case). Keys inside nested dicts are sent verbatim.\n\n```python\n# Agent.CLAUDE_CODE → ClaudeCodeAgentOptions\n{\n    \"max_turns\": 20,\n    \"max_budget_usd\": 1.0,\n    \"effort\": \"high\",  # \"low\" | \"medium\" | \"high\" | \"max\"\n    # nested dicts are forwarded verbatim — keep `budgetTokens` camelCase here\n    \"thinking\": {\"type\": \"adaptive\"},  # or {\"type\": \"enabled\", \"budgetTokens\": 8000} / {\"type\": \"disabled\"}\n    \"disallowed_tools\": [\"Bash\"],\n    \"agents\": {\"reviewer\": {...}},  # custom subagent definitions\n    \"prompt_suggestions\": False,\n    \"fallback_model\": \"anthropic/claude-sonnet-4-5\",\n    \"system_prompt\": \"You are a release engineer.\",\n}\n\n# Agent.CODEX → CodexAgentOptions\n{\n    \"model_reasoning_effort\": \"high\",  # \"none\" | \"minimal\" | \"low\" | \"medium\" | \"high\" | \"xhigh\"\n    \"model_reasoning_summary\": \"concise\",  # \"auto\" | \"concise\" | \"detailed\" | \"none\"\n    \"personality\": \"pragmatic\",  # \"friendly\" | \"pragmatic\" | \"none\"\n    \"web_search\": \"live\",  # or True / False\n}\n\n# Agent.OPEN_CODE → OpenCodeAgentOptions\n{\n    \"reasoning_effort\": \"high\",  # \"low\" | \"medium\" | \"high\"\n    \"text_verbosity\": \"low\",  # \"low\" | \"medium\" | \"high\"\n    \"reasoning_summary\": \"auto\",  # \"auto\" | \"concise\" | \"detailed\" | \"none\"\n    \"thinking\": {\"type\": \"enabled\", \"budgetTokens\": 8000},  # Anthropic-backed models\n}\n\n# Agent.CURSOR → free-form dict\n```\n\nUnlike the JS generic `AgentOptions<TProvider>`, Python does not narrow\n`options` by harness — the type is the union of all shapes plus a raw dict.\n\n### Harness & model\n\n`harness` is required. Model enums: `ClaudeCode`, `OpenAICodex`, `OpenCodeModel`,\n`CursorModel`, `OpenRouterModel`, `VercelModel` — or any provider-prefixed string.\n\n```python\nfrom upstash_box import ClaudeCode, OpenAICodex, OpenCodeModel, CursorModel, OpenRouterModel, VercelModel\n\nClaudeCode.FABLE_5_1  # \"anthropic/claude-fable-5-1\"\nClaudeCode.OPUS_5  # \"anthropic/claude-opus-5\"\nClaudeCode.SONNET_5  # \"anthropic/claude-sonnet-5\"\nOpenAICodex.GPT_6_ASTRA  # \"openai/gpt-6-astra\"\nOpenAICodex.GPT_5_6  # \"openai/gpt-5.6\"\nOpenCodeModel.CLAUDE_OPUS_5  # \"opencode/claude-opus-5\"\nCursorModel.COMPOSER_2_5  # \"cursor/composer-2.5\"\nOpenRouterModel.CLAUDE_OPUS_5  # \"openrouter/anthropic/claude-opus-5\"\nVercelModel.GPT_5_5  # \"vercel/openai/gpt-5.5\"\n\n# Read / change the box's harness + model at runtime\nbox.model_config  # {\"harness\": ..., \"model\": ...}\nbox.configure_model(\"anthropic/claude-opus-4-8\")\n\n# Which harness a bare model string implies (prefix-based)\nfrom upstash_box import infer_default_provider\n\ninfer_default_provider(\"openai/gpt-5.6\")  # Agent.CODEX\ninfer_default_provider(\"cursor/default\")  # Agent.CURSOR\n```\n\n### Custom harness\n\nRun your own agent process inside the box instead of a managed harness.\n\n```python\nimport asyncio\nfrom upstash_box import Agent, Box, CustomHarnessDone, run_custom_harness\n\nbox = Box.create(\n    agent={\n        \"harness\": Agent.CUSTOM,\n        \"model\": \"my-agent\",  # label forwarded to the process\n        # command: name on PATH, or an absolute path under /workspace/home or /home/boxuser\n        \"custom_harness\": {\n            \"command\": \"python\",\n            \"args\": [\"/workspace/home/agent.py\"],\n            \"protocol\": \"box-sse-v1\",  # default\n        },\n    },\n)\nbox.configure_custom_harness({\"command\": \"python\", \"args\": [\"/workspace/home/agent2.py\"]})\n\n# Inside the box, agent.py emits box-sse-v1 events. The backend appends\n# `-p <prompt> --model <model> --stream` (+ `--session <id>` when resuming).\n# ctx: CustomHarnessContext(prompt, model, stream, args, session_id)\nasync def handler(ctx, emit):\n    emit.text(\"working...\")\n    emit.reasoning(\"thinking out loud\")  # -> `thinking` event\n    emit.tool({\"tool_call_id\": \"1\", \"name\": \"Bash\", \"input\": {\"command\": \"ls\"}})\n    emit.tool_result({\"tool_call_id\": \"1\", \"output\": \"file.txt\"})\n    emit.emit(\"custom-event\", {\"any\": \"payload\"})  # raw escape hatch\n    # emit.error(\"boom\") to fail the run\n    return CustomHarnessDone(\n        output=\"done\",\n        input_tokens=10,\n        output_tokens=5,\n        cached_input_tokens=0,\n        total_cost_usd=0.01,\n        session_id=ctx.session_id,\n    )  # returning a plain string is shorthand for CustomHarnessDone(output=...)\n\nasyncio.run(run_custom_harness(handler))  # run_custom_harness is async; handler may be sync or async\n```\n\n## Run Fields\n\nEvery `run` (agent, command, or code) returns a `Run`:\n\n```python\nrun = box.exec.command(\"npm test\")\nrun.id         # run ID\nrun.status     # \"completed\" | \"failed\" | ...\nrun.result     # stdout on success, stderr on failure (or typed result with response_schema)\nrun.stdout     # raw stdout (command/code runs)\nrun.stderr     # raw stderr (command/code runs)\nrun.exit_code  # int | None (None for agent runs)\nrun.cost       # RunCost(input_tokens, output_tokens, cached_input_tokens, compute_ms, total_usd)\n\nrun.cancel()          # cancel a running run\nlogs = run.logs()     # [RunLog(timestamp, level, message)]\n\n# Box-level history\nentries = box.logs(limit=100, offset=0)  # [LogEntry(timestamp, level, source, message)]\nruns = box.list_runs()         # backend run records, newest first\n```\n\n## Shell Execution\n\n```python\n# Run commands\nrun = box.exec.command(\"echo hello && ls -la\")\n\n# Run code snippets — lang: \"js\" | \"ts\" | \"python\"\nrun2 = box.exec.code(code=\"print(1 + 1)\", lang=\"python\", timeout=10_000)\n\n# Streaming shell / code\nstream = box.exec.stream(\"npm run build\")\nstream2 = box.exec.stream_code(code=\"print('hi')\", lang=\"python\")\nfor chunk in stream:\n    # chunk: ExecOutputChunk(type=\"output\", data) | ExecExitChunk(type=\"exit\", exit_code, cpu_ns)\n    ...\n```\n\n### Live sessions\n\n`exec.session()` opens a WebSocket to a process that is *still running* — stdin,\nstreamed stdout/stderr, PTY resize, and signals. It is the one feature not carried\nby `httpx`; the `websockets` dependency is imported lazily, only when a session\nopens. Available on `Box`, `AsyncBox`, and both ephemeral clients.\n\n```python\nsession = box.exec.session(\n    cmd=\"sort\",               # run via `bash -lc`; argv=[\"sort\"] runs the program with\n                              # no shell and takes precedence over cmd\n    cwd=\"/workspace/home\",    # defaults to the box's tracked cwd\n    env=[\"LOG_LEVEL=debug\"],  # KEY=VALUE entries overlaid on the box environment\n    on_stdout=lambda b: print(b.decode(), end=\"\"),  # bytes\n    on_stderr=lambda b: print(b.decode(), end=\"\"),  # separate stream unless tty\n)\n\nsession.pid       # in-box PID, always non-zero\nsession.exec_id   # server-side exec id\nsession.write(\"banana\\napple\\n\")\nsession.end_stdin()             # EOF — a command that reads to EOF now exits by itself\nexit_code = session.wait()      # -1 if torn down while still running; wait(timeout=5) raises TimeoutError\nsession.close()                 # hang up; also kills the process\n\n# Interactive programs / TUIs — tty allocates a real PTY, merging stderr into stdout\nwith box.exec.session(argv=[\"bash\", \"-i\"], tty=True, rows=40, cols=120) as shell:\n    shell.resize(50, 160)\n    shell.kill(\"INT\")      # allowlist: TERM KILL INT HUP TSTP QUIT USR1 USR2 (default TERM)\n    shell.terminate(5000)  # server-side SIGTERM, then SIGKILL after the grace (first call wins)\n```\n\nThe session owns the process: `close()` (or leaving the `with` block), a dropped\nconnection, or the program exiting all kill the command, and a session cannot be\nreattached. Use `wait()` to run something to completion.\n\nOn `AsyncBox` every handle method is a coroutine (`await session.write(...)`,\n`await session.wait()`, `async with await box.exec.session(...) as s:`) and an\n`async` callback is awaited; `wait()` there takes no timeout. In the sync client\nthe callbacks run on a background reader\nthread — keep them short, and never call `wait()` from inside one, since the exit\nframe it waits for arrives on the very thread it is blocking.\n\n## Filesystem\n\n```python\nbox.files.write(path=\"/workspace/home/app.py\", content=\"print('hi')\")\ncontent = box.files.read(\"/workspace/home/app.py\")\nentries = box.files.list(\"/workspace/home\")  # [FileEntry(name, path, size, is_dir, mod_time)]\n\n# Binary files — use encoding=\"base64\" for read and write\nbox.files.write(path=\"/workspace/home/image.png\", content=base64_string, encoding=\"base64\")\nb64 = box.files.read(\"/workspace/home/image.png\", encoding=\"base64\")\n\n# Bounded byte-range read — the *presence* of `length` selects the range, so\n# length=0 reads zero bytes rather than the whole file. Server caps it at 8 MiB.\nhead = box.files.read(\"/workspace/home/big.log\", length=64 * 1024)\nchunk = box.files.read(\"/workspace/home/big.log\", offset=1024, length=512)\n\n# Metadata — defaults to lstat, so a symlink reports type \"symlink\"\nstat = box.files.stat(\"/workspace/home/app.py\")\n# stat: FileStat(type=\"file\" | \"directory\" | \"symlink\" | \"other\", size, mod_time, inode, version)\n# `version` is an opaque freshness token (inode + mtime + size) for optimistic-concurrency\n# guards — compare it for equality, never parse it.\ntarget = box.files.stat(\"/workspace/home/link\", follow=True)  # dereference\n\n# Directories, moves, deletes\nbox.files.mkdir(\"build/cache\", parents=True)  # parents mirrors `mkdir -p`\nbox.files.rename(\"draft.md\", \"docs/final.md\")  # positional (from_path, to_path)\nbox.files.remove(\"build/cache\", recursive=True)  # recursive required for a directory\n\n# Upload local files\nbox.files.upload([{\"path\": \"./local/file.txt\", \"destination\": \"/workspace/home/file.txt\"}])\n\n# Download — `folder` is a path INSIDE the box; files land in ./<basename>\nbox.files.download(folder=\"src\")  # → ./src\nbox.files.download()  # whole cwd → ./workspace\n```\n\n## cd / Working Directory\n\nThe SDK tracks `cwd` client-side. All operations (exec, files, git, agent) run relative to it.\n\n```python\nbox.cwd  # current working directory (starts at /workspace/home)\nbox.cd(\"my-repo\")                  # relative to current cwd\nbox.cd(\"/workspace/home/other\")    # absolute path\n```\n\n## Git\n\nClones land inside the box's isolated container, never on the caller's machine. Cloned\ncode is data until something runs it — treat an untrusted repo as untrusted input, and\npair it with a restrictive `network_policy` (see below) before running its build or tests.\n\nEvery git call except `clone` runs in the box's current directory, so `cd` into the\nclone first. At the workspace root there is no repository, and `status` comes back\nempty, which reads as a clean tree.\n\n```python\nbox.git.clone(repo=\"github.com/org/repo\", branch=\"main\")\nbox.git.clone(repo=\"github.com/org/repo\", depth=1)  # shallow clone\nbox.git.clone(repo=\"github.com/org/repo\", folder=\"my-app\")  # destination\nbox.cd(\"repo\")  # the clone lands in a directory named after the repo\n\nstatus = box.git.status()\ndiff = box.git.diff()\nresult = box.git.commit(  # GitCommitResult(sha, message)\n    message=\"fix: resolve bug\",\n    author_name=\"Jane Doe\",  # optional per-commit override\n    author_email=\"jane@example.com\",\n)\nbox.git.push(branch=\"feature/fix\")\n\nbox.git.checkout(branch=\"release/v2\")\npr = box.git.create_pr(title=\"Fix bug\", body=\"...\", base=\"main\")\n# pr: PullRequest(url, number, title, base)\n\n# Update the box-wide git identity\ncfg = box.git.update_config(user_name=\"Bot\", user_email=\"bot@example.com\")\n# cfg: GitConfigResult(git_user_name, git_user_email)\n\n# Arbitrary git commands — returns the output string only. Unlike the JS SDK, which\n# returns { output, exit_code }, Python drops the status, so a failure (exit 128 when\n# the cwd is not a repository) is indistinguishable from success — check the output.\noutput = box.git.exec(args=[\"log\", \"--oneline\", \"-5\"])\n```\n\n## Schedules\n\nCron tasks on a box — shell commands or agent prompts. Available on `Box` and `EphemeralBox`. Cron is UTC.\n\n```python\nexec_schedule = box.schedule.exec(\n    cron=\"* * * * *\",\n    command=[\"bash\", \"-c\", \"date >> /workspace/home/cron.log\"],\n    folder=\"/workspace/home\",  # optional cwd override\n    webhook_url=\"https://example.com/hook\",\n    webhook_headers={\"Authorization\": \"Bearer ...\"},\n)\n\nagent_schedule = box.schedule.agent(\n    cron=\"0 9 * * *\",\n    prompt=\"Run the test suite and fix any failures\",\n    folder=\"/workspace/home/repo\",  # optional cwd override\n    model=\"anthropic/claude-sonnet-5\",  # optional override\n    options={\"max_budget_usd\": 1.0, \"effort\": \"high\"},\n    timeout=300_000,\n    webhook_url=\"https://example.com/hook\",\n    webhook_headers={\"Authorization\": \"Bearer ...\"},\n)\n\nschedules = box.schedule.list()\none = box.schedule.get(agent_schedule.id)\n\n# Partial update — omitted args keep their value, \"\" / [] / {} clear a field,\n# options=None clears agent options. The schedule's type cannot change.\n# Updatable: cron, command, prompt, folder, model, options, timeout,\n#            webhook_url, webhook_headers\nbox.schedule.update(agent_schedule.id, cron=\"0 18 * * *\", webhook_url=\"\")\n\nbox.schedule.pause(agent_schedule.id)\nbox.schedule.resume(agent_schedule.id)\nbox.schedule.delete(agent_schedule.id)\n```\n\n## Snapshots\n\n```python\n# Snapshot — checkpoint workspace state\nsnap = box.snapshot(name=\"after-setup\")\n# snap: Snapshot(id, name, box_id, size_bytes, status, created_at)\n\n# from_snapshot takes the same BoxConfig kwargs as create (shared request body):\n# name, labels, size, keep_alive, init_command, runtime, browser, agent, git, env,\n# attach_headers, network_policy, skills, mcp_servers. Note the JS SDK's\n# Box.fromSnapshot() drops browser / skills / mcpServers — Python forwards them.\nrestored = Box.from_snapshot(\n    snap.id,\n    size=\"medium\",\n    keep_alive=True,\n    # the git identity is forwarded, not just the token\n    git={\"token\": os.environ[\"GITHUB_TOKEN\"], \"user_name\": \"Bot\", \"user_email\": \"bot@example.com\"},\n    env={\"DATABASE_URL\": \"...\"},\n)\nsnaps = box.list_snapshots()\nbox.delete_snapshot(snap.id)\n```\n\n## Browser\n\nCreate the box with `browser=True` to drive a headless Chromium. Tab management\nlives on `box.browser`; every page operation lives on the tab handle.\n`extract` / `observe` / `act(instruction)` are AI-powered and metered;\n`act(action)` replays an already-resolved action with no LLM call and no tokens.\n\n```python\nfrom pydantic import BaseModel\n\nbox = Box.create(browser=True, agent={\"harness\": Agent.CLAUDE_CODE, \"model\": ClaudeCode.SONNET_4_5})\n\n# Tabs\ntab = box.browser.tab.create(\"https://example.com\", wait_until=\"load\", timeout=30_000)\ntabs = box.browser.list_tabs()\nagain = box.browser.get_tab(tab.id)  # no network call\ntab.id, tab.url, tab.title  # handle metadata, no network call\n\n# Page operations\ncontent = tab.goto(\"https://news.ycombinator.com\")  # BrowserContent(title, url, text, links)\ncurrent = tab.content()\npng = tab.screenshot()  # bytes\nb64 = tab.screenshot(encoding=\"base64\", full_page=True)\n\n# AI operations (metered) — schema is a Pydantic model or a raw JSON-schema dict.\n# extract / observe / act take an optional model= override, defaulting to the box's\n# model (or anthropic/claude-sonnet-4-5 when it has none).\nclass Story(BaseModel):\n    title: str\n    points: int\n\ndata = tab.extract(\"Top story title and points\", Story, model=\"anthropic/claude-sonnet-4-5\")\n\n# observe → actionable elements, each carrying a replayable method + arguments\nelements = tab.observe(\"What can I click?\", model=\"openai/gpt-5.6\").elements\n# elements: [BrowserObserveElement(description, selector, url, method, arguments)]\n\nacted = tab.act(\"Click the first headline\")\n# BrowserActResult(success, message, action_description, actions, cache_status,\n#                  input_tokens, output_tokens)\n\n# Replay a pre-resolved action — no LLM call, no tokens, no model provider key.\n# Pass a BrowserObserveElement or BrowserActAction instead of a string; `model` is\n# ignored in this form, and an action without a `selector` raises BoxError.\ntab.act(elements[0])\ntab.act(acted.actions[0])\n\n# Live view + raw CDP\nlive_url = tab.live_view_url()  # view-only screencast page/iframe\ncdp_url = box.browser.cdp_url()  # wss://…?token=… — no extra auth wiring\ntab.close()\n\n# Drive the same browser from Playwright (pip install playwright)\nfrom playwright.sync_api import sync_playwright\n\nwith sync_playwright() as p:\n    remote = p.chromium.connect_over_cdp(cdp_url)\n    context = remote.contexts[0] if remote.contexts else remote.new_context()\n    page = context.pages[0] if context.pages else context.new_page()\n    page.goto(\"https://example.com\")\n\n# Session recordings (HLS playback URL + MP4 download, chapter markers).\n# One active recording per box; captures all tabs and follows the foreground.\n# Auto-stops after max_duration_seconds or ~3 minutes of no on-screen activity.\nhandle = box.browser.recordings.start(max_duration_seconds=600)  # default & max 600\nrecording = handle.stop()\n# or stop whatever is recording on the box, without a handle:\n# recording = box.browser.recordings.stop()\n# BrowserRecording(id, box_id, status, started_at, ended_at, duration_ms, size_bytes,\n#                  mp4_size_bytes, segment_count, markers, stopped_reason,\n#                  max_duration_seconds, expires_at, playlist_url)\n# markers: BrowserRecordingMarker(type=\"tab_switch\", at_ms, end_ms, label, tab_id)\n# expires_at is epoch ms (videos retained 14 days); playlist_url is API-served — fetch it\n# with an `X-Box-Api-Key: <api_key>` header (hls.js / Safari / ffplay).\nall_recordings = box.browser.recordings.list()\none_recording = box.browser.recordings.get(recording.id)\n\n# Download the video to a local file — returns the path written.\n# Defaults to ./box-recording-<id>.mp4 (.ts for recordings captured before MP4 support).\nfile = box.browser.recordings.download(recording.id)\nbox.browser.recordings.download(recording.id, path=\"./out/demo.mp4\")\n```\n\n### Multi-step browser goals\n\n`tab.run()` — the autonomous multi-step browser agent — was **removed**, along with\nthe `BrowserRunResult` / `BrowserRunStep` types (Stagehand v4 dropped the underlying\nagent primitive). The browser now exposes `observe`, `act`, and `extract` only.\nThree replacements:\n\n**1. Drive your own loop** — resolve steps once with `observe`, then replay them\nwith `act(action)` so the model stays out of the hot path; `extract` is the stop check.\n\n```python\nclass Product(BaseModel):\n    title: str\n    price: str\n\nelements = tab.observe(\"the product links in the listing\").elements\nactions = [e for e in elements if e.selector]\n\nfor action in actions[:5]:\n    tab.goto(START)   # deterministic reset, no browser-AI tokens\n    tab.act(action)   # replay the resolved click: no LLM, no tokens\n    item = tab.extract(\"title and price\", Product)\n```\n\n**2. Hand the goal to the in-box agent** — `browser=True` auto-wires the\nchrome-devtools MCP (Chromium already warmed on 127.0.0.1:9222) into the box's\ncoding agent, so `box.agent.run(prompt=...)` drives the browser itself and iterates\nuntil done. No `tab.create()` needed first. This bills coding-agent model tokens\nrather than browser-AI metering, and needs an agent harness + key.\n\n**3. Connect over CDP** with Playwright via `box.browser.cdp_url()` when the flow is\nfully deterministic.\n\n## EphemeralBox\n\nLightweight, short-lived boxes (max 3 days). Supports `exec`, `files`, `schedule`,\n`cd`, network policy, and snapshots. No `agent`, `git`, `skills`, `labels`\nnamespace, browser, or public URLs.\n\n```python\nfrom upstash_box import EphemeralBox\n\nebox = EphemeralBox.create(\n    name=\"scratch-box\",\n    runtime=\"python\",\n    size=\"small\",\n    ttl=3600,  # seconds, max 259200 (3 days), default 259200\n    env={\"API_KEY\": \"...\"},\n    labels=[\"scratch\"],  # settable at create time; filter via Box.list(label=...)\n    network_policy={\"mode\": \"deny-all\"},\n    attach_headers={\"api.stripe.com\": {\"Authorization\": \"Bearer sk_live_...\"}},\n)\n\nebox.network_policy\n\nebox.expires_at  # unix timestamp when auto-deleted\nebox.exec.command(\"python -c 'print(1+1)'\")\nebox.exec.code(code=\"print('hi')\", lang=\"python\")\nebox.exec.session(argv=[\"bash\", \"-i\"], tty=True)  # whole exec namespace, session included\nebox.files.write(path=\"/workspace/home/data.json\", content=\"{}\")\nebox.files.stat(\"/workspace/home/data.json\")  # whole files namespace, stat/mkdir/rename/remove included\nebox.schedule.exec(cron=\"* * * * *\", command=[\"bash\", \"-c\", \"date\"])\nebox.cd(\"subdir\")\nsnap = ebox.snapshot(name=\"checkpoint\")\nebox.list_snapshots()\nebox.delete_snapshot(snap.id)\nstatus = ebox.get_status()[\"status\"]\nebox.delete()\n\n# Restore from snapshot\nebox2 = EphemeralBox.from_snapshot(snap.id, ttl=7200)\n\n# Statics: EphemeralBox.delete_boxes(box_ids=[...]) / EphemeralBox.delete_snapshots(...)\n# are the Box ones. EphemeralBox.get_by_name() returns a full `Box`, not an\n# `EphemeralBox` (quirk mirrored from the JS SDK).\n# `AsyncEphemeralBox` is the async variant (`await AsyncEphemeralBox.create(...)`).\n```\n\n## Public URLs\n\nExpose box ports as public URLs with optional auth.\n\n```python\npublic_url = box.get_public_url(3000)\n# public_url: PublicURL(url=\"https://{id}-3000.preview.box.upstash.com\", port)\n\nauthed = box.get_public_url(3000, bearer_token=True)\n# authed: PublicURL(url, port, token)\n\nbasic = box.get_public_url(3000, basic_auth=True)\n# basic: PublicURL(url, port, username, password)\n\nresult = box.list_public_urls()  # {\"public_urls\": [PublicURL, ...]}\nbox.delete_public_url(3000)\n```\n\n## Skills\n\nInstall agent skills from the Context7 registry. Format: `owner/repo/skill-name`.\n\nAn installed skill becomes instructions for the box's agent, so pin skills to owners you\ntrust the same way you would a dependency. Skills resolve from the registry at box\ncreation, not from arbitrary URLs, and they only ever run inside the box's container.\n\n```python\nbox = Box.create(skills=[\"upstash/qstash-js/qstash-js\"])\n\nbox.skills.add(\"upstash/workflow-js/workflow-js\")\nenabled = box.skills.list()\nbox.skills.remove(\"upstash/workflow-js/workflow-js\")\n```\n\n## Labels\n\n```python\nlabels = box.labels.add(\"prod\")  # returns the updated set\nbox.labels.remove(\"beta\")\ncurrent = box.labels.list()\nprod_boxes = Box.list(label=\"prod\")\n```\n\n## Network Policy & Outbound Headers\n\n```python\nbox = Box.create(\n    # mode: \"allow-all\" (default) | \"deny-all\" | \"custom\"\n    # custom takes any of allowed_domains / allowed_cidrs / denied_cidrs\n    network_policy={\n        \"mode\": \"custom\",\n        \"allowed_domains\": [\"api.example.com\"],\n        \"allowed_cidrs\": [\"203.0.113.0/24\"],\n        \"denied_cidrs\": [\"10.0.0.0/8\"],\n    },\n    # Inject secret headers into matching outbound HTTPS requests (write-only, never read back)\n    attach_headers={\n        \"api.stripe.com\": {\"Authorization\": \"Bearer sk_live_...\"},\n        \"*.example.com\": {\"X-Custom-Token\": \"secret123\"},\n    },\n)\n\nbox.network_policy\nbox.update_network_policy({\"mode\": \"deny-all\"})\n```\n\n## MCP Servers\n\nAttach MCP servers to the box agent. An attached server supplies tools the agent can call,\nso use servers you control or trust — and keep `network_policy` restrictive when the agent\nalso handles untrusted input.\n\n```python\nbox = Box.create(\n    agent={\"harness\": Agent.CLAUDE_CODE, \"model\": ClaudeCode.SONNET_4_5},\n    mcp_servers=[\n        {\"name\": \"fs\", \"package\": \"@modelcontextprotocol/server-filesystem\"},\n        {\"name\": \"custom\", \"url\": \"<your-mcp-server-url>\", \"headers\": {\"Authorization\": \"...\"}},\n    ],\n)\n```\n\n## Errors & SSH\n\n```python\nfrom upstash_box import BoxError\n\ntry:\n    box.agent.run(prompt=\"...\")\nexcept BoxError as e:\n    print(e, e.status_code)\n```\n\nShell into a box directly (Box API key is the SSH password):\n\n```bash\nssh <box-id>@us-east-1.box.upstash.com\n```\n\n## Async client\n\nThe async client mirrors the sync API exactly — `await` the calls and use `async for` to stream.\n\n```python\nimport asyncio\nfrom upstash_box import AsyncBox, Agent\n\nasync def main():\n    box = await AsyncBox.create(runtime=\"node\", agent={\"harness\": Agent.CLAUDE_CODE})\n    async with box:\n        run = await box.agent.run(prompt=\"Set up a Next.js project\")\n        print(run.result)\n\n        stream = await box.agent.stream(prompt=\"Build a REST API\")\n        async for chunk in stream:\n            print(chunk)\n\n        await box.delete()\n\nasyncio.run(main())\n```\n\n`asyncio.gather` over many `AsyncBox.create(...)` / `box.agent.run(...)` calls runs boxes in parallel.\n\n## Gotchas\n\n- Public API option keys are **snake_case** in Python: `api_key`, `user_name`, `network_policy`, `response_schema`, `max_retries`, `on_tool_use`, `attach_headers`, and agent `options` like `max_turns`, `max_budget_usd`.\n- Agent config takes **`harness`** (not the deprecated `provider`/`runner`) — `harness` is required.\n- `response_schema` accepts a Pydantic `BaseModel` subclass (returns a typed instance) or a raw JSON-schema `dict` (returns a `dict`). Browser `schema` follows the same contract.\n- Default working directory is `/workspace/home`, not `/home` or `/`.\n- `box.cd()` is client-side tracking — it validates the path exists but doesn't change the box's shell cwd. All SDK methods use it automatically.\n- `EphemeralBox` does NOT support `agent`, `git`, `skills`, the `labels` namespace, the browser, or public URLs — use full `Box` for those (it does support `schedule` and snapshots).\n- `run.exit_code` is `None` for agent runs, only available for exec commands.\n- `run.result` is stdout on success and stderr on failure — a command that exits 0 writing only to stderr yields `\"\"`; read `run.stderr` for it.\n- `files.download(folder=...)` takes a path *inside the box*; output lands in `./<basename>` locally.\n- `files.read()` slices only when `length` is given — `offset=` alone reads the whole file, and `length=0` reads nothing.\n- `files.stat()` is an lstat by default: a symlink reports `type=\"symlink\"` unless you pass `follow=True`.\n- `files.remove()` needs `recursive=True` for a directory, and `files.mkdir()` needs `parents=True` for nested paths.\n- `files.rename(from_path, to_path)` takes positional arguments (`from` is a Python keyword); JS spells it `rename(from, to)`.\n- `exec.session()` handles own the process — `close()` or a dropped connection kills the command, and sessions cannot be reattached. `tty=True` merges stderr into stdout, so `on_stderr` never fires for a PTY session.\n- The sync `session.wait(timeout=...)` has no async counterpart (`await handle.wait()` blocks until exit); it raises `TimeoutError` when the timeout elapses.\n- `box.browser` requires a box created with `browser=True`.\n- There is **no** `tab.run()` — the autonomous browser agent was removed. Loop `observe` + `act(action)` + `extract` yourself, hand the goal to the in-box agent, or drive Playwright over `cdp_url()`.\n- `tab.act(action)` (replaying an `observe()` result) costs no tokens and needs no model provider key; only `act(instruction)` with a string is metered.\n- `get_init_command` / `set_init_command` / `delete_init_command` raise unless the box was created with `keep_alive=True`.\n- The JS static `Box.delete({boxIds})` is `Box.delete_boxes(box_ids=...)` here, to avoid clashing with the instance `delete()`.\n- `box.delete()` is irreversible — snapshot first if you need the state.\n- Git operations require `git.token` in the box config for private repos and PRs.\n- `Box.from_snapshot()` creates a new box — it does not modify the original. It reuses the full create body, so `browser` / `skills` / `mcp_servers` are forwarded (the JS `Box.fromSnapshot()` drops those).\n- `EphemeralBox` has no `update_network_policy` — set `network_policy` at create time.\n- All `timeout` values are in **milliseconds** (matching the JS SDK), default `600000`.\n- When breaking out of a stream early, call `stream.close()` / `await stream.aclose()` so the run is marked `detached`.\n- Close the transport when done: `box.delete()` closes it, or use `with box:` / `box.close()` (`async with` / `await box.aclose()` for `AsyncBox`).\n"
}