upstash-box-py
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
Chineseupstash-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-boxSet env var or pass to constructors.
UPSTASH_BOX_API_KEYapi_keyThe SDK ships both a synchronous (used in the examples below) and an
asynchronous (, ).
The async surface is identical with and .
BoxAsyncBoxbox = await AsyncBox.create(...)await box.agent.run(...)awaitasync forAnonymous telemetry headers are sent with every request; opt out with the
env var.
UPSTASH_DISABLE_TELEMETRYbash
pip install upstash-box设置环境变量,或在构造函数中传入参数。
UPSTASH_BOX_API_KEYapi_keySDK同时提供同步版本(以下示例使用该版本)和异步版本(使用方式为、)。异步接口与同步接口完全一致,仅需添加和关键字。
BoxAsyncBoxbox = await AsyncBox.create(...)await box.agent.run(...)awaitasync for每次请求会发送匿名遥测头信息;可通过设置环境变量选择退出。
UPSTASH_DISABLE_TELEMETRYBox Lifecycle
Box生命周期
python
import os
from upstash_box import Box, Agent, ClaudeCode, BoxApiKeypython
import os
from upstash_box import Box, Agent, ClaudeCode, BoxApiKeyCreate 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 is here
Box.delete_snapshots(snapshot_ids=["snap_1"]) # omit ids → delete all
deletedelete_boxesundefinedBox.delete_boxes(box_ids=["box_1", "box_2"]) # JS中的静态方法对应Python中的
Box.delete_snapshots(snapshot_ids=["snap_1"]) # 省略ids参数会删除所有快照
deletedelete_boxesundefinedAccount-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 BaseModelpython
from pydantic import BaseModelStructured 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
.type流式输出——chunk为类型化数据类,通过.type
区分类型
.typestream = 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 ..."}},
)
undefinedbox.agent.run(
prompt="运行测试",
webhook={"url": "https://example.com/hook", "headers": {"Authorization": "Bearer ..."}},
)
undefinedHarness & model
Harness与模型
harnessClaudeCodeOpenAICodexOpenCodeModelCursorModelOpenRouterModelVercelModelpython
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"harnessClaudeCodeOpenAICodexOpenCodeModelCursorModelOpenRouterModelVercelModelpython
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")
undefinedbox.model_config # 返回{"harness": ..., "model": ...}
box.configure_model("anthropic/claude-opus-4-8")
undefinedCustom 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
undefinedasync 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为异步函数
undefinedRun Fields
Run对象属性
Every (agent, command, or code) returns a :
runRunpython
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)]所有(Agent、命令或代码执行)都会返回一个对象:
runRunpython
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
undefinedentries = box.logs(limit=100) # 返回[LogEntry(timestamp, level, source, message)]列表
runs = box.list_runs() # 返回后端存储的Run记录,按时间倒序排列
undefinedShell Execution
Shell执行
python
undefinedpython
undefinedRun 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)
...
undefinedstream = 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)
...
undefinedFilesystem
文件系统
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下载文件——folder
为Box内部的路径;文件会下载到本地的./<basename>目录
folderbox.files.download(folder="src") # → ./src
box.files.download() # whole cwd → ./workspace
undefinedbox.files.download(folder="src") # 下载到./src目录
box.files.download() # 下载当前工作目录到./workspace目录
undefinedcd / Working Directory
cd / 工作目录
The SDK tracks client-side. All operations (exec, files, git, agent) run relative to it.
cwdpython
box.cwd # current working directory (starts at /workspace/home)
box.cd("my-repo") # relative to current cwd
box.cd("/workspace/home/other") # absolute pathSDK会在客户端跟踪(当前工作目录)。所有操作(exec、files、git、agent)都会相对于该目录执行。
cwdpython
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"])
undefinedoutput = box.git.exec(args=["log", "--oneline", "-5"])
undefinedSchedules
定时任务
Cron tasks on a box — shell commands or agent prompts. Available on and . Cron is UTC.
BoxEphemeralBoxpython
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提示词。和均支持该功能。Cron时间基于UTC时区。
BoxEphemeralBoxpython
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)
undefinedbox.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)
undefinedSnapshots
快照
python
undefinedpython
undefinedSnapshot — 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)
undefinedrestored = Box.from_snapshot(snap.id, size="medium", keep_alive=True)
snaps = box.list_snapshots()
box.delete_snapshot(snap.id)
undefinedBrowser
浏览器操作
Create the box with to drive a headless Chromium. Tab management
lives on ; every page operation lives on the tab handle.
/ / / are AI-powered and metered.
browser=Truebox.browserextractobserveactrunpython
from pydantic import BaseModel
box = Box.create(browser=True, agent={"harness": Agent.CLAUDE_CODE, "model": ClaudeCode.SONNET_4_5})创建Box时需指定,以驱动无头Chromium。标签页管理通过实现;所有页面操作通过标签句柄完成。
/ / / 为AI驱动的功能,会产生计量费用。
browser=Truebox.browserextractobserveactrunpython
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")
undefinedfile = box.browser.recordings.download(recording.id)
box.browser.recordings.download(recording.id, path="./out/demo.mp4")
undefinedEphemeralBox
EphemeralBox
Lightweight, short-lived boxes (max 3 days). Supports , , ,
, network policy, and snapshots. No , , ,
namespace, browser, or public URLs.
execfilesschedulecdagentgitskillslabelspython
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天)。支持、、、、网络策略和快照。不支持、、、标签命名空间、浏览器或公共URL。
execfilesschedulecdagentgitskillspython
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)
undefinedebox2 = EphemeralBox.from_snapshot(snap.id, ttl=7200)
undefinedPublic 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)
undefinedresult = box.list_public_urls() # 返回{"public_urls": [PublicURL, ...]}
box.delete_public_url(3000)
undefinedSkills
Skills
Install agent skills from the Context7 registry. Format: .
owner/repo/skill-namepython
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-namepython
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.compython
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.comAsync client
异步客户端
The async client mirrors the sync API exactly — the calls and use to stream.
awaitasync forpython
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.gatherAsyncBox.create(...)box.agent.run(...)异步客户端与同步API完全一致——只需在调用前添加,并使用处理流式输出。
awaitasync forpython
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())通过并行调用多个 / ,可实现Box的并行运行。
asyncio.gatherAsyncBox.create(...)box.agent.run(...)Gotchas
注意事项
- Public API option keys are snake_case in Python: ,
api_key,user_name,network_policy,response_schema,max_retries,on_tool_use, and agentattach_headerslikeoptions,max_turns.max_budget_usd - Agent config takes (not the deprecated
harness/provider) —runneris required.harness - accepts a Pydantic
response_schemasubclass (returns a typed instance) or a raw JSON-schemaBaseModel(returns adict). Browserdictfollows the same contract.schema - Default working directory is , not
/workspace/homeor/home./ - is client-side tracking — it validates the path exists but doesn't change the box's shell cwd. All SDK methods use it automatically.
box.cd() - does NOT support
EphemeralBox,agent,git, theskillsnamespace, the browser, or public URLs — use fulllabelsfor those (it does supportBoxand snapshots).schedule - is
run.exit_codefor agent runs, only available for exec commands.None - is stdout on success and stderr on failure — a command that exits 0 writing only to stderr yields
run.result; read""for it.run.stderr - takes a path inside the box; output lands in
files.download(folder=...)locally../<basename> - requires a box created with
box.browser.browser=True - /
get_init_command/set_init_commandraise unless the box was created withdelete_init_command.keep_alive=True - The JS static is
Box.delete({boxIds})here, to avoid clashing with the instanceBox.delete_boxes(box_ids=...).delete() - is irreversible — snapshot first if you need the state.
box.delete() - Git operations require in the box config for private repos and PRs.
git.token - creates a new box — it does not modify the original.
Box.from_snapshot() - All values are in milliseconds (matching the JS SDK), default
timeout.600000 - When breaking out of a stream early, call /
stream.close()so the run is markedawait stream.aclose().detached - Close the transport when done: closes it, or use
box.delete()/with box:(box.close()/async withforawait box.aclose()).AsyncBox
- 公开API的参数键在Python中为蛇形命名(snake_case):、
api_key、user_name、network_policy、response_schema、max_retries、on_tool_use,以及Agent的attach_headers如options、max_turns。max_budget_usd - Agent配置需传入****(而非已废弃的
harness/provider)——runner为必填项。harness - 接受Pydantic
response_schema子类(返回类型化实例)或原始JSON Schema字典(返回BaseModel)。浏览器的dict遵循相同规则。schema - 默认工作目录为,而非
/workspace/home或/home。/ - 为客户端侧跟踪——会验证路径是否存在,但不会改变Box的Shell工作目录。所有SDK方法会自动使用该目录。
box.cd() - 不支持
EphemeralBox、agent、git、标签命名空间、浏览器或公共URL——如需这些功能请使用完整的skills(Box支持EphemeralBox和快照)。schedule - 在Agent运行场景下为
run.exit_code,仅在命令执行场景下可用。None - 在成功时返回stdout,失败时返回stderr——若命令以0退出但仅向stderr输出内容,
run.result会返回run.result;需读取""获取对应内容。run.stderr - 接受Box内部的路径;输出文件会保存到本地的
files.download(folder=...)目录。./<basename> - 要求创建Box时指定
box.browser。browser=True - /
get_init_command/set_init_command仅在Box以delete_init_command创建时可用,否则会报错。keep_alive=True - JS中的静态方法对应Python中的
Box.delete({boxIds}),以避免与实例方法Box.delete_boxes(box_ids=...)冲突。delete() - 操作不可逆——如需保留状态请先创建快照。
box.delete() - Git操作访问私有仓库和创建PR时,需在Box配置中传入。
git.token - 会创建新的Box——不会修改原始Box。
Box.from_snapshot() - 所有值的单位为毫秒(与JS SDK一致),默认值为
timeout。600000 - 提前终止流式输出时,请调用/
stream.close(),以便将任务标记为await stream.aclose()。detached - 使用完毕后请关闭连接:会关闭连接,也可使用
box.delete()/with box:(异步场景下使用box.close()/async with)。await box.aclose()