| name | upstash-box-py |
| description | Work with the upstash-box Python SDK for sandboxed cloud containers with AI agents, shell, filesystem, git, cron schedules, and a headless browser. Use when building with Upstash Box in Python, creating sandboxed environments, running AI agents in containers, browser automation from a box, or orchestrating parallel boxes. |
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
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
import os
from upstash_box import Box, Agent, ClaudeCode, BoxApiKey
box = Box.create(
name="my-box",
runtime="node",
size="small",
labels=["beta", "x-team"],
keep_alive=True,
init_command="npm install && npm run dev",
browser=True,
agent={
"harness": Agent.CLAUDE_CODE,
"model": ClaudeCode.SONNET_4_5,
"api_key": BoxApiKey.UPSTASH_KEY,
},
git={
"token": os.environ["GITHUB_TOKEN"],
"user_name": "Bot",
"user_email": "bot@example.com",
},
env={"DATABASE_URL": "..."},
skills=["upstash/qstash-js/qstash-js"],
timeout=600_000,
debug=False,
)
same = Box.get(box.id, git_token="ghp_...")
by_name = Box.get_by_name("my-box")
all_boxes = Box.list()
beta = Box.list(label="beta")
box.pause()
box.resume()
box.delete()
status = box.get_status()["status"]
box.id, box.size, box.keep_alive, box.cwd, box.network_policy
box.set_init_command("npm run dev")
script = box.get_init_command()
box.delete_init_command()
Box.delete_boxes(box_ids=["box_1", "box_2"])
Box.delete_snapshots(snapshot_ids=["snap_1"])
Account-level env vars
Injected into every box you create.
Box.set_env("API_TOKEN", "secret")
env = Box.list_env()
Box.set_all_env({"A": "1", "B": "2"})
Box.delete_env("API_TOKEN")
Agent Runs
from pydantic import BaseModel
class Finding(BaseModel):
severity: str
file: str
issue: str
class Review(BaseModel):
verdict: str
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"},
on_tool_use=lambda tool: print(tool["name"], tool["input"]),
on_tool_result=lambda result: print(result["tool_call_id"], result["output"]),
)
run.status
run.result
run.cost
box.agent.run(prompt="Describe this", files=["./screenshot.png"])
box.agent.run(
prompt="Describe this",
files=[{"data": b64, : , : }],
)
stream = box.agent.stream(prompt=)
chunk stream:
chunk. == :
(chunk.text, end=)
chunk. == :
(chunk.text, end=)
chunk. == :
(chunk.tool_name, chunk.)
chunk. == :
(chunk.output)
chunk. == :
(chunk.usage.input_tokens, chunk.usage.cached_input_tokens, chunk.session_id)
box.agent.run(
prompt=,
webhook={: , : {: }},
)
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.
{
"max_turns": 20,
"max_budget_usd": 1.0,
"effort": "high",
"thinking": {"type": "adaptive"},
"disallowed_tools": ["Bash"],
"agents": {"reviewer": {...}},
"prompt_suggestions": False,
"fallback_model": "anthropic/claude-sonnet-4-5",
"system_prompt": "You are a release engineer.",
}
{
"model_reasoning_effort": "high",
"model_reasoning_summary": "concise",
"personality": "pragmatic",
"web_search": "live",
}
{
"reasoning_effort": "high",
"text_verbosity": "low",
: ,
: {: , : },
}
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.
from upstash_box import ClaudeCode, OpenAICodex, OpenCodeModel, CursorModel, OpenRouterModel, VercelModel
ClaudeCode.OPUS_5
ClaudeCode.SONNET_5
OpenAICodex.GPT_5_6
OpenCodeModel.CLAUDE_OPUS_5
CursorModel.COMPOSER_2_5
OpenRouterModel.CLAUDE_OPUS_5
VercelModel.GPT_5_5
box.model_config
box.configure_model("anthropic/claude-opus-4-8")
from upstash_box import infer_default_provider
infer_default_provider("openai/gpt-5.6")
infer_default_provider("cursor/default")
Custom harness
Run your own agent process inside the box instead of a managed harness.
import asyncio
from upstash_box import Agent, Box, CustomHarnessDone, run_custom_harness
box = Box.create(
agent={
"harness": Agent.CUSTOM,
"model": "my-agent",
"custom_harness": {
"command": "python",
"args": ["/workspace/home/agent.py"],
"protocol": "box-sse-v1",
},
},
)
box.configure_custom_harness({"command": "python", "args": ["/workspace/home/agent2.py"]})
async def handler(ctx, emit):
emit.text("working...")
emit.reasoning("thinking out loud")
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"})
CustomHarnessDone(
output=,
input_tokens=,
output_tokens=,
cached_input_tokens=,
total_cost_usd=,
session_id=ctx.session_id,
)
asyncio.run(run_custom_harness(handler))
Run Fields
Every run (agent, command, or code) returns a Run:
run = box.exec.command("npm test")
run.id
run.status
run.result
run.stdout
run.stderr
run.exit_code
run.cost
run.cancel()
logs = run.logs()
entries = box.logs(limit=100, offset=0)
runs = box.list_runs()
Shell Execution
run = box.exec.command("echo hello && ls -la")
run2 = box.exec.code(code="print(1 + 1)", lang="python", timeout=10_000)
stream = box.exec.stream("npm run build")
stream2 = box.exec.stream_code(code="print('hi')", lang="python")
for chunk in stream:
...
Filesystem
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")
box.files.write(path="/workspace/home/image.png", content=base64_string, encoding="base64")
b64 = box.files.read("/workspace/home/image.png", encoding="base64")
box.files.upload([{"path": "./local/file.txt", "destination": "/workspace/home/file.txt"}])
box.files.download(folder="src")
box.files.download()
cd / Working Directory
The SDK tracks cwd client-side. All operations (exec, files, git, agent) run relative to it.
box.cwd
box.cd("my-repo")
box.cd("/workspace/home/other")
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.
box.git.clone(repo="github.com/org/repo", branch="main")
box.git.clone(repo="github.com/org/repo", depth=1)
box.cd("repo")
status = box.git.status()
diff = box.git.diff()
result = box.git.commit(
message="fix: resolve bug",
author_name="Jane Doe",
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")
cfg = box.git.update_config(user_name="Bot", user_email="bot@example.com")
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.
exec_schedule = box.schedule.exec(
cron="* * * * *",
command=["bash", "-c", "date >> /workspace/home/cron.log"],
folder="/workspace/home",
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",
model="anthropic/claude-sonnet-5",
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)
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
snap = box.snapshot(name="after-setup")
restored = Box.from_snapshot(
snap.id,
size="medium",
keep_alive=True,
git={"token": os.environ["GITHUB_TOKEN"], "user_name": "Bot", "user_email": "bot@example.com"},
env={"DATABASE_URL": "..."},
)
snaps = box.list_snapshots()
box.delete_snapshot(snap.id)
Browser
Create the box with browser=True to drive a headless Chromium. Tab management
lives on box.browser; every page operation lives on the tab handle.
extract / observe / act(instruction) are AI-powered and metered;
act(action) replays an already-resolved action with no LLM call and no tokens.
from pydantic import BaseModel
box = Box.create(browser=True, agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5})
tab = box.browser.tab.create("https://example.com", wait_until="load", timeout=30_000)
tabs = box.browser.list_tabs()
again = box.browser.get_tab(tab.id)
tab.id, tab.url, tab.title
content = tab.goto("https://news.ycombinator.com")
current = tab.content()
png = tab.screenshot()
b64 = tab.screenshot(encoding="base64", full_page=True)
class Story(BaseModel):
title: str
points: int
data = tab.extract("Top story title and points", Story, model="anthropic/claude-sonnet-4-5")
elements = tab.observe("What can I click?", model="openai/gpt-5.6").elements
acted = tab.act("Click the first headline")
tab.act(elements[])
tab.act(acted.actions[])
live_url = tab.live_view_url()
cdp_url = box.browser.cdp_url()
tab.close()
playwright.sync_api sync_playwright
sync_playwright() p:
remote = p.chromium.connect_over_cdp(cdp_url)
context = remote.contexts[] remote.contexts remote.new_context()
page = context.pages[] context.pages context.new_page()
page.goto()
handle = box.browser.recordings.start(max_duration_seconds=)
recording = handle.stop()
all_recordings = box.browser.recordings.()
one_recording = box.browser.recordings.get(recording.)
file = box.browser.recordings.download(recording.)
box.browser.recordings.download(recording., path=)
Multi-step browser goals
tab.run() — the autonomous multi-step browser agent — was removed, along with
the BrowserRunResult / BrowserRunStep types (Stagehand v4 dropped the underlying
agent primitive). The browser now exposes observe, act, and extract only.
Three replacements:
1. Drive your own loop — resolve steps once with observe, then replay them
with act(action) so the model stays out of the hot path; extract is the stop check.
class Product(BaseModel):
title: str
price: str
elements = tab.observe("the product links in the listing").elements
actions = [e for e in elements if e.selector]
for action in actions[:5]:
tab.goto(START)
tab.act(action)
item = tab.extract("title and price", Product)
2. Hand the goal to the in-box agent — browser=True auto-wires the
chrome-devtools MCP (Chromium already warmed on 127.0.0.1:9222) into the box's
coding agent, so box.agent.run(prompt=...) drives the browser itself and iterates
until done. No tab.create() needed first. This bills coding-agent model tokens
rather than browser-AI metering, and needs an agent harness + key.
3. Connect over CDP with Playwright via box.browser.cdp_url() when the flow is
fully deterministic.
EphemeralBox
Lightweight, short-lived boxes (max 3 days). Supports exec, files, schedule,
cd, network policy, and snapshots. No agent, git, skills, labels
namespace, browser, or public URLs.
from upstash_box import EphemeralBox
ebox = EphemeralBox.create(
name="scratch-box",
runtime="python",
size="small",
ttl=3600,
env={"API_KEY": "..."},
labels=["scratch"],
network_policy={"mode": "deny-all"},
attach_headers={"api.stripe.com": {"Authorization": "Bearer sk_live_..."}},
)
ebox.network_policy
ebox.expires_at
ebox.exec.command("python -c 'print(1+1)'")
ebox.exec.code(code="print('hi')", lang="python")
ebox.files.write(path="/workspace/home/data.json", content="{}")
ebox.schedule.exec(cron="* * * * *", command=["bash", "-c", "date"])
ebox.cd("subdir")
snap = ebox.snapshot(name="checkpoint")
ebox.list_snapshots()
ebox.delete_snapshot(snap.id)
status = ebox.get_status()["status"]
ebox.delete()
ebox2 = EphemeralBox.from_snapshot(snap.id, ttl=7200)
Public URLs
Expose box ports as public URLs with optional auth.
public_url = box.get_public_url(3000)
authed = box.get_public_url(3000, bearer_token=True)
basic = box.get_public_url(3000, basic_auth=True)
result = box.list_public_urls()
box.delete_public_url(3000)
Skills
Install agent skills from the Context7 registry. Format: owner/repo/skill-name.
An installed skill becomes instructions for the box's agent, so pin skills to owners you
trust the same way you would a dependency. Skills resolve from the registry at box
creation, not from arbitrary URLs, and they only ever run inside the box's container.
box = Box.create(skills=["upstash/qstash-js/qstash-js"])
box.skills.add("upstash/workflow-js/workflow-js")
enabled = box.skills.list()
box.skills.remove("upstash/workflow-js/workflow-js")
Labels
labels = box.labels.add("prod")
box.labels.remove("beta")
current = box.labels.list()
prod_boxes = Box.list(label="prod")
Network Policy & Outbound Headers
box = Box.create(
network_policy={
"mode": "custom",
"allowed_domains": ["api.example.com"],
"allowed_cidrs": ["203.0.113.0/24"],
"denied_cidrs": ["10.0.0.0/8"],
},
attach_headers={
"api.stripe.com": {"Authorization": "Bearer sk_live_..."},
"*.example.com": {"X-Custom-Token": "secret123"},
},
)
box.network_policy
box.update_network_policy({"mode": "deny-all"})
MCP Servers
Attach MCP servers to the box agent. An attached server supplies tools the agent can call,
so use servers you control or trust — and keep network_policy restrictive when the agent
also handles untrusted input.
box = Box.create(
agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5},
mcp_servers=[
{"name": "fs", "package": "@modelcontextprotocol/server-filesystem"},
{"name": "custom", "url": "<your-mcp-server-url>", "headers": {"Authorization": "..."}},
],
)
Errors & SSH
from upstash_box import BoxError
try:
box.agent.run(prompt="...")
except BoxError as e:
print(e, e.status_code)
Shell into a box directly (Box API key is the SSH password):
ssh <box-id>@us-east-1.box.upstash.com
Async client
The async client mirrors the sync API exactly — await the calls and use async for to stream.
import asyncio
from upstash_box import AsyncBox, Agent
async def main():
box = await AsyncBox.create(runtime="node", agent={"harness": Agent.CLAUDE_CODE})
async with box:
run = await box.agent.run(prompt="Set up a Next.js project")
print(run.result)
stream = await box.agent.stream(prompt="Build a REST API")
async for chunk in stream:
print(chunk)
await box.delete()
asyncio.run(main())
asyncio.gather over many AsyncBox.create(...) / box.agent.run(...) calls runs boxes in parallel.
Gotchas
- 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.
- Agent config takes
harness (not the deprecated provider/runner) — harness is required.
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.
- Default working directory is
/workspace/home, not /home or /.
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.
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).
run.exit_code is None for agent runs, only available for exec commands.
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.
files.download(folder=...) takes a path inside the box; output lands in ./<basename> locally.
box.browser requires a box created with browser=True.
- 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().
- (replaying an result) costs no tokens and needs no model provider key; only with a string is metered.