upstash-box-py

upstash-box Python SDK

带内置 AI Agent、Shell、文件系统、Git、Cron 计划任务以及可选无头浏览器的沙盒云容器。

与 @upstash/box TypeScript SDK（upstash-box-js skill）一致，使用 snake_case 名称；有意保留的差异列在「注意事项」下。

安装与设置

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 生命周期

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

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

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

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

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

# Bulk delete (classmethods, by ID). An empty or blank id list raises BoxError —
# it is never read as "everything". Omit snapshot_ids entirely to delete all.
Box.delete_boxes(box_ids=["box_1", "box_2"])  # JS static `delete` is `delete_boxes` here
Box.delete_snapshots(snapshot_ids=["snap_1"])
Box.delete_snapshots()  # delete all

账户级环境变量

注入到你创建的每个 box 中。

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

Agent 运行

from pydantic import BaseModel

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

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

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

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

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

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

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

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

Agent 选项（按执行框架）

options 会转发给执行框架；接受的键取决于 box 运行的执行框架。 Python 中的键使用 snake_case；SDK 会将顶层键转换为各执行框架后端所用的大小写格式（Claude Code / OpenCode 使用 camelCase，Codex 使用 snake_case）。嵌套字典中的键会原样发送。

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

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

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

# Agent.CURSOR → free-form dict

与 JS 通用的 AgentOptions<TProvider> 不同，Python 不会根据执行框架缩小 options 的类型范围；类型是所有形状的联合类型加上原始字典。

执行框架与模型

harness 是必填项。模型枚举：ClaudeCode、OpenAICodex、OpenCodeModel、 CursorModel、OpenRouterModel、VercelModel，也可以使用任何带提供方前缀的字符串。

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

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

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

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

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

自定义执行框架

在 box 内运行你自己的 Agent 进程，而不是使用托管执行框架。

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

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

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

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

运行字段

每个 run（Agent、命令或代码）都会返回一个 Run：

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

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

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

Shell 执行

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

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

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

实时会话

exec.session() 会通过 WebSocket 连接到仍在运行的进程，支持 stdin、 流式 stdout/stderr、PTY 调整大小和信号。这是唯一未通过 httpx 提供的功能；websockets 依赖会延迟导入，仅在会话 开启时导入。Box、AsyncBox 以及两个临时客户端均支持此功能。

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

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

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

会话拥有该进程：调用 close()（或离开 with 代码块）、连接断开 或程序退出，都会终止命令；会话也无法重新连接。使用 wait() 运行任务直至完成。

在 AsyncBox 中，每个句柄方法都是协程（await session.write(...)、 await session.wait()、async with await box.exec.session(...) as s:），并且异步 回调会被等待；其中的 wait() 不接受超时参数。在同步客户端中，回调会在后台读取器 线程上运行——请让回调保持简短，并且绝不要在回调中调用 wait()，因为它等待的退出 帧会在它正在阻塞的同一线程上到达。

文件系统

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

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

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

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

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

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

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

cd / 工作目录

SDK 在客户端跟踪 cwd。所有操作（exec、文件、git、agent）都相对于该目录运行。

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

Git

克隆的内容会放在 box 的隔离容器中，不会放在调用方的机器上。克隆的 代码在运行之前都只是数据——请将不受信任的仓库视为不受信任的输入，并在运行其构建或测试前 配合限制严格的 network_policy（见下文）。

除 clone 外，每个 git 调用都会在 box 的当前目录中运行，因此请先 cd 到 克隆目录。在工作区根目录中没有仓库，因此 status 会返回空结果，表示工作区干净。

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

status = box.git.status()
diff = box.git.diff()
result = box.git.commit(  # GitCommitResult(sha, message)
    message="fix: resolve bug",
    author_name="Jane Doe",  # optional per-commit override
    author_email="[email protected]",
)
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, warning)
issue = box.git.create_issue(title="Flaky test in CI", body="...")
# issue: Issue(url, number, title, warning)

# Attach images or videos to a PR or issue. Paths are relative to the box's
# working directory; alt text for an image is written as "shot.png#alt text"
# (a video takes none). Reference an attachment from the body as
# ![alt](./shot.png) and GitHub rewrites it to the uploaded asset.
box.git.create_pr(
    title="Fix the tower layout",
    body="![before](./before.png)\n![after](./after.png)",
    attach=["before.png#before", "after.png#after"],
)
# `warning` is set when `gh` exited non-zero but still returned a URL: the item
# exists while an attachment failed to upload, or the PR was already open.

# Update the box-wide git identity
cfg = box.git.update_config(user_name="Bot", user_email="[email protected]")
# cfg: GitConfigResult(git_user_name, git_user_email)

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

计划任务

box 上的 Cron 任务——Shell 命令或 Agent 提示词。Box 和 EphemeralBox 均支持。Cron 使用 UTC。

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

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

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

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

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

快照

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

# from_snapshot takes the same BoxConfig kwargs as create (shared request body):
# name, labels, size, keep_alive, init_command, runtime, browser, agent, git, env,
# attach_headers, network_policy, skills, mcp_servers. Note the JS SDK's
# Box.fromSnapshot() drops browser / skills / mcpServers — Python forwards them.
restored = Box.from_snapshot(
    snap.id,
    size="medium",
    keep_alive=True,
    # the git identity is forwarded, not just the token
    git={"token": os.environ["GITHUB_TOKEN"], "user_name": "Bot", "user_email": "[email protected]"},
    env={"DATABASE_URL": "..."},
)
snaps = box.list_snapshots()
box.delete_snapshot(snap.id)

浏览器

创建 box 时设置 browser=True，即可驱动无头 Chromium。标签页管理位于 box.browser 上；每个页面操作都通过标签页句柄执行。 extract / observe / act(instruction) 由 AI 驱动并按量计费； act(action) 会重放已解析的操作，不调用 LLM，也不消耗 token。

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.id, tab.url, tab.title  # handle metadata, no network call

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

# AI operations (metered) — schema is a Pydantic model or a raw JSON-schema dict.
# extract / observe / act take an optional model= override, defaulting to the box's
# model (or anthropic/claude-sonnet-4-5 when it has none).
class Story(BaseModel):
    title: str
    points: int

data = tab.extract("Top story title and points", Story, model="anthropic/claude-sonnet-4-5")

# observe → actionable elements, each carrying a replayable method + arguments
elements = tab.observe("What can I click?", model="openai/gpt-5.6").elements
# elements: [BrowserObserveElement(description, selector, url, method, arguments)]

acted = tab.act("Click the first headline")
# BrowserActResult(success, message, action_description, actions, cache_status,
#                  input_tokens, output_tokens)

# Replay a pre-resolved action — no LLM call, no tokens, no model provider key.
# Pass a BrowserObserveElement or BrowserActAction instead of a string; `model` is
# ignored in this form, and an action without a `selector` raises BoxError.
tab.act(elements[0])
tab.act(acted.actions[0])

# Live view + raw CDP
live_url = tab.live_view_url()  # view-only screencast page/iframe
cdp_url = box.browser.cdp_url()  # wss://…?token=… — no extra auth wiring
tab.close()

# Drive the same browser from Playwright (pip install playwright)
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    remote = p.chromium.connect_over_cdp(cdp_url)
    context = remote.contexts[0] if remote.contexts else remote.new_context()
    page = context.pages[0] if context.pages else context.new_page()
    page.goto("https://example.com")

# Session recordings (HLS playback URL + MP4 download, chapter markers).
# One active recording per box; captures all tabs and follows the foreground.
# Auto-stops after max_duration_seconds or ~3 minutes of no on-screen activity.
handle = box.browser.recordings.start(max_duration_seconds=600)  # default & max 600
recording = handle.stop()
# or stop whatever is recording on the box, without a handle:
# recording = box.browser.recordings.stop()
# BrowserRecording(id, box_id, status, started_at, ended_at, duration_ms, size_bytes,
#                  mp4_size_bytes, segment_count, markers, stopped_reason,
#                  max_duration_seconds, expires_at, playlist_url)
# markers: BrowserRecordingMarker(type="tab_switch", at_ms, end_ms, label, tab_id)
# expires_at is epoch ms (videos retained 14 days); playlist_url is API-served — fetch it
# with an `X-Box-Api-Key: <api_key>` header (hls.js / Safari / ffplay).
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).
file = box.browser.recordings.download(recording.id)
box.browser.recordings.download(recording.id, path="./out/demo.mp4")

多步骤浏览器目标

tab.run() — 自主多步骤浏览器 Agent — 已被移除，BrowserRunResult / BrowserRunStep 类型也一并移除（Stagehand v4 移除了底层的 Agent 基元）。浏览器现在仅提供 observe、act 和 extract。有三种替代方案：

1. 自行驱动循环 — 用 observe 一次性解析步骤，然后用 act(action) 重放；这样模型就不会进入关键路径；extract 用于检查是否停止。

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)   # deterministic reset, no browser-AI tokens
    tab.act(action)   # replay the resolved click: no LLM, no tokens
    item = tab.extract("title and price", Product)

2. 将目标交给内置 Agent — browser=True 会自动将 chrome-devtools MCP（Chromium 已在 127.0.0.1:9222 上预热）接入 box 的编码 Agent，因此 box.agent.run(prompt=...) 会自行驱动浏览器并反复执行，直到完成。无需先调用 tab.create()。此方式会计入编码 Agent 模型的 token 用量，而非浏览器 AI 计量，并且需要 Agent 框架和密钥。

3. 通过 CDP 连接 — 当整个流程完全确定时，使用 Playwright 和 box.browser.cdp_url()。

EphemeralBox

轻量级短期 box（最长 3 天）。支持 exec、files、schedule、cd、网络策略和快照。不支持 agent、git、skills、labels 命名空间、浏览器或公共 URL。

from upstash_box import EphemeralBox

ebox = EphemeralBox.create(
    name="scratch-box",
    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=...)
    network_policy={"mode": "deny-all"},
    attach_headers={"api.stripe.com": {"Authorization": "Bearer sk_live_..."}},
)

ebox.network_policy

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.exec.session(argv=["bash", "-i"], tty=True)  # whole exec namespace, session included
ebox.files.write(path="/workspace/home/data.json", content="{}")
ebox.files.stat("/workspace/home/data.json")  # whole files namespace, stat/mkdir/rename/remove included
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()

# Restore from snapshot
ebox2 = EphemeralBox.from_snapshot(snap.id, ttl=7200)

# Statics: EphemeralBox.delete_boxes(box_ids=[...]) / EphemeralBox.delete_snapshots(...)
# are the Box ones. EphemeralBox.get_by_name() returns a full `Box`, not an
# `EphemeralBox` (quirk mirrored from the JS SDK).
# `AsyncEphemeralBox` is the async variant (`await AsyncEphemeralBox.create(...)`).

公共 URL

将 box 端口公开为公共 URL，并可选择启用身份验证。

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

authed = box.get_public_url(3000, bearer_token=True)
# authed: PublicURL(url, port, token)

basic = box.get_public_url(3000, basic_auth=True)
# basic: PublicURL(url, port, username, password)

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

技能

从 Context7 注册表安装 Agent 技能。格式：owner/repo/skill-name。

已安装的技能会成为 box 中 Agent 的指令，因此应像对待依赖项一样，只固定使用你信任的所有者提供的技能。技能会在创建 box 时从注册表解析，而不是从任意 URL 获取，并且只会在 box 容器内运行。

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 = box.labels.add("prod")  # returns the updated set
box.labels.remove("beta")
current = box.labels.list()
prod_boxes = Box.list(label="prod")

网络策略与出站标头

box = Box.create(
    # mode: "allow-all" (default) | "deny-all" | "custom"
    # custom takes any of allowed_domains / allowed_cidrs / denied_cidrs
    network_policy={
        "mode": "custom",
        "allowed_domains": ["api.example.com"],
        "allowed_cidrs": ["203.0.113.0/24"],
        "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"})

MCP 服务

将 MCP 服务附加到 box Agent。附加的服务会提供 Agent 可调用的工具，因此请使用你控制或信任的服务；如果 Agent 还会处理不可信输入，请保持 network_policy 的限制性。

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": "..."}},
    ],
)

错误与 SSH

from upstash_box import BoxError

try:
    box.agent.run(prompt="...")
except BoxError as e:
    print(e, e.status_code)

直接通过 Shell 连接到 box（Box API 密钥即 SSH 密码）：

ssh <box-id>@us-east-1.box.upstash.com

异步客户端

异步客户端与同步 API 完全一致——对调用使用 await，并使用 async for 进行流式处理。

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

对多个 AsyncBox.create(...) / box.agent.run(...) 调用使用 asyncio.gather，即可并行运行多个 box。

注意事项

• Python 中的公共 API 选项键使用 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（返回 dict）。Browser 的 schema 遵循相同约定。
• 默认工作目录是 /workspace/home，而非 /home 或 /。
• box.cd() 仅在客户端跟踪目录——它会验证路径是否存在，但不会更改 Box 的 shell 当前工作目录。所有 SDK 方法都会自动使用它。
• EphemeralBox 不支持 agent、git、skills、labels 命名空间、浏览器或公共 URL——这些功能请使用完整的 Box（但它支持 schedule 和快照）。
• 对于 Agent 运行，run.exit_code 为 None；该值仅适用于 exec 命令。
• 成功时，run.result 是标准输出；失败时则是标准错误——退出码为 0、且只向标准错误写入内容的命令会返回 ""；请读取 run.stderr 获取该内容。
• files.download(folder=...) 接受 Box 内部的路径；输出会保存到本地的 ./<basename>。
• 只有在指定 length 时，files.read() 才会切片读取——单独指定 offset= 会读取整个文件，而 length=0 则不读取任何内容。
• 默认情况下，files.stat() 执行的是 lstat：符号链接会报告为 type="symlink"，除非传入 follow=True。
• 对目录调用 files.remove() 需要设置 recursive=True；对嵌套路径调用 files.mkdir() 需要设置 parents=True。
• files.rename(from_path, to_path) 接受位置参数（from 是 Python 关键字）；JS 中写作 rename(from, to)。
• exec.session() 返回的句柄会管理进程——调用 close() 或连接断开都会终止命令，且无法重新连接到会话。tty=True 会将标准错误合并到标准输出，因此 PTY 会话不会触发 on_stderr。
• 同步方法 session.wait(timeout=...) 没有对应的异步方法（await handle.wait() 会阻塞到进程退出）；超时后会抛出 TimeoutError。
• box.browser 需要通过 browser=True 创建 Box。
• 没有 tab.run()——自主浏览器 Agent 已移除。请自行循环调用 observe + act(action) + extract，将目标交给 Box 内的 Agent，或通过 cdp_url() 驱动 Playwright。
• tab.act(action)（重放 observe() 的结果）不消耗 Token，也不需要模型提供商密钥；只有传入字符串的 act(instruction) 才会计量。
• 除非 Box 创建时设置了 keep_alive=True，否则调用 get_init_command / set_init_command / delete_init_command 会抛出异常。
• JS 静态方法 Box.delete({boxIds}) 在这里对应 Box.delete_boxes(box_ids=...)，以避免与实例方法 delete() 冲突。
• Box.delete_boxes(box_ids=[]) 和 Box.delete_snapshots(snapshot_ids=[]) 会抛出 BoxError——空列表绝不会被视为“删除全部”。如需删除全部快照，请调用不带 ID 的 Box.delete_snapshots()。
• box.delete() 不可撤销——如果需要保留状态，请先创建快照。
• 对私有仓库和 PR 执行 Git 操作时，Box 配置中必须包含 git.token。
• Box.from_snapshot() 会创建新的 Box，不会修改原 Box。它会复用完整的创建请求体，因此会转发 browser / skills / mcp_servers（JS 的 Box.fromSnapshot() 不会转发这些参数）。
• EphemeralBox 没有 update_network_policy——请在创建时设置 network_policy。
• 所有 timeout 值均以毫秒为单位（与 JS SDK 一致），默认值为 600000。
• 提前退出流时，请调用 stream.close() / await stream.aclose()，以便将运行标记为 detached。
• 使用完毕后请关闭传输：box.delete() 会将其关闭，也可以使用 with box: / box.close()（对于 AsyncBox，使用 async with / await box.aclose()）。