upstash-box-py

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

upstash-box Python SDK

upstash-box Python SDK

Sandboxed cloud containers with built-in AI agents, shell, filesystem, git, cron schedules, and an optional headless browser.
内置AI Agent、Shell、文件系统、Git、定时任务及可选无头浏览器的沙箱化云容器。

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.
bash
pip install upstash-box
设置
UPSTASH_BOX_API_KEY
环境变量,或在构造函数中传入
api_key
参数。
SDK同时提供同步版本
Box
(以下示例使用该版本)和异步版本
AsyncBox
(使用方式为
box = await AsyncBox.create(...)
await box.agent.run(...)
)。异步接口与同步接口完全一致,仅需添加
await
async for
关键字。
每次请求会发送匿名遥测头信息;可通过设置
UPSTASH_DISABLE_TELEMETRY
环境变量选择退出。

Box Lifecycle

Box生命周期

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

Create with agent + git + env vars

创建包含Agent、Git及环境变量的Box

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, )
box = Box.create( name="my-box", runtime="node", # 可选值:"node" | "python" | "golang" | "ruby" | "rust"(及对应的"-alpine"变体) size="small", # 可选值:"small"(2核/4GB) | "medium"(4核/8GB) | "large"(8核/16GB) labels=["beta", "x-team"], # 最多5个标签,每个标签长度不超过20字符 keep_alive=True, # 禁用空闲暂停机制 init_command="npm install && npm run dev", # 仅对keep-alive模式的Box生效 browser=True, # 为box.browser配置无头Chromium agent={ "harness": Agent.CLAUDE_CODE, # 可选值:Agent.CODEX | Agent.OPEN_CODE | Agent.CURSOR | Agent.CUSTOM "model": ClaudeCode.SONNET_4_5, # 也可直接传入字符串"anthropic/claude-sonnet-4-5" # api_key选项: # 省略 → 由服务器决定使用哪个密钥 # BoxApiKey.UPSTASH_KEY → 使用Upstash提供的LLM密钥 # BoxApiKey.STORED_KEY → 使用之前通过Upstash控制台存储的密钥 # "sk-..." → 直接传入API密钥字符串 "api_key": BoxApiKey.UPSTASH_KEY, }, git={ # 所有字段均为可选 "token": os.environ["GITHUB_TOKEN"], # 或通过Upstash控制台关联GitHub账户 "user_name": "Bot", "user_email": "bot@example.com", }, env={"DATABASE_URL": "..."}, skills=["upstash/qstash-js/qstash-js"], # 格式:owner/repo/skill-name timeout=600_000, # 请求超时时间(毫秒) debug=False, )

Reconnect, list, delete, pause/resume

重新连接、列举、删除、暂停/恢复Box

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() box.resume() box.delete() # irreversible status = box.get_status()["status"]
box.id, box.size, box.keep_alive, box.cwd, box.network_policy
same = Box.get(box.id, git_token="ghp_...") # 重新连接时传入git_token,而非git={...} 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

Init command (keep-alive boxes only — raises otherwise)

设置初始化命令(仅对keep-alive模式的Box生效,否则会报错)

box.set_init_command("npm run dev") script = box.get_init_command() box.delete_init_command()
box.set_init_command("npm run dev") script = box.get_init_command() box.delete_init_command()

Bulk delete (classmethods, by ID)

批量删除(类方法,按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
undefined
Box.delete_boxes(box_ids=["box_1", "box_2"]) # JS中的静态方法
delete
对应Python中的
delete_boxes
Box.delete_snapshots(snapshot_ids=["snap_1"]) # 省略ids参数会删除所有快照
undefined

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")
会注入到你创建的所有Box中。
python
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

Agent运行

python
from pydantic import BaseModel
python
from pydantic import BaseModel

Structured output with a Pydantic model (or a raw JSON-schema dict)

使用Pydantic模型定义结构化输出(也可传入原始JSON Schema字典)

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)
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="检查代码中的安全问题", response_schema=Review, timeout=120_000, max_retries=2, options={"max_turns": 20, "max_budget_usd": 1.0, "effort": "high"}, # 特定于harness的选项 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 # 符合Schema的类型化结果(Review实例) 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)

为提示词附加文件(最多10个文件,单个文件不超过10MB)

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"}], )
box.agent.run(prompt="描述这个文件", files=["./screenshot.png"]) box.agent.run( prompt="描述这个文件", files=[{"data": b64, "media_type": "image/png", "filename": "shot.png"}], )

Streaming — chunks are typed dataclasses discriminated on
.type

流式输出——chunk为类型化数据类,通过
.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)
stream = box.agent.stream(prompt="构建一个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) # 其他类型:StartChunk(run_id) | StatsChunk(cpu_ns, memory_peak_bytes) | UnknownChunk(event, data)

Fire-and-forget with webhook

异步调用并通过Webhook返回结果

box.agent.run( prompt="Run tests", webhook={"url": "https://example.com/hook", "headers": {"Authorization": "Bearer ..."}}, )
undefined
box.agent.run( prompt="运行测试", webhook={"url": "https://example.com/hook", "headers": {"Authorization": "Bearer ..."}}, )
undefined

Harness & model

Harness与模型

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

ClaudeCode.OPUS_5  # "anthropic/claude-opus-5"
ClaudeCode.SONNET_5  # "anthropic/claude-sonnet-5"
OpenAICodex.GPT_5_6  # "openai/gpt-5.6"
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"
harness
为必填项。模型枚举包括:
ClaudeCode
OpenAICodex
OpenCodeModel
CursorModel
OpenRouterModel
VercelModel
——也可传入任何带提供商前缀的字符串。
python
from upstash_box import ClaudeCode, OpenAICodex, CursorModel, OpenRouterModel, VercelModel

ClaudeCode.OPUS_5  # 对应字符串"anthropic/claude-opus-5"
ClaudeCode.SONNET_5  # 对应字符串"anthropic/claude-sonnet-5"
OpenAICodex.GPT_5_6  # 对应字符串"openai/gpt-5.6"
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的harness与模型配置

box.model_config # {"harness": ..., "model": ...} box.configure_model("anthropic/claude-opus-4-8")
undefined
box.model_config # 返回{"harness": ..., "model": ...} box.configure_model("anthropic/claude-opus-4-8")
undefined

Custom harness

自定义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
        "custom_harness": {"command": "python", "args": ["/workspace/home/agent.py"]},
    },
)
box.configure_custom_harness({"command": "python", "args": ["/workspace/home/agent2.py"]})
在Box内部运行你自己的Agent进程,而非使用托管Harness。
python
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"]},
    },
)
box.configure_custom_harness({"command": "python", "args": ["/workspace/home/agent2.py"]})

Inside the box, agent.py emits box-sse-v1 events:

在Box内部,agent.py需要发送box-sse-v1事件:

async def handler(ctx, emit): emit.text("working...") emit.tool({"name": "Bash", "input": {"command": "ls"}}) return CustomHarnessDone(output="done", input_tokens=10, output_tokens=5)
asyncio.run(run_custom_harness(handler)) # run_custom_harness is async
undefined
async def handler(ctx, emit): emit.text("working...") emit.tool({"name": "Bash", "input": {"command": "ls"}}) return CustomHarnessDone(output="done", input_tokens=10, output_tokens=5)
asyncio.run(run_custom_harness(handler)) # run_custom_harness为异步函数
undefined

Run Fields

Run对象属性

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)]
所有
run
(Agent、命令或代码执行)都会返回一个
Run
对象:
python
run = box.exec.command("npm test")
run.id         # Run的ID
run.status     # 可选状态:"completed" | "failed" | ...
run.result     # 成功时返回stdout,失败时返回stderr(若指定response_schema则返回类型化结果)
run.stdout     # 原始标准输出(命令/代码执行场景)
run.stderr     # 原始标准错误(命令/代码执行场景)
run.exit_code  # 整数 | None(Agent运行场景下为None)
run.cost       # RunCost对象,包含input_tokens、output_tokens、cached_input_tokens、compute_ms、total_usd

run.cancel()          # 取消正在运行的任务
logs = run.logs()     # 返回[RunLog(timestamp, level, message)]列表

Box-level history

Box级别的历史记录

entries = box.logs(limit=100) # [LogEntry(timestamp, level, source, message)] runs = box.list_runs() # backend run records, newest first
undefined
entries = box.logs(limit=100) # 返回[LogEntry(timestamp, level, source, message)]列表 runs = box.list_runs() # 返回后端存储的Run记录,按时间倒序排列
undefined

Shell Execution

Shell执行

python
undefined
python
undefined

Run commands

运行命令

run = box.exec.command("echo hello && ls -la")
run = box.exec.command("echo hello && ls -la")

Run code snippets — lang: "js" | "ts" | "python"

运行代码片段——lang可选值:"js" | "ts" | "python"

run2 = box.exec.code(code="print(1 + 1)", lang="python", timeout=10_000)
run2 = box.exec.code(code="print(1 + 1)", lang="python", timeout=10_000)

Streaming shell / code

流式输出Shell/代码执行结果

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) ...
undefined
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) ...
undefined

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)]
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

二进制文件——读写时需指定encoding="base64"

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.write(path="/workspace/home/image.png", content=base64_string, encoding="base64") b64 = box.files.read("/workspace/home/image.png", encoding="base64")

Upload local files

上传本地文件

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

Download —
folder
is a path INSIDE the box; files land in ./<basename>

下载文件——
folder
为Box内部的路径;文件会下载到本地的./<basename>目录

box.files.download(folder="src") # → ./src box.files.download() # whole cwd → ./workspace
undefined
box.files.download(folder="src") # 下载到./src目录 box.files.download() # 下载当前工作目录到./workspace目录
undefined

cd / Working Directory

cd / 工作目录

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
SDK会在客户端跟踪
cwd
(当前工作目录)。所有操作(exec、files、git、agent)都会相对于该目录执行。
python
box.cwd  # 当前工作目录(初始为/workspace/home)
box.cd("my-repo")                  # 相对路径,基于当前cwd
box.cd("/workspace/home/other")    # 绝对路径

Git

Git操作

python
box.git.clone(repo="github.com/org/repo", branch="main")
box.git.clone(repo="github.com/org/repo", depth=1)  # shallow clone
box.cd("repo")  # cd into cloned 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")
python
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(  # 返回GitCommitResult(sha, message)
    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")

pr: PullRequest(url, number, title, base)

pr对象包含:url, number, title, base

Update the box-wide git identity

更新Box全局的Git身份信息

cfg = box.git.update_config(user_name="Bot", user_email="bot@example.com")
cfg = box.git.update_config(user_name="Bot", user_email="bot@example.com")

cfg: GitConfigResult(git_user_name, git_user_email)

cfg对象包含:git_user_name, git_user_email

Arbitrary git commands — returns the output string

执行任意Git命令——返回输出字符串

output = box.git.exec(args=["log", "--oneline", "-5"])
undefined
output = box.git.exec(args=["log", "--oneline", "-5"])
undefined

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",
    model="anthropic/claude-sonnet-5",  # optional override
    options={"max_budget_usd": 1.0, "effort": "high"},
    timeout=300_000,
)

schedules = box.schedule.list()
one = box.schedule.get(agent_schedule.id)
为Box配置Cron定时任务——支持Shell命令或Agent提示词。
Box
EphemeralBox
均支持该功能。Cron时间基于UTC时区。
python
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="运行测试套件并修复所有失败用例",
    model="anthropic/claude-sonnet-5",  # 可选,覆盖模型配置
    options={"max_budget_usd": 1.0, "effort": "high"},
    timeout=300_000,
)

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.

options=None会清空Agent选项。任务类型无法修改。

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)
undefined
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)
undefined

Snapshots

快照

python
undefined
python
undefined

Snapshot — checkpoint workspace state

创建快照——保存工作区状态

snap = box.snapshot(name="after-setup")
snap = box.snapshot(name="after-setup")

snap: Snapshot(id, name, box_id, size_bytes, status, created_at)

snap对象包含:id, name, box_id, size_bytes, status, created_at

restored = Box.from_snapshot(snap.id, size="medium", keep_alive=True) snaps = box.list_snapshots() box.delete_snapshot(snap.id)
undefined
restored = Box.from_snapshot(snap.id, size="medium", keep_alive=True) snaps = box.list_snapshots() box.delete_snapshot(snap.id)
undefined

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
/
run
are AI-powered and metered.
python
from pydantic import BaseModel

box = Box.create(browser=True, agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5})
创建Box时需指定
browser=True
,以驱动无头Chromium。标签页管理通过
box.browser
实现;所有页面操作通过标签句柄完成。
extract
/
observe
/
act
/
run
为AI驱动的功能,会产生计量费用。
python
from pydantic import BaseModel

box = Box.create(browser=True, agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5})

Tabs

标签页操作

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) # no network call
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) # 无需网络请求

Page operations

页面操作

content = tab.goto("https://news.ycombinator.com") # BrowserContent(title, url, text, links) current = tab.content() png = tab.screenshot() # bytes b64 = tab.screenshot(encoding="base64", full_page=True)
content = tab.goto("https://news.ycombinator.com") # 返回BrowserContent对象,包含title, url, text, links current = tab.content() png = tab.screenshot() # 返回字节数据 b64 = tab.screenshot(encoding="base64", full_page=True)

AI operations (metered) — schema is a Pydantic model or a raw JSON-schema dict

AI驱动操作(计量收费)——schema可为Pydantic模型或原始JSON Schema字典

class Story(BaseModel): title: str points: int
data = tab.extract("Top story title and points", Story) elements = tab.observe("What can I click?").elements acted = tab.act("Click the first headline") # BrowserActResult(success, message, actions, ...)
class Summary(BaseModel): summary: str
result = tab.run( "Find the top comment and summarize it", schema=Summary, max_steps=10, # default 15, max 30 model="anthropic/claude-sonnet-4-5", ) result.data, result.result, result.completed, result.steps
class Story(BaseModel): title: str points: int
data = tab.extract("获取顶部故事的标题和点赞数", Story) elements = tab.observe("我可以点击哪些元素?").elements acted = tab.act("点击第一个标题") # 返回BrowserActResult对象,包含success, message, actions等字段
class Summary(BaseModel): summary: str
result = tab.run( "找到顶部评论并总结内容", schema=Summary, max_steps=10, # 默认15,最大30 model="anthropic/claude-sonnet-4-5", ) result.data, result.result, result.completed, result.steps

Live view + raw CDP

实时视图 + 原始CDP协议

live_url = tab.live_view_url() # view-only screencast page/iframe cdp_url = box.browser.cdp_url() # Playwright / Puppeteer / Stagehand tab.close()
live_url = tab.live_view_url() # 仅查看的录屏页面/iframe cdp_url = box.browser.cdp_url() # 支持Playwright / Puppeteer / Stagehand tab.close()

Session recordings (HLS playback URL + MP4 download, chapter markers)

会话录制(提供HLS播放URL + MP4下载链接,包含章节标记)

handle = box.browser.recordings.start(max_duration_seconds=600) # default & max 600 recording = handle.stop()
handle = box.browser.recordings.start(max_duration_seconds=600) # 默认及最大值为600秒 recording = handle.stop()

BrowserRecording(id, box_id, status, started_at, ended_at, duration_ms, size_bytes,

BrowserRecording对象包含:id, box_id, status, started_at, ended_at, duration_ms, size_bytes,

mp4_size_bytes, segment_count, markers, stopped_reason, expires_at, playlist_url)

mp4_size_bytes, segment_count, markers, stopped_reason, expires_at, playlist_url

all_recordings = box.browser.recordings.list() one_recording = box.browser.recordings.get(recording.id)
all_recordings = box.browser.recordings.list() one_recording = box.browser.recordings.get(recording.id)

Download the video to a local file — returns the path written.

下载视频到本地文件——返回写入路径。

Defaults to ./box-recording-<id>.mp4 (.ts for recordings captured before MP4 support).

默认保存为./box-recording-<id>.mp4(MP4支持前捕获的录制文件为.ts格式)。

file = box.browser.recordings.download(recording.id) box.browser.recordings.download(recording.id, path="./out/demo.mp4")
undefined
file = box.browser.recordings.download(recording.id) box.browser.recordings.download(recording.id, path="./out/demo.mp4")
undefined

EphemeralBox

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.
python
from upstash_box import EphemeralBox

ebox = EphemeralBox.create(
    runtime="python",
    size="small",
    ttl=3600,  # seconds, max 259200 (3 days), default 259200
    env={"API_KEY": "..."},
    labels=["scratch"],  # settable at create time; filter via Box.list(label=...)
)

ebox.expires_at  # unix timestamp when auto-deleted
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.delete()
轻量级、短期存在的Box(最长3天)。支持
exec
files
schedule
cd
、网络策略和快照。不支持
agent
git
skills
、标签命名空间、浏览器或公共URL。
python
from upstash_box import EphemeralBox

ebox = EphemeralBox.create(
    runtime="python",
    size="small",
    ttl=3600,  # 存活时间(秒),最大259200(3天),默认259200
    env={"API_KEY": "..."},
    labels=["scratch"],  # 创建时可设置,可通过Box.list(label=...)筛选
)

ebox.expires_at  # 自动删除的Unix时间戳
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.delete()

Restore from snapshot

从快照恢复

ebox2 = EphemeralBox.from_snapshot(snap.id, ttl=7200)
undefined
ebox2 = EphemeralBox.from_snapshot(snap.id, ttl=7200)
undefined

Public URLs

公共URL

Expose box ports as public URLs with optional auth.
python
public_url = box.get_public_url(3000)
将Box端口暴露为公共URL,可选择添加认证。
python
public_url = box.get_public_url(3000)

public_url: PublicURL(url="https://{id}-3000.preview.box.upstash.com", port)

public_url对象包含:url="https://{id}-3000.preview.box.upstash.com", port

authed = box.get_public_url(3000, bearer_token=True)
authed = box.get_public_url(3000, bearer_token=True)

authed: PublicURL(url, port, token)

authed对象包含:url, port, token

basic = box.get_public_url(3000, basic_auth=True)
basic = box.get_public_url(3000, basic_auth=True)

basic: PublicURL(url, port, username, password)

basic对象包含:url, port, username, password

result = box.list_public_urls() # {"public_urls": [PublicURL, ...]} box.delete_public_url(3000)
undefined
result = box.list_public_urls() # 返回{"public_urls": [PublicURL, ...]} box.delete_public_url(3000)
undefined

Skills

Skills

Install agent skills from the Context7 registry. Format:
owner/repo/skill-name
.
python
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")
从Context7注册表安装Agent Skills。格式:
owner/repo/skill-name
python
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

标签

python
labels = box.labels.add("prod")  # returns the updated set
box.labels.remove("beta")
current = box.labels.list()
prod_boxes = Box.list(label="prod")
python
labels = box.labels.add("prod")  # 返回更新后的标签集合
box.labels.remove("beta")
current = box.labels.list()
prod_boxes = Box.list(label="prod")

Network Policy & Outbound Headers

网络策略与出站请求头

python
box = Box.create(
    # mode: "allow-all" (default) | "deny-all" | "custom"
    network_policy={
        "mode": "custom",
        "allowed_domains": ["api.example.com"],
        "denied_cidrs": ["10.0.0.0/8"],
    },
    # Inject secret headers into matching outbound HTTPS requests (write-only, never read back)
    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"})
python
box = Box.create(
    # mode可选值:"allow-all"(默认) | "deny-all" | "custom"
    network_policy={
        "mode": "custom",
        "allowed_domains": ["api.example.com"],
        "denied_cidrs": ["10.0.0.0/8"],
    },
    # 为匹配的出站HTTPS请求注入保密请求头(仅可写入,无法读取)
    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

MCP服务器

Attach MCP servers to the box agent.
python
box = Box.create(
    agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5},
    mcp_servers=[
        {"name": "fs", "package": "@modelcontextprotocol/server-filesystem"},
        {"name": "custom", "url": "https://mcp.example.com/sse", "headers": {"Authorization": "..."}},
    ],
)
为Box Agent附加MCP服务器。
python
box = Box.create(
    agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5},
    mcp_servers=[
        {"name": "fs", "package": "@modelcontextprotocol/server-filesystem"},
        {"name": "custom", "url": "https://mcp.example.com/sse", "headers": {"Authorization": "..."}},
    ],
)

Errors & SSH

错误处理与SSH

python
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):
bash
ssh <box-id>@us-east-1.box.upstash.com
python
from upstash_box import BoxError

try:
    box.agent.run(prompt="...")
except BoxError as e:
    print(e, e.status_code)
直接通过Shell连接Box(Box API密钥作为SSH密码):
bash
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.
python
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.
异步客户端与同步API完全一致——只需在调用前添加
await
,并使用
async for
处理流式输出。
python
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="搭建一个Next.js项目")
        print(run.result)

        stream = await box.agent.stream(prompt="构建一个REST API")
        async for chunk in stream:
            print(chunk)

        await box.delete()

asyncio.run(main())
通过
asyncio.gather
并行调用多个
AsyncBox.create(...)
/
box.agent.run(...)
,可实现Box的并行运行。

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
    .
  • get_init_command
    /
    set_init_command
    /
    delete_init_command
    raise unless the box was created with
    keep_alive=True
    .
  • The JS static
    Box.delete({boxIds})
    is
    Box.delete_boxes(box_ids=...)
    here, to avoid clashing with the instance
    delete()
    .
  • box.delete()
    is irreversible — snapshot first if you need the state.
  • Git operations require
    git.token
    in the box config for private repos and PRs.
  • Box.from_snapshot()
    creates a new box — it does not modify the original.
  • All
    timeout
    values are in milliseconds (matching the JS SDK), default
    600000
    .
  • When breaking out of a stream early, call
    stream.close()
    /
    await stream.aclose()
    so the run is marked
    detached
    .
  • Close the transport when done:
    box.delete()
    closes it, or use
    with box:
    /
    box.close()
    (
    async with
    /
    await box.aclose()
    for
    AsyncBox
    ).
  • 公开API的参数键在Python中为蛇形命名(snake_case)
    api_key
    user_name
    network_policy
    response_schema
    max_retries
    on_tool_use
    attach_headers
    ,以及Agent的
    options
    max_turns
    max_budget_usd
  • Agent配置需传入**
    harness
    **(而非已废弃的
    provider
    /
    runner
    )——
    harness
    为必填项。
  • response_schema
    接受Pydantic
    BaseModel
    子类(返回类型化实例)或原始JSON Schema字典(返回
    dict
    )。浏览器的
    schema
    遵循相同规则。
  • 默认工作目录为
    /workspace/home
    ,而非
    /home
    /
  • box.cd()
    为客户端侧跟踪——会验证路径是否存在,但不会改变Box的Shell工作目录。所有SDK方法会自动使用该目录。
  • EphemeralBox
    不支持
    agent
    git
    skills
    、标签命名空间、浏览器或公共URL——如需这些功能请使用完整的
    Box
    EphemeralBox
    支持
    schedule
    和快照)。
  • run.exit_code
    在Agent运行场景下为
    None
    ,仅在命令执行场景下可用。
  • run.result
    在成功时返回stdout,失败时返回stderr——若命令以0退出但仅向stderr输出内容,
    run.result
    会返回
    ""
    ;需读取
    run.stderr
    获取对应内容。
  • files.download(folder=...)
    接受Box内部的路径;输出文件会保存到本地的
    ./<basename>
    目录。
  • box.browser
    要求创建Box时指定
    browser=True
  • get_init_command
    /
    set_init_command
    /
    delete_init_command
    仅在Box以
    keep_alive=True
    创建时可用,否则会报错。
  • JS中的静态方法
    Box.delete({boxIds})
    对应Python中的
    Box.delete_boxes(box_ids=...)
    ,以避免与实例方法
    delete()
    冲突。
  • box.delete()
    操作不可逆——如需保留状态请先创建快照。
  • Git操作访问私有仓库和创建PR时,需在Box配置中传入
    git.token
  • Box.from_snapshot()
    会创建新的Box——不会修改原始Box。
  • 所有
    timeout
    值的单位为毫秒(与JS SDK一致),默认值为
    600000
  • 提前终止流式输出时,请调用
    stream.close()
    /
    await stream.aclose()
    ,以便将任务标记为
    detached
  • 使用完毕后请关闭连接:
    box.delete()
    会关闭连接,也可使用
    with box:
    /
    box.close()
    (异步场景下使用
    async with
    /
    await box.aclose()
    )。