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

Claude
OpenCode
Codex

Get your Claude API key from the Claude Console.

.env

UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxx
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxx

import { 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 os
from upstash_box import Box, Agent

box = Box.create(
    runtime="node",
    agent={
        "harness": Agent.CLAUDE_CODE,
        "model": "anthropic/claude-sonnet-4-5",
        "api_key": os.environ["ANTHROPIC_API_KEY"],
    },
)

.env

UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxx
OPENCODE_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxx

import { 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 os
from upstash_box import Box, Agent

box = Box.create(
    runtime="node",
    agent={
        "harness": Agent.OPEN_CODE,
        "model": "opencode/claude-sonnet-4-6",
        "api_key": os.environ["OPENCODE_API_KEY"],
    },
)

.env

UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxx

import { 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 os
from upstash_box import Box, Agent

box = 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, systemPrompt
Codex: modelReasoningEffort, modelReasoningSummary, personality, webSearch
OpenCode: 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

Refactor Code
Review Changes
Structured Output
Open a PR
Process Documents

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 BaseModel

class Customer(BaseModel):
    name: str
    revenue: float

class 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 os
from upstash_box import Box, Agent

box = 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 asyncio
import os
from pydantic import BaseModel, Field
from upstash_box import AsyncBox

class Candidate(BaseModel):
    name: str
    email: str
    years_of_experience: int
    skills: list[str]
    score: int = Field(ge=0, le=100)
    summary: str

job = "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 BaseModel

class Customer(BaseModel):
    name: str
    revenue: float

class 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.

Introduction

Basics

Lifecycle

Security

Guides

Configure an Agent

Agent Options

Quickstart

Codebase refactor

Fast code review summary

Structured analysis

API

Prompt (required)

Options

Timeout

onToolUse

responseSchema

maxRetries

Webhook

​Configure an Agent

​Agent Options

​Quickstart

​Codebase refactor

​Fast code review summary

​Structured analysis

​API

​Prompt (required)

​Options

​Timeout

​onToolUse

​responseSchema

​maxRetries

​Webhook

Configure an Agent

Agent Options

Quickstart

Codebase refactor

Fast code review summary

Structured analysis

API

Prompt (required)

Options

Timeout

onToolUse

responseSchema

maxRetries

Webhook