pip install upstash-box

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

# Create with agent + git + env vars
box = Box.create(
    runtime="node",  # "node" | "python" | "golang" | "ruby" | "rust"
    agent={
        "harness": Agent.CLAUDE_CODE,  # Agent.CODEX | Agent.OPEN_CODE
        "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"],  # GitHub repos as agent skills
)

# Reconnect, list, delete, pause/resume
same = Box.get(box.id)
all_boxes = Box.list()
box.pause()
box.resume()
box.delete()  # irreversible
status = box.get_status()["status"]

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,
    on_tool_use=lambda tool: print(tool["name"], tool["input"]),
)

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

# Streaming
stream = box.agent.stream(prompt="Build a REST API")
for chunk in stream:
    print(chunk)

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

run = box.exec.command("npm test")
run.id         # run ID
run.status     # "completed" | "failed" | ...
run.result     # string output (or typed result with response_schema)
run.exit_code  # int | None (None for agent runs)
run.cost       # RunCost(input_tokens, output_tokens, compute_ms, total_usd)

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

# 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
stream = box.exec.stream("npm run build")
for chunk in stream:
    # chunk: ExecOutputChunk(type="output", data) | ExecExitChunk(type="exit", exit_code, cpu_ns)
    ...

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

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

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

box.git.clone(repo="github.com/org/repo", branch="main")
box.cd("repo")  # cd into cloned repo

status = box.git.status()
diff = box.git.diff()
result = box.git.commit(message="fix: resolve bug")  # GitCommitResult(sha, message)
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)

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

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

restored = Box.from_snapshot(snap.id)
snaps = box.list_snapshots()
box.delete_snapshot(snap.id)

from upstash_box import EphemeralBox

ebox = EphemeralBox.create(
    runtime="python",
    ttl=3600,  # seconds, max 259200 (3 days)
    env={"API_KEY": "..."},
)

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.cd("subdir")
ebox.delete()

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

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)

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

import asyncio
from upstash_box import AsyncBox, Agent

async def main():
    box = await AsyncBox.create(runtime="node", agent={"harness": Agent.CLAUDE_CODE})
    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())