Agent
For every box, you can configure a built-in agent like Claude Code, Codex or OpenCode. It has access to the filesystem, git, and shell commands and should simulate running an Agent on your own computer.
You can choose between:
run()when you want to wait for completion and then read the final typed result.stream()when you want real-time output while the agent is running.
Configure an Agent
Get your Claude API key from the Claude Console.
UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxxANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxximport { Agent, Box } from "@upstash/box"const box = await Box.create({ runtime: "node", agent: { harness: Agent.ClaudeCode, model: "anthropic/claude-sonnet-4-5", apiKey: process.env.ANTHROPIC_API_KEY!, },})import osfrom upstash_box import Box, Agentbox = Box.create( runtime="node", agent={ "harness": Agent.CLAUDE_CODE, "model": "anthropic/claude-sonnet-4-5", "api_key": os.environ["ANTHROPIC_API_KEY"], },)UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxxOPENCODE_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxximport { Agent, Box } from "@upstash/box"const box = await Box.create({ runtime: "node", agent: { harness: Agent.OpenCode, model: "opencode/claude-sonnet-4-6", apiKey: process.env.OPENCODE_API_KEY!, },})import osfrom upstash_box import Box, Agentbox = Box.create( runtime="node", agent={ "harness": Agent.OPEN_CODE, "model": "opencode/claude-sonnet-4-6", "api_key": os.environ["OPENCODE_API_KEY"], },)UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxxOPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxximport { Agent, Box } from "@upstash/box"const box = await Box.create({ runtime: "node", agent: { harness: Agent.Codex, model: "openai/gpt-5.3-codex", apiKey: process.env.OPENAI_API_KEY!, },})import osfrom upstash_box import Box, Agentbox = Box.create( runtime="node", agent={ "harness": Agent.CODEX, "model": "openai/gpt-5.3-codex", "api_key": os.environ["OPENAI_API_KEY"], },)Agent Options
You can pass provider-specific agent options with the options field on box.agent.run() and box.agent.stream().
const run = await box.agent.run({ prompt: "Refactor the auth flow and keep changes minimal", options: { effort: "medium", maxTurns: 12, },})run = box.agent.run( prompt="Refactor the auth flow and keep changes minimal", options={ "effort": "medium", "max_turns": 12, },)The exact option shape depends on the configured agent:
ClaudeCode:maxTurns,maxBudgetUsd,effort,thinking,disallowedTools,agents,promptSuggestions,fallbackModel,systemPromptCodex:modelReasoningEffort,modelReasoningSummary,personality,webSearchOpenCode:reasoningEffort,textVerbosity,reasoningSummary,thinking
In the Python SDK these option keys are snake_case — e.g. max_turns, max_budget_usd, model_reasoning_effort, web_search, reasoning_effort.
To bring your own agent process, use a custom agent.
Quickstart
Codebase refactor
Use Claude Code when you want an agent to work through a larger code change with filesystem, shell, and git access.
const stream = await box.agent.stream({ prompt: `Refactor the payment module to:- move shared validation into src/payment/validation.ts- keep the public API unchanged- update tests if needed- summarize the main risks before finishing `,})for await (const chunk of stream) { if (chunk.type === "text-delta") process.stdout.write(chunk.text)}console.log(stream.status)console.log(stream.result)stream = box.agent.stream( prompt="""Refactor the payment module to:- move shared validation into src/payment/validation.ts- keep the public API unchanged- update tests if needed- summarize the main risks before finishing""",)for chunk in stream: if chunk.type == "text-delta": print(chunk.text, end="")print(stream.status)print(stream.result)Fast code review summary
Use OpenCode when you want a concise review or summary over an existing diff or repository state.
const run = await box.agent.run({ prompt: `Review the current git diff and return:- the top 3 risks- missing test coverage- whether the change looks safe to merge `,})console.log(run.result)console.log(run.cost.totalUsd)run = box.agent.run( prompt="""Review the current git diff and return:- the top 3 risks- missing test coverage- whether the change looks safe to merge""",)print(run.result)print(run.cost.total_usd)Structured analysis
Use Codex when you want a strongly structured response that you can feed into another system.
import { z } from "zod"const result = await box.agent.run({ prompt: "Analyze /work/report.csv and return the top 10 customers by revenue", responseSchema: z.object({ customers: z.array( z.object({ name: z.string(), revenue: z.number(), }), ), }),})console.log(result.result.customers)from pydantic import BaseModelclass Customer(BaseModel): name: str revenue: floatclass Result(BaseModel): customers: list[Customer]result = box.agent.run( prompt="Analyze /work/report.csv and return the top 10 customers by revenue", response_schema=Result,)print(result.result.customers)Clone a repository, run the agent, and open a pull request.
import { Agent, Box } from "@upstash/box"const box = await Box.create({ runtime: "node", agent: { harness: Agent.ClaudeCode, model: "anthropic/claude-opus-4-5", apiKey: process.env.ANTHROPIC_API_KEY, }, git: { token: process.env.GITHUB_TOKEN, },})await box.git.clone({ repo: "github.com/your-org/your-repo" })const stream = await box.agent.stream({ prompt: "Fix the null token bug in src/auth.ts and add tests",})for await (const chunk of stream) { if (chunk.type === "text-delta") process.stdout.write(chunk.text)}await box.git.createPR({ title: "Fix null token bug", base: "main",})import osfrom upstash_box import Box, Agentbox = Box.create( runtime="node", agent={ "harness": Agent.CLAUDE_CODE, "model": "anthropic/claude-opus-4-5", "api_key": os.environ["ANTHROPIC_API_KEY"], }, git={"token": os.environ["GITHUB_TOKEN"]},)box.git.clone(repo="github.com/your-org/your-repo")stream = box.agent.stream(prompt="Fix the null token bug in src/auth.ts and add tests")for chunk in stream: if chunk.type == "text-delta": print(chunk.text, end="")box.git.create_pr(title="Fix null token bug", base="main")Run one box per file, process in parallel, and return ranked structured results.
import { Agent, Box } from "@upstash/box"import { readdir } from "fs/promises"import { z } from "zod"const responseSchema = z.object({ name: z.string(), email: z.string(), yearsOfExperience: z.number(), skills: z.array(z.string()), score: z.number().min(0).max(100), summary: z.string(),})const job = "Senior Backend Engineer (Node.js, PostgreSQL)."const files = await readdir("./resumes")const resumes = files.filter((file) => file.endsWith(".pdf"))const results = await Promise.all( resumes.map(async (file) => { const box = await Box.create({ runtime: "node", agent: { model: "anthropic/claude-opus-4-5", apiKey: process.env.ANTHROPIC_API_KEY, }, }) await box.files.upload([ { path: `./resumes/${file}`, destination: "/work/resume.pdf" }, ]) const run = await box.agent.run({ prompt: `Read /work/resume.pdf. Extract candidate data and score 0-100 for: ${job}`, responseSchema, }) await box.delete() return { file, ...run.result, cost: run.cost.totalUsd } }),)const ranked = results.sort((a, b) => b.score - a.score)import asyncioimport osfrom pydantic import BaseModel, Fieldfrom upstash_box import AsyncBoxclass Candidate(BaseModel): name: str email: str years_of_experience: int skills: list[str] score: int = Field(ge=0, le=100) summary: strjob = "Senior Backend Engineer (Node.js, PostgreSQL)."resumes = [f for f in os.listdir("./resumes") if f.endswith(".pdf")]async def process(file: str) -> dict: box = await AsyncBox.create( runtime="node", agent={"model": "anthropic/claude-opus-4-5", "api_key": os.environ["ANTHROPIC_API_KEY"]}, ) await box.files.upload([{"path": f"./resumes/{file}", "destination": "/work/resume.pdf"}]) run = await box.agent.run( prompt=f"Read /work/resume.pdf. Extract candidate data and score 0-100 for: {job}", response_schema=Candidate, ) await box.delete() return {"file": file, "candidate": run.result, "cost": run.cost.total_usd}async def main() -> list: results = await asyncio.gather(*(process(f) for f in resumes)) return sorted(results, key=lambda r: r["candidate"].score, reverse=True)ranked = asyncio.run(main())API
Prompt (required)
Type: string
Supported on: box.agent.run() and box.agent.stream()
The task instruction sent to the agent.
Options
Type: AgentOptions
Supported on: box.agent.run(), box.agent.stream(), and box.schedule.agent()
Provider-specific agent options forwarded to the underlying runner.
const stream = await box.agent.stream({ prompt: "Review the latest git diff and summarize risks", options: { effort: "high", maxTurns: 20, },})stream = box.agent.stream( prompt="Review the latest git diff and summarize risks", options={ "effort": "high", "max_turns": 20, },)Timeout
Type: number
Supported on: box.agent.run(), box.agent.stream(), and box.schedule.agent()
Default: no execution timeout
Execution timeout in milliseconds. When reached, the run is aborted.
onToolUse
Type: { name: string; input: Record<string, unknown> }
Supported on: box.agent.run() and box.agent.stream()
Called whenever the agent invokes a tool (for example file, shell, or git tools).
responseSchema
Type: Zod Schema
Supported on: box.agent.run()
Attach a Zod schema to get typed output.
import { z } from "zod"const responseSchema = z.object({ customers: z.array( z.object({ name: z.string(), revenue: z.number(), }), ),})const analysis = await box.agent.run({ prompt: "Analyze /work/report.csv and return top customers by revenue", responseSchema,})console.log(analysis.result.customers)from pydantic import BaseModelclass Customer(BaseModel): name: str revenue: floatclass Analysis(BaseModel): customers: list[Customer]analysis = box.agent.run( prompt="Analyze /work/report.csv and return top customers by revenue", response_schema=Analysis,)print(analysis.result.customers)maxRetries
Type: number
Supported on: box.agent.run()
Default: 0
Retry count to compensate temporary provider outages or similar transient errors. Retries use exponential backoff (1s, 2s, 4s, ...) capped at 30s.
Webhook
Type: WebhookConfig
Supported on: box.agent.run()
Useful for fire-and-forget mode. The SDK returns immediately and sends the completion payload to your webhook URL when the run succeeds or fails.