Chapter 13: Agent SDK (Python / TypeScript)

What You Will Learn

What the Agent SDK is and how it differs from the CLI and Managed Agents
Build a PR Review Bot that reads diffs, runs subagent analysis, and posts review comments
Build a Dependency Auditor that scans lockfiles for outdated and vulnerable packages
Build a Multi-step Migration Pipeline using session resumption
SDK patterns: tool use, extended thinking, subagents, hooks, and the Batches API

What Is the Agent SDK?

The Agent SDK wraps Claude Code's orchestration loop (the "harness") into a library you can import in Python or TypeScript. Your code calls query(), and Claude executes using the same tools it has in the terminal -- Read, Write, Edit, Bash, Glob, Grep -- but under your programmatic control.

CLI:            You type in the terminal     Claude runs interactively
Agent SDK:      Your code calls query()      Claude runs on your server
Managed Agents: Your API calls the REST API  Claude runs on Anthropic's cloud

When to use each:

Need	Use
Interactive development	CLI
Embedding Claude in your app, CI scripts, internal tools	Agent SDK
Long-running cloud tasks, fault tolerance, zero infrastructure	Managed Agents

Installation

bash

# Python
pip install claude-code-sdk

# TypeScript
npm install @anthropic-ai/claude-code-sdk

Research note: The Agent SDK's architecture decision to expose the same tool set as the CLI was intentional -- it means agent behavior tested interactively in the terminal transfers directly to programmatic use, reducing the gap between development and production. This "same tools, different interface" principle is documented in Anthropic's SDK design notes.

Basic Usage

python

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def main():
    async for message in query(
        prompt="Find all TODO comments in this project and list them",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Glob", "Grep"],
        ),
    ):
        if message.type == "text":
            print(message.text, end="")

asyncio.run(main())

Expected output:

terminal

Found 12 TODO comments across the project:

src/auth/middleware.ts:47   # TODO: Add rate limiting for failed login attempts
src/auth/middleware.ts:89   # TODO: Support refresh token rotation
src/routes/users.ts:23     # TODO: Add pagination to user list endpoint
src/routes/orders.ts:156   # TODO: Implement order cancellation logic
src/services/email.ts:12   # TODO: Switch to async email queue
src/services/email.ts:78   # TODO: Add HTML template support
tests/auth.test.ts:5       # TODO: Add integration tests for OAuth flow
tests/orders.test.ts:34    # TODO: Mock payment gateway responses
lib/db.ts:91               # TODO: Add connection pooling
lib/cache.ts:15            # TODO: Implement cache invalidation strategy
lib/cache.ts:44            # TODO: Add TTL support
config/app.ts:8            # TODO: Move secrets to environment variables

Demo 34: PR Review Bot

PR Review Bot with Parallel Subagents

Advanced~25 min

What We Are Building

A Python script that:

Fetches a PR diff from GitHub
Spawns three subagents in parallel: security reviewer, logic reviewer, and style reviewer
Aggregates their findings
Posts a structured review comment on the PR

This is a realistic internal tool. Teams at companies like Sentry and Notion run similar setups.

The Code

python

# pr_review_bot.py

import asyncio
import json
import subprocess
from claude_code_sdk import query, ClaudeCodeOptions

async def review_pr(repo: str, pr_number: int):
    """
    Run a multi-agent review on a GitHub PR.
    
    Args:
        repo: "owner/repo" format
        pr_number: The PR number to review
    """
    
    # Step 1: Fetch the PR diff
    result = subprocess.run(
        ["gh", "api", f"repos/{repo}/pulls/{pr_number}",
         "--header", "Accept: application/vnd.github.v3.diff"],
        capture_output=True, text=True
    )
    diff = result.stdout
    
    if not diff:
        print(f"Could not fetch diff for {repo}#{pr_number}")
        return
    
    # Step 2: Fetch PR metadata
    meta = subprocess.run(
        ["gh", "api", f"repos/{repo}/pulls/{pr_number}"],
        capture_output=True, text=True
    )
    pr_info = json.loads(meta.stdout)
    pr_title = pr_info.get("title", "")
    pr_body = pr_info.get("body", "")
    
    # Step 3: Run three review agents in parallel
    reviews = await asyncio.gather(
        run_security_review(diff, pr_title),
        run_logic_review(diff, pr_title, pr_body),
        run_style_review(diff),
    )
    
    security_findings, logic_findings, style_findings = reviews
    
    # Step 4: Aggregate and post
    comment = format_review_comment(
        pr_title, security_findings, logic_findings, style_findings
    )
    
    subprocess.run(
        ["gh", "api", f"repos/{repo}/issues/{pr_number}/comments",
         "--method", "POST",
         "--field", f"body={comment}"],
    )
    
    print(f"Review posted on {repo}#{pr_number}")


async def run_security_review(diff: str, title: str) -> str:
    """Subagent focused on security vulnerabilities."""
    output = []
    async for msg in query(
        prompt=f"""You are a security reviewer. Analyze this PR diff for
security vulnerabilities.

PR Title: {title}

Diff:
{diff[:30000]}

Check for:
- SQL injection, XSS, SSRF, path traversal
- Authentication/authorization gaps
- Hardcoded secrets or credentials
- Insecure deserialization
- Command injection

Output a JSON array of findings. Each finding:
{{"severity": "critical|high|medium", "file": "...", "line": N, "issue": "...", "fix": "..."}}

If no issues found, output: []""",
        options=ClaudeCodeOptions(
            allowed_tools=[],  # Read-only analysis, no tools needed
            max_tokens=4096,
        ),
    ):
        if msg.type == "text":
            output.append(msg.text)
    return "".join(output)


async def run_logic_review(diff: str, title: str, body: str) -> str:
    """Subagent focused on correctness and logic errors."""
    output = []
    async for msg in query(
        prompt=f"""You are a code reviewer focused on correctness. Analyze this PR.

PR Title: {title}
PR Description: {body[:2000]}

Diff:
{diff[:30000]}

Check for:
- Logic errors (off-by-one, wrong conditions, missing edge cases)
- Error handling gaps (unhandled exceptions, missing null checks)
- Race conditions in async code
- Incorrect API usage
- Missing input validation

Output a JSON array of findings.
{{"severity": "critical|high|medium", "file": "...", "line": N, "issue": "...", "suggestion": "..."}}

If no issues found, output: []""",
        options=ClaudeCodeOptions(
            allowed_tools=[],
            max_tokens=4096,
        ),
    ):
        if msg.type == "text":
            output.append(msg.text)
    return "".join(output)


async def run_style_review(diff: str) -> str:
    """Subagent focused on code style and maintainability."""
    output = []
    async for msg in query(
        prompt=f"""You are a code style reviewer. Focus only on substantial
maintainability issues, not nitpicks.

Diff:
{diff[:30000]}

Check for:
- Functions longer than 50 lines that should be split
- Duplicated logic that should be extracted
- Missing type annotations on public APIs
- Confusing naming that will slow down future readers
- Dead code or unreachable branches

Output a JSON array of findings.
{{"severity": "medium|low", "file": "...", "line": N, "issue": "...", "suggestion": "..."}}

If no issues found, output: []""",
        options=ClaudeCodeOptions(
            allowed_tools=[],
            max_tokens=4096,
        ),
    ):
        if msg.type == "text":
            output.append(msg.text)
    return "".join(output)


def format_review_comment(
    title: str, security: str, logic: str, style: str
) -> str:
    """Format the three reviews into a single PR comment."""
    
    def parse_findings(raw: str) -> list:
        try:
            # Find the JSON array in the output
            start = raw.find("[")
            end = raw.rfind("]") + 1
            if start >= 0 and end > start:
                return json.loads(raw[start:end])
        except json.JSONDecodeError:
            pass
        return []
    
    sec_items = parse_findings(security)
    logic_items = parse_findings(logic)
    style_items = parse_findings(style)
    
    total = len(sec_items) + len(logic_items) + len(style_items)
    has_critical = any(
        f.get("severity") == "critical"
        for f in sec_items + logic_items
    )
    
    parts = [f"## Code Review: {title}\n"]
    
    if has_critical:
        parts.append("**BLOCKING**: Critical issues found that must be resolved before merge.\n")
    elif total == 0:
        parts.append("No significant issues found. Looks good to merge.\n")
    else:
        parts.append(f"Found {total} issue(s) to address.\n")
    
    if sec_items:
        parts.append("### Security\n")
        for f in sec_items:
            parts.append(
                f"- **{f['severity'].upper()}** `{f.get('file', '?')}:{f.get('line', '?')}` "
                f"-- {f.get('issue', '')}\n  Fix: {f.get('fix', f.get('suggestion', ''))}\n"
            )
    
    if logic_items:
        parts.append("\n### Logic & Correctness\n")
        for f in logic_items:
            parts.append(
                f"- **{f['severity'].upper()}** `{f.get('file', '?')}:{f.get('line', '?')}` "
                f"-- {f.get('issue', '')}\n  Suggestion: {f.get('suggestion', '')}\n"
            )
    
    if style_items:
        parts.append("\n### Maintainability\n")
        for f in style_items:
            parts.append(
                f"- `{f.get('file', '?')}:{f.get('line', '?')}` "
                f"-- {f.get('issue', '')}\n  {f.get('suggestion', '')}\n"
            )
    
    return "\n".join(parts)


if __name__ == "__main__":
    import sys
    if len(sys.argv) != 3:
        print("Usage: python pr_review_bot.py owner/repo PR_NUMBER")
        sys.exit(1)
    
    asyncio.run(review_pr(sys.argv[1], int(sys.argv[2])))

Running It

bash

# Review PR #42 on your repo
python pr_review_bot.py myorg/myapp 42

What Happens

pr_review_bot.py
  ├── Fetches PR #42 diff via gh CLI
  ├── Spawns 3 subagents (parallel):
  │   ├── Security reviewer → JSON findings
  │   ├── Logic reviewer → JSON findings
  │   └── Style reviewer → JSON findings
  ├── Aggregates findings into a structured comment
  └── Posts the comment via GitHub API

Typical runtime: 15-30 seconds
Typical cost: $0.03-0.10 per review

Expected terminal output:

terminal

$ python pr_review_bot.py myorg/myapp 42
Review posted on myorg/myapp#42

On the PR itself, the posted comment looks like:

terminal

## Code Review: Add user profile editing

Found 3 issue(s) to address.

### Security
- **HIGH** src/routes/profile.ts:47 -- User ID taken from request body
  instead of auth token, allowing users to edit other profiles.
  Fix: Use req.user.id from the authenticated token instead of req.body.userId

### Logic & Correctness
- **MEDIUM** src/services/profile.ts:89 -- Missing null check on
  database query result before accessing .email property.
  Suggestion: Add guard clause: if (!user) return res.status(404)

### Maintainability
- src/routes/profile.ts:12 -- updateProfile function is 78 lines long
  with deeply nested conditionals.
  Extract validation logic into a separate validateProfileUpdate() function

What Just Happened?

Bash (subprocess)

gh api repos/myorg/myapp/pulls/42

↓

query() - Security Agent

Diff text (truncated to 30k chars)

↓

query() - Logic Agent

Diff text plus PR description

↓

query() - Style Agent

Diff text only

↓

Bash (subprocess)

gh api repos/myorg/myapp/issues/42/comments

Demo 35: Dependency Auditor

Dependency Auditor with Tool-Enabled Scanning

Advanced~15 min

What We Are Building

A script that scans your project's dependency files (package.json, requirements.txt, go.mod, etc.), identifies outdated and vulnerable packages, and produces an actionable report.

python

# dependency_auditor.py

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def audit_dependencies(project_dir: str):
    """
    Scan a project for dependency issues.
    
    Claude uses actual tools to:
    1. Find dependency files
    2. Run package manager audit commands
    3. Check for outdated packages
    4. Cross-reference with known CVE databases
    """
    
    output = []
    async for msg in query(
        prompt=f"""Audit the dependencies in {project_dir}. Do the following:

1. Find all dependency files:
   - package.json / package-lock.json (Node.js)
   - requirements.txt / Pipfile / pyproject.toml (Python)
   - go.mod (Go)
   - Cargo.toml (Rust)
   - pom.xml / build.gradle (Java)

2. For each dependency file found:
   a. Run the appropriate audit command:
      - npm: npm audit --json
      - pip: pip-audit --format json (if available) or pip install safety && safety check
      - go: go list -m -json all
   b. Run the outdated check:
      - npm: npm outdated --json
      - pip: pip list --outdated --format json

3. Produce a report with these sections:

CRITICAL VULNERABILITIES (must fix immediately):
- package@version: CVE-XXXX-XXXX - description - fixed in version X.Y.Z

HIGH VULNERABILITIES:
- ...

OUTDATED (major version behind):
- package: current X.Y.Z -> latest A.B.C (major bump, check changelog)

OUTDATED (minor/patch behind):
- package: current X.Y.Z -> latest X.Y.W

SUMMARY:
- Total dependencies: N
- Vulnerable: N (critical: N, high: N, moderate: N)
- Outdated: N (major: N, minor: N, patch: N)
- Recommendation: [fix critical vulns first, then update majors one at a time]

Be specific. Include actual CVE numbers, affected versions, and fixed versions.
Do not hallucinate CVE numbers -- only report what the audit tools actually find.""",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Glob", "Grep", "Bash"],
            permission_mode="auto",
            # Only allow safe read-only and audit commands
            allowed_bash_commands=[
                "npm audit *",
                "npm outdated *",
                "npm list *",
                "pip-audit *",
                "pip list *",
                "safety check *",
                "go list *",
                "cargo audit *",
                "cat *",
                "find * -name *",
            ],
        ),
    ):
        if msg.type == "text":
            output.append(msg.text)
    
    report = "".join(output)
    print(report)
    
    # Save to file
    with open(f"{project_dir}/dependency-audit-report.txt", "w") as f:
        f.write(report)
    
    return report


if __name__ == "__main__":
    import sys
    project = sys.argv[1] if len(sys.argv) > 1 else "."
    asyncio.run(audit_dependencies(project))

Running It

bash

# Audit the current project
python dependency_auditor.py .

# Audit a specific project
python dependency_auditor.py /path/to/your/project

Expected terminal output:

terminal

$ python dependency_auditor.py .

CRITICAL VULNERABILITIES (must fix immediately):
- jsonwebtoken@8.5.1: CVE-2022-23529 - Insecure key retrieval allows
  JWT forgery - fixed in 9.0.0

HIGH VULNERABILITIES:
- express@4.17.1: CVE-2024-29041 - Open redirect vulnerability in
  res.redirect - fixed in 4.19.2
- axios@0.21.1: CVE-2023-45857 - SSRF due to unexpected behavior with
  non-standard URL parsing - fixed in 1.6.0

OUTDATED (major version behind):
- typescript: 4.9.5 -> 5.4.5 (major bump, check changelog for breaking changes)
- eslint: 7.32.0 -> 9.3.0 (major bump, plugin compatibility may be affected)

OUTDATED (minor/patch behind):
- prisma: 5.10.2 -> 5.14.0
- vitest: 1.4.0 -> 1.6.0

SUMMARY:
- Total dependencies: 847 (412 direct, 435 transitive)
- Vulnerable: 3 (critical: 1, high: 2, moderate: 0)
- Outdated: 4 (major: 2, minor: 1, patch: 1)
- Recommendation: Fix jsonwebtoken (critical) immediately, then upgrade
  express and axios. Major version bumps for typescript and eslint can
  be scheduled separately.

Report saved to ./dependency-audit-report.txt

What Just Happened?

Glob

package.json, requirements.txt, go.mod, etc.

↓

Read

package.json

↓

Bash

npm audit --json

↓

Bash

npm outdated --json

↓

Bash

npm list --json --depth=0

Demo 36: Multi-step Migration Pipeline

Three-Phase Migration with Session Resumption

Advanced~30 min

What We Are Building

A three-phase migration pipeline that converts a JavaScript Express API to TypeScript. Each phase runs in its own session, and session resumption carries context forward.

This pattern comes from HumanLayer's RPI (Research-Plan-Implement) approach, which handled a 300k LOC Rust codebase by breaking work into discrete phases with intentional compaction between them.

python

# migration_pipeline.py

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def run_migration(project_dir: str):
    """
    Three-phase migration: Research -> Plan -> Execute
    
    Each phase uses session resumption to carry forward context
    while keeping the context window manageable.
    """
    
    # Phase 1: Research the current state
    print("=" * 60)
    print("PHASE 1: RESEARCH")
    print("=" * 60)
    
    session_id = None
    research_output = []
    
    async for msg in query(
        prompt=f"""Research the current state of the JavaScript Express API in {project_dir}.

I need to migrate this to TypeScript. Before planning anything, answer:

1. Project structure: How many files? What is the directory layout?
2. Express version and middleware used
3. Database layer: ORM? Raw queries? Which database?
4. Existing type information: Any JSDoc? Any .d.ts files? Any prop-types?
5. Test setup: What test framework? How many tests? What is the coverage?
6. Build setup: Babel? Webpack? Plain Node?
7. External API contracts: Any OpenAPI specs? GraphQL schemas?
8. Complexity assessment: Which modules will be hardest to convert?

Read the actual files. Do not guess. I need precise answers.""",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Glob", "Grep", "Bash"],
            permission_mode="auto",
            allowed_bash_commands=["find *", "wc *", "cat *", "head *", "npm list *"],
        ),
    ):
        if msg.type == "text":
            research_output.append(msg.text)
            print(msg.text, end="")
        if hasattr(msg, "session_id") and msg.session_id:
            session_id = msg.session_id
    
    research_summary = "".join(research_output)
    
    # Phase 2: Plan the migration
    print("\n\n" + "=" * 60)
    print("PHASE 2: PLAN")
    print("=" * 60)
    
    plan_output = []
    
    async for msg in query(
        prompt=f"""Based on your research, create a detailed migration plan.

The plan should:

1. Define the migration order (which files to convert first)
   - Start with leaf modules (no internal dependencies)
   - Work inward toward the core
   
2. For each file or module, specify:
   - Input: filename.js
   - Output: filename.ts
   - Type definitions needed (interfaces, enums, type aliases)
   - Dependencies that need @types/* packages
   - Risk level (low/medium/high) and why

3. Define checkpoints:
   - After each module conversion, what tests should pass?
   - What is the rollback plan if something breaks?

4. Estimate effort:
   - Mechanical changes (rename .js to .ts, add type annotations)
   - Structural changes (refactor untyped patterns that do not translate)
   - Test changes (update test imports, add type assertions)

Save the plan to {project_dir}/MIGRATION_PLAN.md

This plan needs to be specific enough that another engineer could execute it
without asking questions.""",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Write", "Glob", "Grep", "Bash"],
            resume=session_id,  # Carry forward the research context
        ),
    ):
        if msg.type == "text":
            plan_output.append(msg.text)
            print(msg.text, end="")
        if hasattr(msg, "session_id") and msg.session_id:
            session_id = msg.session_id
    
    # Phase 3: Execute the migration
    print("\n\n" + "=" * 60)
    print("PHASE 3: EXECUTE")
    print("=" * 60)
    
    async for msg in query(
        prompt=f"""Execute the migration plan you just created.

Rules:
1. Convert one module at a time, in the order specified in the plan
2. After each module:
   - Run the type checker: npx tsc --noEmit
   - Run the tests: npm test
   - If either fails, fix the issues before moving on
3. Install @types/* packages as needed (npm install --save-dev @types/express etc.)
4. Create a tsconfig.json if one does not exist
5. Update package.json scripts to use ts-node or tsx
6. Do NOT change any business logic. This is a type-only migration.
7. Commit after each successful module conversion with a descriptive message

Start with the first module in the plan.""",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Write", "Edit", "Glob", "Grep", "Bash"],
            resume=session_id,
            permission_mode="auto",
            allowed_bash_commands=[
                "npm install *",
                "npm test*",
                "npx tsc *",
                "npx jest *",
                "git add *",
                "git commit *",
                "git status",
                "git diff*",
                "mv *",
                "mkdir *",
            ],
        ),
    ):
        if msg.type == "text":
            print(msg.text, end="")
    
    print("\n\nMigration complete.")


if __name__ == "__main__":
    import sys
    project = sys.argv[1] if len(sys.argv) > 1 else "."
    asyncio.run(run_migration(project))

Running It

bash

python migration_pipeline.py /path/to/express-api

Expected terminal output (abbreviated):

terminal

$ python migration_pipeline.py ./my-express-api

============================================================
PHASE 1: RESEARCH
============================================================

Project Structure:
- 23 JavaScript files across src/, lib/, and tests/
- Directory layout:
    src/
      routes/     (6 files)
      middleware/  (3 files)
      services/   (5 files)
      models/     (4 files)
    lib/          (2 files)
    tests/        (3 files)

Express version: 4.18.2
Middleware: cors, helmet, morgan, express-validator
Database: PostgreSQL via Sequelize 6.35.0 (ORM)
Type info: JSDoc on 8 of 23 files, no .d.ts files
Tests: Jest, 47 tests, ~62% coverage
Build: Plain Node with nodemon for dev
API contracts: No OpenAPI spec found
Hardest modules: src/services/order.service.js (complex state
  machine), src/middleware/auth.js (many edge cases)

============================================================
PHASE 2: PLAN
============================================================

Migration Plan saved to ./my-express-api/MIGRATION_PLAN.md

Order of conversion:
  1. lib/constants.js          (leaf, no deps)        LOW risk
  2. lib/helpers.js            (leaf, pure functions)  LOW risk
  3. src/models/user.model.js  (Sequelize model)      MEDIUM risk
  ...
  23. src/app.js               (root, all deps)       HIGH risk

============================================================
PHASE 3: EXECUTE
============================================================

[1/23] Converting lib/constants.js -> lib/constants.ts
  - Added type annotations to 12 exports
  - npx tsc --noEmit: PASS
  - npm test: 47/47 pass
  - Committed: "Convert constants to TypeScript"

[2/23] Converting lib/helpers.js -> lib/helpers.ts
  - Added 6 interfaces for helper function parameters
  - Installed @types/lodash
  - npx tsc --noEmit: PASS
  - npm test: 47/47 pass
  - Committed: "Convert helpers to TypeScript with type definitions"

...

[23/23] Converting src/app.js -> src/app.ts
  - Updated all import paths
  - npx tsc --noEmit: PASS
  - npm test: 47/47 pass
  - Committed: "Convert app entry point to TypeScript - migration complete"

Migration complete.

What Just Happened?

query() Phase 1

Project directory

↓

query() Phase 2

Resumed session from Phase 1

↓

query() Phase 3

Resumed session from Phase 2

Why Session Resumption Matters

Without session resumption, each phase starts cold -- Claude has to re-read the entire codebase to understand what phase 2 should do. With resumption:

Phase 1 (Research):  Reads 23 files, builds mental model     Session: sess_abc
Phase 2 (Plan):      Resumes sess_abc, already knows layout  No re-reading
Phase 3 (Execute):   Resumes sess_abc, knows the plan        Starts immediately

Context efficiency: ~40% fewer tokens compared to three independent sessions.

SDK Patterns Reference

Extended Thinking

For complex architectural decisions, enable extended thinking to let Claude reason through the problem before responding:

python

options = ClaudeCodeOptions(
    allowed_tools=["Read", "Glob", "Grep"],
    model="claude-opus-4-6",
    # Extended thinking gives Claude internal scratchpad space
    # for complex multi-step reasoning
    extended_thinking=True,
)

Appendix link: A09 Extended Thinking covers when thinking helps, budget management, and how thinking interacts with tool calls.

Hooks (Callbacks)

python

from claude_code_sdk import ClaudeCodeOptions, HookMatcher

async def log_file_changes(tool_input, tool_use_id, context):
    """Log every file modification to an audit trail."""
    file_path = tool_input.get("file_path", "unknown")
    with open("./audit.log", "a") as f:
        f.write(f"{context.timestamp}: modified {file_path}\n")

options = ClaudeCodeOptions(
    hooks={
        "PostToolUse": [
            HookMatcher(
                matcher="Edit|Write",
                hooks=[log_file_changes]
            )
        ]
    }
)

Subagent Definitions

Define specialized subagents that the main agent can delegate to:

python

from claude_code_sdk import ClaudeCodeOptions, AgentDefinition

options = ClaudeCodeOptions(
    allowed_tools=["Read", "Glob", "Grep", "Agent"],
    agents={
        "security_reviewer": AgentDefinition(
            description="Reviews code changes for security vulnerabilities",
            instructions="Focus on OWASP Top 10. Output JSON findings.",
            tools=["Read", "Glob", "Grep"],
        ),
        "test_writer": AgentDefinition(
            description="Writes unit tests for new or modified code",
            instructions="Use the existing test framework. Match test style.",
            tools=["Read", "Write", "Glob", "Grep", "Bash"],
        ),
    },
)

Batches API

For processing many items (e.g., auditing 100 PRs), use the Batches API for 50% cost reduction:

python

from claude_code_sdk import create_batch, BatchItem

# Define batch items
items = [
    BatchItem(
        id=f"pr-{pr_num}",
        prompt=f"Review PR #{pr_num}: {pr_title}",
        options=ClaudeCodeOptions(allowed_tools=["Read", "Glob", "Grep"]),
    )
    for pr_num, pr_title in prs_to_review
]

# Submit batch (processes asynchronously, results within 24 hours)
batch = await create_batch(items)
print(f"Batch submitted: {batch.id}")

# Check results later
results = await batch.get_results()
for result in results:
    print(f"{result.id}: {result.output}")

When Things Go Wrong

SDK Version Incompatibility

Symptom: You install the SDK but get import errors or unexpected behavior.

terminal

$ python pr_review_bot.py myorg/myapp 42
Traceback (most recent call last):
  File "pr_review_bot.py", line 3, in <module>
    from claude_code_sdk import query, ClaudeCodeOptions
ImportError: cannot import name 'ClaudeCodeOptions' from 'claude_code_sdk'

Cause: The SDK package name and import structure can change between versions. Older versions may use different class names.

Fix:

bash

# Check your installed version
pip show claude-code-sdk

# Upgrade to latest
pip install --upgrade claude-code-sdk

# Verify the import works
python -c "from claude_code_sdk import query, ClaudeCodeOptions; print('OK')"

Also check that your locally installed Claude Code CLI version is compatible with the SDK version. The SDK launches the CLI under the hood, so a version mismatch between the two can cause silent failures.

Async Iterator Errors

Symptom: You get TypeError: 'async_generator' object is not iterable or similar errors.

terminal

$ python my_agent.py
TypeError: 'async_generator' object is not iterable

Cause: Using for instead of async for when iterating over query() results, or calling query() outside of an async context.

Fix:

python

# WRONG - synchronous for loop
for msg in query(prompt="hello", options=opts):
    print(msg)

# RIGHT - async for loop inside an async function
async def main():
    async for msg in query(prompt="hello", options=opts):
        if msg.type == "text":
            print(msg.text, end="")

asyncio.run(main())

Import Errors with Claude Code Not on PATH

Symptom: The SDK starts but immediately fails with a subprocess error about Claude Code not being found.

terminal

$ python pr_review_bot.py myorg/myapp 42
claude_code_sdk.errors.CLINotFoundError: Could not find 'claude' executable.
Ensure Claude Code CLI is installed and on your PATH.

Cause: The Python SDK launches Claude Code as a subprocess. If the claude command is not on your system PATH (common in virtual environments or containerized setups), the SDK cannot find it.

Fix:

bash

# Check if claude is on PATH
which claude

# If not found, install it globally
npm install -g @anthropic-ai/claude-code

# Or specify the path explicitly in your script
export CLAUDE_CODE_PATH="/usr/local/bin/claude"

Exercise

Build a log monitoring agent:

Watch a directory for new log file entries (tail -f style)
When new entries appear, analyze them for anomalous patterns (errors, timeouts, stack traces)
If anomalies are found, create a GitHub issue with:
- The anomalous log entries
- A preliminary root cause analysis
- Suggested files to investigate (by grepping the codebase for relevant function names)
Use subagents: one for log analysis, one for codebase correlation

Chapter Summary

The Agent SDK wraps Claude Code's orchestration loop into Python/TypeScript -- same tools, programmatic control
Subagents run in parallel for tasks like multi-perspective code review (security + logic + style)
Session resumption carries context across pipeline phases, cutting token usage by ~40%
Hooks give you audit trails and custom validation at the tool-call level
The Batches API provides 50% cost reduction for high-volume processing

Appendix links: A04 Agent Architecture Patterns covers ReAct, plan-and-execute, and multi-agent coordination patterns that underpin the SDK's design. A09 Extended Thinking explains when to enable thinking budgets and how they interact with tool calls.

Research reference: The Agent SDK's architecture -- exposing the same tool set across CLI, SDK, and Managed Agents -- reflects a deliberate design decision to minimize the gap between interactive development and production deployment. This is discussed in Anthropic's documentation on building agents that can be tested locally and deployed remotely without behavior changes.

Knowledge Check

What is the primary advantage of running three review subagents in parallel instead of one agent doing all three reviews sequentially?

It costs less because parallel agents share tokens

Each subagent has an isolated context, preventing one review domain from biasing another

Parallel agents use a different, faster model

It is required by the SDK -- you cannot do sequential reviews

In the migration pipeline demo, what does session resumption actually carry forward between phases?

The exact files Claude read, loaded back into the context window

A compacted summary of the previous session context, so Claude does not start from scratch

Only the session ID -- Claude re-reads everything from disk

The full uncompacted context from all previous phases

Why does the dependency auditor use allowed_bash_commands instead of giving Claude unrestricted Bash access?

Unrestricted Bash is slower due to permission checks

The allow-list constrains Claude to read-only audit commands, preventing it from modifying dependencies or running arbitrary code

The SDK does not support unrestricted Bash mode

It is a style preference with no functional difference

Next: Chapter 14: Managed Agents & Harness Architecture

Chapter 13: Agent SDK (Python / TypeScript) ​

What You Will Learn ​

What Is the Agent SDK? ​

Installation ​

Basic Usage ​

Demo 34: PR Review Bot ​

What We Are Building ​

The Code ​

Running It ​

What Happens ​

What Just Happened? ​

Demo 35: Dependency Auditor ​

What We Are Building ​

Running It ​

What Just Happened? ​

Demo 36: Multi-step Migration Pipeline ​

What We Are Building ​

Running It ​

What Just Happened? ​

Why Session Resumption Matters ​

SDK Patterns Reference ​

Extended Thinking ​

Hooks (Callbacks) ​

Subagent Definitions ​

Batches API ​

When Things Go Wrong ​

SDK Version Incompatibility ​

Async Iterator Errors ​

Import Errors with Claude Code Not on PATH ​

Exercise ​

Chapter Summary ​

Knowledge Check ​

Chapter 13: Agent SDK (Python / TypeScript)

What You Will Learn

What Is the Agent SDK?

Installation

Basic Usage

Demo 34: PR Review Bot

What We Are Building

The Code

Running It

What Happens

What Just Happened?

Demo 35: Dependency Auditor

What We Are Building

Running It

What Just Happened?

Demo 36: Multi-step Migration Pipeline

What We Are Building

Running It

What Just Happened?

Why Session Resumption Matters

SDK Patterns Reference

Extended Thinking

Hooks (Callbacks)

Subagent Definitions

Batches API

When Things Go Wrong

SDK Version Incompatibility

Async Iterator Errors

Import Errors with Claude Code Not on PATH

Exercise

Chapter Summary

Knowledge Check