Claude Agent SDK 完全指南

# 基础 anthropic SDK：你得自己写循环处理工具调用
import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "修复 auth.py 的 bug"}],
    tools=[...]  # 你得自己定义工具
)
# Claude 说要调用某个工具
while response.stop_reason == "tool_use":
    result = your_tool_executor(response.tool_use)  # 你得自己执行
    response = client.messages.create(tool_result=result, **params)  # 你得自己喂回去

# Agent SDK：一行搞定，Claude 自己读文件、找 bug、改代码
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="修复 auth.py 的 bug",
    options=ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Bash"]),
):
    print(message)  # Claude 自己读文件、定位问题、修改代码

# Python
pip install claude-agent-sdk

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

export ANTHROPIC_API_KEY=your-api-key

from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Hello",
    options=ClaudeAgentOptions(
        env={
            "ANTHROPIC_BASE_URL": "https://your-proxy.example.com",
            "ANTHROPIC_API_KEY": "your-api-key",
        }
    ),
):
    print(message)

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    async for message in query(
        prompt="这个目录下有哪些文件？",
        options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"]),
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    session_id = None

    # 第一轮：读认证模块
    async for message in query(
        prompt="读一下认证模块的代码",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"]),
    ):
        if hasattr(message, "subtype") and message.subtype == "init":
            session_id = message.session_id

    # 第二轮：基于上下文继续工作
    async for message in query(
        prompt="找出所有调用它的地方",
        options=ClaudeAgentOptions(resume=session_id),
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

# 只读 agent：只能看，不能改
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep"],
    permission_mode="bypassPermissions"
)

# 全能 agent：能读能写能跑命令
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"]
)

from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher

# 每次文件被修改时，记录到审计日志
async def log_file_change(input_data, tool_use_id, context):
    file_path = input_data.get("tool_input", {}).get("file_path", "unknown")
    with open("./audit.log", "a") as f:
        f.write(f"{datetime.now()}: modified {file_path}\n")
    return {}

async def main():
    async for message in query(
        prompt="重构 utils.py 提升可读性",
        options=ClaudeAgentOptions(
            permission_mode="acceptEdits",
            hooks={
                "PostToolUse": [
                    HookMatcher(matcher="Edit|Write", hooks=[log_file_change])
                ]
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)

from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

async for message in query(
    prompt="用 code-reviewer agent 审查这个项目的代码质量",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep", "Task"],
        agents={
            "code-reviewer": AgentDefinition(
                description="专业代码审查员，负责质量和安全审查",
                prompt="分析代码质量，找出潜在问题并给出改进建议。",
                tools=["Read", "Glob", "Grep"],
            ),
            "test-writer": AgentDefinition(
                description="测试专家，负责编写单元测试",
                prompt="为缺少测试的函数编写单元测试。",
                tools=["Read", "Write", "Bash"],
            ),
        },
    ),
):
    if hasattr(message, "result"):
        print(message.result)

# 接入 Playwright，让 agent 能操作浏览器
async for message in query(
    prompt="打开 example.com 并描述你看到了什么",
    options=ClaudeAgentOptions(
        mcp_servers={
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"]
            }
        }
    ),
):
    if hasattr(message, "result"):
        print(message.result)

async for message in query(
    prompt="用户反馈登录时偶尔报 500 错误，请排查 src/auth/ 目录下的代码并修复",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash", "Glob", "Grep"],
        permission_mode="acceptEdits",
    ),
):
    print(message)

async for message in query(
    prompt="审查 src/ 目录下的代码，关注安全漏洞、性能问题和代码规范",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep"],
        permission_mode="bypassPermissions",
    ),
):
    if hasattr(message, "result"):
        print(message.result)

async for message in query(
    prompt="运行 npm test，分析失败的测试用例，修复代码使所有测试通过",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash", "Glob"],
        max_turns=20,
    ),
):
    print(message)

async for message in query(
    prompt="调研 2026 年主流的 Python Web 框架，对比 FastAPI、Django、Litestar，输出技术选型报告到 report.md",
    options=ClaudeAgentOptions(
        allowed_tools=["WebSearch", "WebFetch", "Write"],
    ),
):
    print(message)

async for message in query(
    prompt="修复首页的样式问题，然后打开浏览器截图确认效果",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash"],
        mcp_servers={
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"]
            }
        },
    ),
):
    print(message)

PR 提交
  │
  ▼
┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  代码审查     │───▶│  安全扫描     │───▶│  自动修复     │
│  Agent       │    │  Agent       │    │  Agent       │
│ (只读)       │    │ (只读)       │    │ (可写)       │
└─────────────┘    └─────────────┘    └─────────────┘
                                            │
                                            ▼
                                     ┌─────────────┐    ┌─────────────┐
                                     │  测试验证     │───▶│  报告生成     │
                                     │  Agent       │    │  Agent       │
                                     │ (Bash)       │    │ (Write)      │
                                     └─────────────┘    └─────────────┘
                                                              │
                                                              ▼
                                                        Slack 通知

import asyncio
import json
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher

# 审计日志：记录每个 agent 的每一步操作
audit_log = []

async def audit_hook(input_data, tool_use_id, context):
    audit_log.append({
        "time": datetime.now().isoformat(),
        "tool": input_data.get("tool_name"),
        "input": input_data.get("tool_input", {}),
    })
    return {}

# 通用 hook 配置：所有 agent 共享审计能力
audit_hooks = {
    "PostToolUse": [HookMatcher(matcher=".*", hooks=[audit_hook])]
}

async def run_code_review(pr_diff: str) -> str:
    """只读 agent，审查代码质量，输出结构化报告"""
    result_text = ""
    async for message in query(
        prompt=f"""审查以下 PR diff，从这几个维度分析：
1. 代码规范：命名、格式、注释
2. 逻辑问题：边界条件、空指针、竞态
3. 性能隐患：N+1 查询、内存泄漏、不必要的循环
4. 可维护性：函数过长、职责不清、魔法数字

PR Diff:
{pr_diff}

输出 JSON 格式：{{"issues": [{{"severity": "high/medium/low", "file": "...", "line": ..., "description": "..."}}], "summary": "..."}}""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep"],
            permission_mode="bypassPermissions",
            hooks=audit_hooks,
            max_turns=10,
        ),
    ):
        if hasattr(message, "result"):
            result_text = message.result
    return result_text

async def run_security_scan() -> str:
    """只读 agent，专注安全漏洞扫描"""
    result_text = ""
    async for message in query(
        prompt="""扫描项目代码中的安全漏洞：
1. SQL 注入、XSS、CSRF
2. 硬编码的密钥或凭证
3. 不安全的依赖版本
4. 权限校验缺失

输出 JSON：{{"vulnerabilities": [{{"severity": "critical/high/medium", "type": "...", "file": "...", "description": "...", "fix_suggestion": "..."}}]}}""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "Bash"],
            permission_mode="bypassPermissions",
            hooks=audit_hooks,
            max_turns=15,
        ),
    ):
        if hasattr(message, "result"):
            result_text = message.result
    return result_text

async def run_auto_fix(review_result: str, security_result: str) -> str:
    """可写 agent，根据审查和扫描结果自动修复代码"""
    result_text = ""
    async for message in query(
        prompt=f"""根据以下审查结果修复代码：

代码审查报告：
{review_result}

安全扫描报告：
{security_result}

修复规则：
1. 只修复 severity 为 high 或 critical 的问题
2. 每次修改后运行相关测试确认没有破坏现有功能
3. 不要重构无关代码，只做最小修复
4. 修复完成后输出修改文件列表""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Edit", "Bash", "Glob", "Grep"],
            permission_mode="acceptEdits",
            hooks=audit_hooks,
            max_turns=30,
        ),
    ):
        if hasattr(message, "result"):
            result_text = message.result
    return result_text

async def run_test_and_report(fix_result: str) -> str:
    """运行测试，生成最终报告"""
    result_text = ""
    async for message in query(
        prompt=f"""执行以下操作：
1. 运行完整测试套件（npm test 或 pytest）
2. 统计测试通过率
3. 生成 Markdown 格式的质量报告到 pr-report.md，包含：
   - 代码审查发现的问题数量和严重程度分布
   - 安全漏洞数量
   - 自动修复的内容：{fix_result}
   - 测试通过率
   - 最终结论：是否建议合并""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Bash", "Write", "Glob"],
            hooks=audit_hooks,
            max_turns=15,
        ),
    ):
        if hasattr(message, "result"):
            result_text = message.result
    return result_text

import subprocess

async def run_pipeline():
    """完整的 PR 质量守护流水线"""
    print("🔍 阶段 1/4：代码审查...")
    pr_diff = subprocess.run(
        ["git", "diff", "main...HEAD"], capture_output=True, text=True
    ).stdout
    review_result = await run_code_review(pr_diff)

    print("🛡️ 阶段 2/4：安全扫描...")
    security_result = await run_security_scan()

    print("🔧 阶段 3/4：自动修复...")
    fix_result = await run_auto_fix(review_result, security_result)

    print("✅ 阶段 4/4：测试验证 + 生成报告...")
    report = await run_test_and_report(fix_result)

    # 保存审计日志
    with open("audit-log.json", "w") as f:
        json.dump(audit_log, f, indent=2, ensure_ascii=False)

    print(f"流水线完成，审计日志已保存（共 {len(audit_log)} 条操作记录）")
    return report

asyncio.run(run_pipeline())

from claude_agent_sdk import query, CLINotFoundError, ProcessError

try:
    async for msg in query(prompt="分析代码"):
        print(msg)
except CLINotFoundError:
    print("Claude Code CLI 未安装，请先安装")
except ProcessError as e:
    print(f"进程异常退出，退出码: {e.exit_code}")