Getting started with the Claude Agent SDK

 your prompt
     │
     ▼
 ┌─────────────────────────────┐
 │ Claude thinks               │◄──────────────────────────────┐
 │ "I need to read utils.py"   │                               │
 └──────────────┬──────────────┘                               │
                │ tool call(s)                                 │ tool results
                ▼                                              │
 ┌─────────────────────────────┐                               │
 │ SDK runs the tool(s)        │───────────────────────────────┘
 └─────────────────────────────┘
          ... repeat (each repeat is one TURN) ...
                │
                ▼ no more tool calls
 ┌─────────────────────────────┐
 │ final text answer           │
 │ + ResultMessage (cost, id)  │
 └─────────────────────────────┘

# Python, with uv (recommended)
uv init && uv add claude-agent-sdk

# Python, with pip
python3 -m venv .venv && source .venv/bin/activate
pip install claude-agent-sdk

# TypeScript
npm init -y && npm pkg set type=module
npm install @anthropic-ai/claude-agent-sdk zod
npm install --save-dev tsx

export ANTHROPIC_API_KEY=sk-ant-...

sample-project/
├── CLAUDE.md                 # project rules the agent will read: "use type hints", "be brief"
├── README.md
├── .claude/skills/security-check/SKILL.md   # a reusable "skill" (part 2)
├── src/app.py                # a hardcoded API key and an SQL injection bug
├── src/utils.py              # two functions that crash on edge cases
└── tests/test_utils.py       # 5 tests; 3 fail until utils.py is fixed

# src/utils.py, as shipped
def calculate_average(numbers):
    total = 0
    for num in numbers:
        total += num
    return total / len(numbers)        # crashes on an empty list


def get_user_name(user):
    return user["name"].upper()        # crashes if user is None or has no "name"

import asyncio
from claude_agent_sdk import ClaudeAgentOptions, query
from demo_util import PROJECT, show          # (1) tiny helpers: the project path, and a message printer

async def main():                             # (2) the SDK is async, so everything lives in an async function
    options = ClaudeAgentOptions(
        cwd=PROJECT,                          # (3) the folder the agent works in
        allowed_tools=["Read", "Glob", "Grep"],   # (4) these tools run without asking for permission
        max_turns=5,                          # (5) a safety cap: at most 5 tool round-trips
    )
    async for message in query(               # (6) start the loop; it yields messages as they happen
        prompt="What does this project do? Answer in two sentences.",
        options=options,
    ):
        show(message)                         # (7) print each message in a compact form

asyncio.run(main())                           # (8) run the async function

[system:init] session=de3a42e9-... model=claude-sonnet-5 cli=2.1.281 mode=default
              tools=['Task', 'Bash', 'Edit', 'Glob', 'Grep', 'NotebookEdit', 'Read', 'Skill', 'ToolSearch', 'WebFetch', 'WebSearch', 'Write', ...]
[assistant:tool_use] Bash({'command': 'ls -la ".../demo/project"', 'description': 'List project directory contents'})
[user:tool_result] total 24\ndrwxr-xr-x 4 root root 4096 Oct 16 06:10 .\n...CLAUDE....
[assistant:tool_use] Read({'file_path': '.../demo/project/README.md'})
[user:tool_result] 1	# Sample project\n2	\n3	A tiny Python project used to demonstrate the Claude Agent SDK...
[assistant:tool_use] Bash({'command': 'ls -la ".../demo/project/src"', 'description': 'List src directory'})
[user:tool_result] total 16\n...app.py\n...utils.py
[assistant:text] This project is a tiny Python 3.13 demo for the Claude Agent SDK, with `src/utils.py` providing
                 helper functions and `src/app.py` implementing a toy user lookup. The lookup intentionally
                 contains a known SQL injection bug for demonstration purposes.
[result] subtype=success turns=4 duration=9773ms cost=$0.1260 session=de3a42e9-...

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "What does this project do? Answer in two sentences.",
  options: { cwd: PROJECT, allowedTools: ["Read", "Glob", "Grep"], maxTurns: 5 },
})) {
  if (message.type === "assistant") {                 // string check instead of isinstance
    for (const block of message.message.content) {    // one level deeper than Python
      if (block.type === "text") console.log(block.text);
      else if (block.type === "tool_use") console.log(`tool: ${block.name}`);
    }
  } else if (message.type === "result" && message.subtype === "success") {
    console.log(message.result, message.total_cost_usd);
  }
}

async for message in query(
    prompt="Review src/utils.py for bugs that would cause crashes. Fix any issues you find.",
    options=ClaudeAgentOptions(
        cwd=PROJECT,
        allowed_tools=["Read", "Edit", "Glob"],  # (1) Edit is now pre-approved
        permission_mode="acceptEdits",           # (2) and file edits need no confirmation
        max_turns=10,
    ),
):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if hasattr(block, "text"):                       # (3) a TextBlock: Claude explaining itself
                print(block.text)
            elif isinstance(block, ToolUseBlock):            # (4) a tool call: show which file it touches
                print(f"Tool: {block.name} -> {block.input.get('file_path', '')}")
    elif isinstance(message, ResultMessage):
        print(f"Done: {message.subtype} (cost ${message.total_cost_usd:.4f})")

Tool: Read -> .../project/src/utils.py
Tool: Edit -> .../project/src/utils.py
Fixed two crash bugs: `calculate_average` divided by zero on an empty list, and `get_user_name`
raised `KeyError` for a missing `"name"` key — both now guard against those cases, with type hints
added per project convention.
Done: success (cost $0.0550)

--- diff against the original file ---
-def calculate_average(numbers):
+def calculate_average(numbers: list[float]) -> float:
+    if not numbers:
+        return 0.0
     total = 0
     ...
-def get_user_name(user):
-    return user["name"].upper()
+def get_user_name(user: dict) -> str:
+    return user.get("name", "").upper()

options = ClaudeAgentOptions(
    cwd="/path/to/repo",
    model="claude-opus-5-5",
    fallback_model="claude-sonnet-5-5",
    effort="high",
    allowed_tools=["Read", "Glob", "Grep"],
    max_turns=8,
    max_budget_usd=2.0,
    env={"ANTHROPIC_BASE_URL": "https://gateway.example.com", "API_TIMEOUT_MS": "120000"},
)

const options = {
  cwd: "/path/to/repo",
  model: "claude-opus-5-5",
  fallbackModel: "claude-sonnet-5-5,claude-haiku-5-5",
  effort: "high",
  allowedTools: ["Read", "Glob", "Grep"],
  maxTurns: 8,
  maxBudgetUsd: 2,
  env: { ...process.env, ANTHROPIC_BASE_URL: "https://gateway.example.com" },  // spread process.env!
};

async with ClaudeSDKClient(options=options) as client:          # (1) opens the connection; closes it at the end
    await client.query("Which function in src/app.py has a SQL injection bug? Name it only.")
    async for message in client.receive_response():            # (2) read until this turn's ResultMessage
        print_response(message)

    await client.query("Show me the one-line fix for that function, as a code snippet only.")  # (3) same session
    async for message in client.receive_response():
        print_response(message)

Claude: `find_user`
[done: success, turns=3, cost=$0.0525]

Claude: ```python
    rows = conn.execute("SELECT * FROM users WHERE name = ?", (name,)).fetchall()

🔍 Turn one took three tool round trips (find and read the file). Turn two took **one** and used no tools: "that function" was resolved from memory, and the file was already in context. The client holds the session; you did nothing.

| Method | Use |
|---|---|
| `query(prompt)` | send a message (a string, or an async generator of messages) |
| `receive_response()` | iterate messages until the next `ResultMessage` |
| `receive_messages()` | iterate forever (for UIs) |
| `interrupt()` | stop the current turn |
| `set_permission_mode(mode)`, `set_model(model)` | change settings mid-session |
| `get_context_usage()`, `get_mcp_status()`, `get_server_info()` | inspect (part 3) |
| `rewind_files(uuid)` | undo file edits (part 3) |

### Streaming input: images and queued messages

An **async generator** of message dicts is the "stream". You can `await` between items (user input, a file arriving) and attach images.

```python
async def message_generator():
    yield {"type": "user", "message": {"role": "user", "content": "Analyze this codebase for security issues"}}
    await asyncio.sleep(2)                                  # wait for anything you like
    with open("diagram.png", "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    yield {"type": "user", "message": {"role": "user", "content": [
        {"type": "text", "text": "Review this architecture diagram"},
        {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": image_data}},
    ]}}

async with ClaudeSDKClient(options) as client:
    await client.query(message_generator())
    async for message in client.receive_response():          # stops at the FIRST result; loop again for the second
        ...

await client.query("Count from 1 to 60, one number per second, using `sleep 1` in bash between numbers.")
await asyncio.sleep(8)
await client.interrupt()                                    # (1) stop it
async for message in client.receive_response():             # (2) ALWAYS drain the interrupted turn first
    if isinstance(message, ResultMessage):
        print(message.subtype, message.terminal_reason)
await client.query("Just say hello instead.")               # (3) now continue

interrupt() sent after 8.0s
  tool call: Bash({'command': 'for i in $(seq 1 60); do echo "$i"; sleep 1; do)
interrupted turn: subtype=error_during_execution terminal_reason='aborted_tools' after 8.0s
next turn: success -> 'Hello!'

>>> /compact
[compact_boundary] trigger=manual pre_tokens=26993
>>> /clear
[conversation reset] new_conversation_id=4ac7c754-...
>>> What did we talk about before? Answer in one sentence.
Claude: This session just started, so there's no prior conversation to refer back to.

try:
    async for message in query(prompt=..., options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read", "Glob"], **opts)):
        if isinstance(message, ResultMessage):
            print(f"ResultMessage: subtype={message.subtype} terminal_reason={message.terminal_reason!r}")
except CLINotFoundError as e:       # (1) no claude binary at all
    print(f"CLINotFoundError: {e}")
except ResultError as e:            # (2) the run ended with an error result. Catch BEFORE ProcessError
    print(f"ResultError: subtype={e.subtype} terminal_reason={e.terminal_reason!r}: {e}")
except ProcessError as e:           # (3) the subprocess died without a result
    print(f"ProcessError: exit_code={e.exit_code}")

=== max_turns=1 ===
ResultMessage: subtype=error_max_turns is_error=True turns=2 cost=$0.0367 terminal_reason='max_turns'
  result field: None
ResultError raised after the result: subtype=error_max_turns terminal_reason='max_turns'
  message: Claude Code returned an error result: Reached maximum number of turns (1) (exit code: 1)

=== max_budget_usd=0.02 ===
ResultMessage: subtype=error_max_budget_usd is_error=True turns=1 cost=$0.0369 terminal_reason='budget_exhausted'
  result field: None
ResultError raised after the result: subtype=error_max_budget_usd terminal_reason='budget_exhausted'

=== bad cli_path ===
CLINotFoundError: Claude Code not found at: /nonexistent/claude

async for message in query(prompt="Read README.md and summarize it in one sentence.", options=options):
    if isinstance(message, StreamEvent):
        event = message.event
        if event.get("type") == "content_block_start" and event["content_block"].get("type") == "tool_use":
            sys.stdout.write(f"\n[Using {event['content_block']['name']}...]"); in_tool = True       # (1) a tool call begins
        elif event.get("type") == "content_block_delta":
            delta = event["delta"]
            if delta.get("type") == "text_delta" and not in_tool:
                sys.stdout.write(delta["text"]); sys.stdout.flush()                                  # (2) print words as they arrive
            elif delta.get("type") == "input_json_delta":
                sys.stdout.write(".")                                                                # (3) a dot per chunk of tool input
        elif event.get("type") == "content_block_stop" and in_tool:
            sys.stdout.write(" done\n"); in_tool = False

[Using Read...].............................. done
This is a tiny demo Python project for the Claude Agent SDK, containing helper functions and a toy app with a known SQL injection bug.

--- Complete (success) ---
Stream event counts: {'message_start': 2, 'content_block_start': 3, 'content_block_delta': 43, 'content_block_stop': 3, 'message_delta': 2, 'message_stop': 2}

class Finding(BaseModel):
    file: str; line: int; severity: str; description: str

class SecurityReport(BaseModel):
    findings: list[Finding]; total_count: int; overall_risk: str

async for message in query(
    prompt="Audit every file under src/ for security problems and report them.",
    options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read", "Glob", "Grep"], max_turns=8,
        output_format={"type": "json_schema", "schema": SecurityReport.model_json_schema()}),   # (1) the schema
):
    if isinstance(message, ResultMessage) and message.subtype == "success" and message.structured_output:  # (2) check BOTH
        report = SecurityReport.model_validate(message.structured_output)                                   # (3) a typed object

{
  "findings": [
    {"file": "src/app.py", "line": 5, "severity": "high",
     "description": "Hardcoded live-looking API key committed directly in source code instead of loaded from environment/secret store."},
    {"file": "src/app.py", "line": 11, "severity": "high",
     "description": "SQL injection: user-controlled `name` is interpolated via f-string directly into the SQL query instead of using a parameterized query."}
  ],
  "total_count": 2,
  "overall_risk": "high"
}