# 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

<Tabs>
<Tab title="Claude">

Get your Claude API key from the [Claude Console](https://platform.claude.com/settings/keys).

```bash title=".env" {2}
UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxx
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxx
```

<CodeGroup>
```tsx box.ts {5-8}
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!,
  },
})
```

```python box.py
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"],
    },
)
```
</CodeGroup>

</Tab>
<Tab title="OpenCode">

```bash title=".env" {2}
UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxx
OPENCODE_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxx
```

<CodeGroup>
```tsx box.ts {5-8}
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!,
  },
})
```

```python box.py
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"],
    },
)
```
</CodeGroup>

</Tab>
<Tab title="Codex">

```bash title=".env" {2}
UPSTASH_BOX_API_KEY=box_xxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxx
```

<CodeGroup>
```tsx box.ts {5-8}
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!,
  },
})
```

```python box.py
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"],
    },
)
```
</CodeGroup>

</Tab>
</Tabs>

### Agent Options

You can pass provider-specific agent options with the `options` field on `box.agent.run()` and `box.agent.stream()`.

<CodeGroup>
```tsx box.ts
const run = await box.agent.run({
  prompt: "Refactor the auth flow and keep changes minimal",
  options: {
    effort: "medium",
    maxTurns: 12,
  },
})
```

```python box.py
run = box.agent.run(
    prompt="Refactor the auth flow and keep changes minimal",
    options={
        "effort": "medium",
        "max_turns": 12,
    },
)
```
</CodeGroup>

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`

<Note>
  In the Python SDK these option keys are **snake_case** — e.g. `max_turns`, `max_budget_usd`, `model_reasoning_effort`, `web_search`, `reasoning_effort`.
</Note>

To bring your own agent process, use a [custom agent](/box/overall/custom-agent).

## Quickstart

<Tabs>
<Tab title="Refactor Code">

### Codebase refactor

Use Claude Code when you want an agent to work through a larger code change with filesystem, shell, and git access.

<CodeGroup>
```tsx box.ts
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)
```

```python box.py
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)
```
</CodeGroup>

</Tab>
<Tab title="Review Changes">

### Fast code review summary

Use OpenCode when you want a concise review or summary over an existing diff or repository state.

<CodeGroup>
```tsx box.ts
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)
```

```python box.py
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)
```
</CodeGroup>

</Tab>
<Tab title="Structured Output">

### Structured analysis

Use Codex when you want a strongly structured response that you can feed into another system.

<CodeGroup>
```tsx box.ts
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)
```

```python box.py
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)
```
</CodeGroup>

</Tab>
<Tab title="Open a PR">

Clone a repository, run the agent, and open a pull request.

<CodeGroup>
```tsx box.ts
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",
})
```

```python box.py
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")
```
</CodeGroup>

</Tab>
<Tab title="Process Documents">

Run one box per file, process in parallel, and return ranked structured results.

<CodeGroup>
```tsx box.ts
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)
```

```python box.py
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())
```
</CodeGroup>

</Tab>
</Tabs>

---

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

<CodeGroup>
```tsx box.ts
const stream = await box.agent.stream({
  prompt: "Review the latest git diff and summarize risks",
  options: {
    effort: "high",
    maxTurns: 20,
  },
})
```

```python box.py
stream = box.agent.stream(
    prompt="Review the latest git diff and summarize risks",
    options={
        "effort": "high",
        "max_turns": 20,
    },
)
```
</CodeGroup>

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

<CodeGroup>
```tsx box.ts
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)
```

```python box.py
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)
```
</CodeGroup>

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

---