Claude Agent SDK 实战指南：3 行 Python 搭建生产级 AI Agent

from claude_agent_sdk import query

async for message in query(prompt="Find and fix the bug in auth.py"):
    print(message)

# 传统方式：你来写 tool loop
response = client.messages.create(...)
while response.stop_reason == "tool_use":
    result = your_tool_executor(response.tool_use)  # 你实现
    response = client.messages.create(tool_result=result, **params)

# 推荐用 uv（更快的包管理器）
uv init my-agent && cd my-agent
uv add claude-agent-sdk

# 或者用传统 pip
python3 -m venv .venv && source .venv/bin/activate
pip install claude-agent-sdk

export ANTHROPIC_API_KEY=your-api-key

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ResultMessage


async def main():
    async for message in query(
        prompt="Review utils.py for bugs that would cause crashes. Fix any issues.",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Edit", "Glob"],
            permission_mode="acceptEdits",
        ),
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text"):
                    print(block.text)
        elif isinstance(message, ResultMessage):
            print(f"\n--- 完成 ---")
            print(f"耗时: {message.duration_ms}ms")
            print(f"花费: ${message.total_cost_usd:.4f}")


asyncio.run(main())

# ❌ 这样不行：第二次 query 不知道第一次读了什么
async for msg in query(prompt="Read auth.py"):
    pass
async for msg in query(prompt="Now find all callers of it"):
    pass  # Claude 不知道 "it" 是什么

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock


async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep", "Edit"],
        permission_mode="acceptEdits",
    )

    async with ClaudeSDKClient(options=options) as client:
        # 第一轮：读取并理解代码
        await client.query("Read the authentication module in src/auth/")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 第二轮：基于上一轮的理解继续分析
        await client.query("Find all places that call it and check for missing error handling")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 第三轮：执行修复
        await client.query("Fix the error handling issues you just found")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")


asyncio.run(main())

# 方式二：用 query() + resume 实现多轮
session_id = None

async for message in query(prompt="Read auth.py", options=options):
    if isinstance(message, SystemMessage) and message.subtype == "init":
        session_id = message.data["session_id"]

# 恢复上一个会话继续
async for message in query(
    prompt="Now find all callers",
    options=ClaudeAgentOptions(resume=session_id),
):
    print(message)

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    async for message in query(
        prompt="Open heyuan110.com and take a screenshot",
        options=ClaudeAgentOptions(
            mcp_servers={
                "playwright": {
                    "command": "npx",
                    "args": ["@playwright/mcp@latest"],
                }
            }
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

import asyncio
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, query


@tool("check_service_health", "Check if a service is running", {"service_name": str})
async def check_health(args):
    """检查服务健康状态——这里替换成你的真实逻辑"""
    service = args["service_name"]
    # 实际场景：调用 HTTP endpoint、检查进程、查询数据库
    return {
        "content": [
            {"type": "text", "text": f"Service '{service}' is running (response time: 42ms)"}
        ]
    }


@tool("restart_service", "Restart a service", {"service_name": str, "force": bool})
async def restart(args):
    """重启服务——生产环境中要加权限校验"""
    service = args["service_name"]
    force = args.get("force", False)
    return {
        "content": [
            {"type": "text", "text": f"Service '{service}' restarted (force={force})"}
        ]
    }


# 把自定义工具打包成 MCP Server
ops_server = create_sdk_mcp_server(
    name="ops-tools",
    version="1.0.0",
    tools=[check_health, restart],
)


async def main():
    async for message in query(
        prompt="Check if the auth service is healthy. If not, restart it.",
        options=ClaudeAgentOptions(
            mcp_servers={"ops": ops_server},
            allowed_tools=[
                "mcp__ops__check_service_health",
                "mcp__ops__restart_service",
            ],
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

# 只读分析 Agent：绝对安全
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep"],
)

# 代码修改 Agent：能编辑但不能执行命令
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Edit", "Glob", "Grep"],
)

# 全自动 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 audit_log(input_data, tool_use_id, context):
    """记录所有文件修改操作到审计日志"""
    file_path = input_data.get("tool_input", {}).get("file_path", "unknown")
    with open("./agent-audit.log", "a") as f:
        f.write(f"{datetime.now()}: {input_data.get('tool_name')} -> {file_path}\n")
    return {}


async def block_system_files(input_data, tool_use_id, context):
    """禁止修改系统目录文件"""
    file_path = input_data.get("tool_input", {}).get("file_path", "")
    if file_path.startswith("/etc/") or file_path.startswith("/system/"):
        return {"decision": "deny", "message": "System file modification blocked"}
    return {"decision": "allow"}


async def main():
    async for message in query(
        prompt="Refactor the project for better code quality",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Edit", "Glob", "Grep"],
            permission_mode="acceptEdits",
            hooks={
                "PreToolUse": [
                    HookMatcher(matcher="Edit|Write", hooks=[block_system_files])
                ],
                "PostToolUse": [
                    HookMatcher(matcher="Edit|Write", hooks=[audit_log])
                ],
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)

"""
code_reviewer.py — 自动代码审查 Agent
用法: python code_reviewer.py [目标目录]
"""

import asyncio
import sys
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ResultMessage


REVIEW_PROMPT = """You are a senior code reviewer. Analyze the code in the current directory and produce a structured review report.

## Review checklist:
1. **Security**: SQL injection, XSS, command injection, hardcoded secrets
2. **Error handling**: unhandled exceptions, missing edge cases
3. **Performance**: O(n²) loops, unnecessary DB queries, memory leaks
4. **Code quality**: dead code, duplicated logic, unclear naming

## Output format:
Create a file called `review-report.md` with:
- Executive summary (1 paragraph)
- Critical issues (must fix)
- Warnings (should fix)
- Suggestions (nice to have)
- Statistics (files scanned, issues found by category)

Be specific: include file paths, line numbers, and code snippets for every issue.
"""


async def main():
    target_dir = sys.argv[1] if len(sys.argv) > 1 else "."

    print(f"🔍 Scanning: {target_dir}")
    print("---")

    async for message in query(
        prompt=REVIEW_PROMPT,
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "Write"],
            permission_mode="acceptEdits",
            cwd=target_dir,
            max_turns=30,
            max_budget_usd=2.0,  # 硬限制：最多花 $2
        ),
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text"):
                    print(block.text)
                elif hasattr(block, "name"):
                    print(f"  [Tool] {block.name}")
        elif isinstance(message, ResultMessage):
            print(f"\n--- Review complete ---")
            print(f"Duration: {message.duration_ms / 1000:.1f}s")
            print(f"Cost: ${message.total_cost_usd:.4f}")
            print(f"Turns: {message.num_turns}")


if __name__ == "__main__":
    asyncio.run(main())

python code_reviewer.py ./src

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition


async def main():
    async for message in query(
        prompt="Review this codebase: use the security-auditor for vulnerabilities and the style-checker for code quality.",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "Agent"],
            agents={
                "security-auditor": AgentDefinition(
                    description="Security expert that finds vulnerabilities",
                    prompt="Find security issues: injection, XSS, hardcoded secrets, unsafe deserialization.",
                    tools=["Read", "Glob", "Grep"],
                ),
                "style-checker": AgentDefinition(
                    description="Code quality reviewer for style and maintainability",
                    prompt="Check code quality: naming conventions, dead code, excessive complexity.",
                    tools=["Read", "Glob", "Grep"],
                ),
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

options = ClaudeAgentOptions(
    max_budget_usd=5.0,     # 单次运行硬上限
    max_turns=50,            # 最大工具调用轮次
    model="claude-sonnet-4-6",  # Sonnet 比 Opus 便宜 10 倍
)

from claude_agent_sdk import (
    ClaudeSDKError,
    CLINotFoundError,
    ProcessError,
)

try:
    async for message in query(prompt="Fix the bug", options=options):
        if isinstance(message, ResultMessage) and message.is_error:
            print(f"Agent error: {message.result}")
            break
        print(message)
except CLINotFoundError:
    print("Error: Node.js 18+ required. Install it first.")
except ProcessError as e:
    print(f"Process failed (exit {e.exit_code}): {e.stderr}")
except ClaudeSDKError as e:
    print(f"SDK error: {e}")

options = ClaudeAgentOptions(
    permission_mode="bypassPermissions",  # 容器内可以放开
    sandbox={"type": "docker", "image": "node:18-slim"},
    cwd="/workspace",  # 容器内的工作目录
)