Chapter 13: Agent SDK (Python / TypeScript)

学习目标

Agent SDK 是什么，与 CLI 和 Managed Agents 的区别
构建一个 PR Review Bot：读取 diff、运行 subagent 分析、发布审查评论
构建一个依赖审计器：扫描 lockfile 查找过期和有漏洞的包
构建一个多步骤迁移流水线：使用 session 恢复
SDK 模式：工具使用、extended thinking、subagents、hooks 和 Batches API

Agent SDK 是什么？

Agent SDK 将 Claude Code 的编排循环（"harness"）封装成你可以在 Python 或 TypeScript 中导入的库。你的代码调用 query()，Claude 使用它在终端中拥有的同样工具——Read、Write、Edit、Bash、Glob、Grep——但在你的程序化控制下。

CLI:            你在终端输入        Claude 交互式运行
Agent SDK:      你的代码调用 query() Claude 在你的服务器上运行
Managed Agents: 你的 API 调用 REST  Claude 在 Anthropic 云端运行

何时使用：

需求	选择
交互式开发	CLI
将 Claude 嵌入你的应用、CI 脚本、内部工具	Agent SDK
长时间运行的云端任务、容错、零基础设施	Managed Agents

安装

bash

# Python
pip install claude-code-sdk

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

研究笔记：Agent SDK 暴露与 CLI 相同工具集的架构决策是有意为之——这意味着在终端中交互测试的 agent 行为可以直接转移到程序化使用中，缩小了开发与生产之间的差距。这种"相同工具、不同接口"的原则记录在 Anthropic 的 SDK 设计说明中。

基本用法

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())

预期输出：

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

我们在构建什么

一个 Python 脚本：

从 GitHub 获取 PR diff
并行启动三个 subagent：安全审查、逻辑审查、风格审查
汇总发现
在 PR 上发布结构化审查评论

这是一个真实的内部工具。Sentry 和 Notion 等公司的团队运行类似的配置。

代码

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])))

运行

bash

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

发生了什么

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

预期终端输出：

terminal

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

PR 上发布的评论效果：

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

刚才发生了什么？

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 with Tool-Enabled Scanning

Advanced~15 min

我们在构建什么

一个扫描项目依赖文件（package.json、requirements.txt、go.mod 等）的脚本，识别过期和有漏洞的包，生成可操作的报告。

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))

运行

bash

# Audit the current project
python dependency_auditor.py .

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

预期终端输出：

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

刚才发生了什么？

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: 多步骤迁移流水线

Three-Phase Migration with Session Resumption

Advanced~30 min

我们在构建什么

一个三阶段迁移流水线，将 JavaScript Express API 转换为 TypeScript。每个阶段运行在自己的 session 中，session 恢复携带上下文前进。

这个模式来自 HumanLayer 的 RPI（Research-Plan-Implement）方法，该方法通过将工作分解为离散阶段并在阶段间有意压缩上下文，成功处理了一个 300k LOC 的 Rust 代码库。

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))

运行

bash

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

预期终端输出（简略）：

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.

刚才发生了什么？

query() Phase 1

Project directory

↓

query() Phase 2

Resumed session from Phase 1

↓

query() Phase 3

Resumed session from Phase 2

为什么 Session 恢复重要

没有 session 恢复，每个阶段都从零开始——Claude 必须重新读取整个代码库才能理解阶段 2 该做什么。有了恢复：

阶段 1（研究）：读取 23 个文件，构建心智模型     Session: sess_abc
阶段 2（规划）：恢复 sess_abc，已经了解布局      无需重读
阶段 3（执行）：恢复 sess_abc，了解计划          立即开始

上下文效率：比三个独立会话节省约 40% 的 token。

SDK 模式参考

Extended Thinking

对于复杂的架构决策，启用 extended thinking 让 Claude 在回应前充分推理：

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,
)

附录链接：A09 Extended Thinking 涵盖何时使用 thinking、预算管理，以及 thinking 如何与工具调用交互。

Hooks（回调函数）

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 定义

定义主 agent 可以委派工作的专用 subagent：

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

处理大量项目（如审计 100 个 PR）时，使用 Batches API 可降低 50% 成本：

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}")

常见问题排查

SDK 版本不兼容

症状：你安装了 SDK 但遇到导入错误或意外行为。

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'

原因：SDK 包名和导入结构可能在版本间变化。旧版本可能使用不同的类名。

修复：

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')"

还要检查本地安装的 Claude Code CLI 版本是否与 SDK 版本兼容。SDK 在底层启动 CLI，所以两者之间的版本不匹配可能导致静默失败。

异步迭代器错误

症状：你遇到 TypeError: 'async_generator' object is not iterable 或类似错误。

terminal

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

原因：迭代 query() 结果时使用了 for 而不是 async for，或者在异步上下文之外调用了 query()。

修复：

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())

Claude Code 不在 PATH 中导致的导入错误

症状：SDK 启动但立即报子进程错误，找不到 Claude Code。

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.

原因：Python SDK 将 Claude Code 作为子进程启动。如果 claude 命令不在你的系统 PATH 中（在虚拟环境或容器化环境中很常见），SDK 就找不到它。

修复：

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"

练习

构建一个日志监控 Agent：

监控目录中的新日志条目（tail -f 风格）
出现新条目时，分析异常模式（错误、超时、堆栈追踪）
如果发现异常，创建 GitHub Issue 包含：
- 异常日志条目
- 初步根因分析
- 建议调查的文件（通过 grep 代码库中的相关函数名）
使用 subagent：一个做日志分析，一个做代码库关联

本章小结

Agent SDK 将 Claude Code 的编排循环封装到 Python/TypeScript 中——同样的工具，程序化控制
Subagent 并行运行用于多角度代码审查（安全 + 逻辑 + 风格）
Session 恢复在流水线阶段间携带上下文，减少约 40% 的 token 使用
Hooks 在工具调用层面提供审计追踪和自定义验证
Batches API 为大批量处理提供 50% 的成本降低

附录链接：A04 Agent 架构模式涵盖 ReAct、plan-and-execute 和多 agent 协调模式——SDK 设计的基础。A09 Extended Thinking 解释何时启用 thinking 预算以及它们如何与工具调用交互。

研究参考：Agent SDK 跨 CLI、SDK 和 Managed Agents 暴露相同工具集的架构——反映了一个有意的设计决策，旨在最小化交互式开发和生产部署之间的差距。这在 Anthropic 关于构建可以本地测试并远程部署且行为不变的 agent 的文档中有讨论。

知识检测

并行运行三个审查 subagent 而不是一个 agent 顺序执行三个审查的主要优势是什么？

更便宜因为并行 agent 共享 token

每个 subagent 有隔离的上下文，防止一个审查领域影响另一个的判断

并行 agent 使用不同的、更快的模型

SDK 要求这样做——你无法进行顺序审查

在迁移流水线 demo 中，session 恢复实际在阶段之间携带了什么？

Claude 读取的确切文件，重新加载到上下文窗口

上一个 session 上下文的压缩摘要，这样 Claude 不用从头开始

只有 session ID——Claude 重新从磁盘读取所有内容

所有之前阶段的完整未压缩上下文

为什么依赖审计器使用 allowed_bash_commands 而不是给 Claude 不受限制的 Bash 访问？

不受限制的 Bash 由于权限检查而更慢

allow 列表将 Claude 限制在只读审计命令中，防止它修改依赖或运行任意代码

SDK 不支持不受限制的 Bash 模式

这是一个风格偏好，没有功能差异

下一章：Chapter 14: Managed Agents 与 Harness 架构

Chapter 13: Agent SDK (Python / TypeScript) ​

学习目标 ​

Agent SDK 是什么？ ​

安装 ​

基本用法 ​

Demo 34: PR Review Bot ​

我们在构建什么 ​

代码 ​

运行 ​

发生了什么 ​

刚才发生了什么？ ​

Demo 35: 依赖审计器 ​

我们在构建什么 ​

运行 ​

刚才发生了什么？ ​

Demo 36: 多步骤迁移流水线 ​

我们在构建什么 ​

运行 ​

刚才发生了什么？ ​

为什么 Session 恢复重要 ​

SDK 模式参考 ​

Extended Thinking ​

Hooks（回调函数） ​

Subagent 定义 ​

Batches API ​

常见问题排查 ​

SDK 版本不兼容 ​

异步迭代器错误 ​

Claude Code 不在 PATH 中导致的导入错误 ​

练习 ​

本章小结 ​

知识检测 ​

Chapter 13: Agent SDK (Python / TypeScript)

学习目标

Agent SDK 是什么？

安装

基本用法

Demo 34: PR Review Bot

我们在构建什么

代码

运行

发生了什么

刚才发生了什么？

Demo 35: 依赖审计器

我们在构建什么

运行

刚才发生了什么？

Demo 36: 多步骤迁移流水线

我们在构建什么

运行

刚才发生了什么？

为什么 Session 恢复重要

SDK 模式参考

Extended Thinking

Hooks（回调函数）

Subagent 定义

Batches API

常见问题排查

SDK 版本不兼容

异步迭代器错误

Claude Code 不在 PATH 中导致的导入错误

练习

本章小结

知识检测