【Agent工坊】Claude Agent SDK 实战：用 Hooks + 子Agent 搭建生产级自动化流水线

# Python（推荐用 uv，Python 3.10+）

uv init && uv add claude-agent-sdk

# 或用传统 pip

pip install claude-agent-sdk

# TypeScript（Node.js 18+）

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

import asyncio

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ResultMessage

async def main():

    # query() 返回一个 async generator，逐个吐出消息

    async for message in query(

        prompt="读取 utils.py，检查潜在的崩溃点（如除零、空列表、None引用），修复所有发现的问题",

        options=ClaudeAgentOptions(

            allowed_tools=["Read", "Edit", "Glob"],

            permission_mode="acceptEdits", # 自动批准文件编辑，开发环境适用

            max_turns=20, # ⚠️ 最多 20 轮，防止死循环烧 API 额度

        ),

    ):

        # 处理两种消息类型

        if isinstance(message, AssistantMessage):

            for block in message.content:

                if hasattr(block, "text"):

                    print(f"💬 {block.text}") # Claude 的思考过程

                elif hasattr(block, "name"):

                    print(f"🔧 调用工具: {block.name}")

        elif isinstance(message, ResultMessage):

            print(f"✅ 任务完成: {message.subtype}")

            # subtype 可能是:

            # "success" — 正常完成

            # "error_max_turns" — 轮次耗尽（你设的 max_turns 到了）

            # "error_max_budget_usd" — 预算耗尽（你设的 max_budget_usd 到了）

asyncio.run(main())

💬 我来审查 utils.py 中的潜在崩溃点。先读取文件内容。

🔧 调用工具: Read

💬 我发现了两个问题：

    1. calculate_average 没有检查空列表，会触发 ZeroDivisionError

    2. get_user_name 对 None 调用 .upper()，会触发 AttributeError

    让我逐一修复。

🔧 调用工具: Edit

💬 已为 calculate_average 添加空列表检查。

🔧 调用工具: Edit

💬 已为 get_user_name 添加 None 检查。让我重新读取文件确认修复完整。

🔧 调用工具: Read

💬 所有修复已确认。文件现在包含完善的错误处理。

✅ 任务完成: success

from claude_agent_sdk import tool, create_sdk_mcp_server

import asyncio

@tool("check_stock", "查询指定股票在指定交易所的实时价格",

      {"symbol": str, "exchange": str})

async def check_stock(args: dict) -> dict:

    """

    这个函数模拟了一个真实的生产工具——调用第三方金融 API。

    """

    try:

        # 实际环境里这里调你的 REST API / gRPC 服务

        await asyncio.sleep(0.5) # 模拟网络延迟

        price = 187.35 # 模拟返回数据

        return {

            "content": [{

                "type": "text",

                "text": f"{args['symbol']}（{args['exchange']}）当前价: ¥{price}，涨跌幅: +2.3%"

            }]

        }

    except ConnectionError as e:

        # ⚠️ 这一行救你的生产线：

        # 返回 isError=True 而不是 raise

        return {

            "content": [{

                "type": "text",

                "text": f"API 连接失败: {str(e)}。建议操作: 使用本地缓存数据或切换到备用数据源。"

            }],

            "isError": True # ← 告诉 Agent Loop "这是个可恢复错误，请继续"

        }

    except ValueError as e:

        # 参数错误——告诉 Agent 修正输入

        return {

            "content": [{

                "type": "text",

                "text": f"参数错误: {str(e)}。请检查股票代码和交易所名称是否正确。"

            }],

            "isError": True

        }

# 把工具注册为 MCP Server

server = create_sdk_mcp_server(

    name="stock-service",

    version="1.0.0",

    tools=[check_stock]

)

# 在 Agent 中加载

options = ClaudeAgentOptions(

    mcp_servers={"stock": server},

    allowed_tools=[

        "Read", "Bash", "Glob",

        "mcp__stock__check_stock", # ← MCP 工具的命名规范

    ],

)

from claude_agent_sdk import (

    ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

)

async def protect_env(input_data, tool_use_id, context):

    """PreToolUse hook：在文件写入前拦截"""

    file_path = input_data["tool_input"].get("file_path", "")

    if file_path.endswith(".env"):

        return {

            "hookSpecificOutput": {

                "hookEventName": input_data["hook_event_name"],

                "permissionDecision": "deny",

                "permissionDecisionReason": "安全策略: 禁止修改 .env 文件"

            }

        }

    return {} # 空对象代表放行

options = ClaudeAgentOptions(

    hooks={

        "PreToolUse": [

            HookMatcher(

                matcher="Write|Edit", # 只在写文件操作时触发

                hooks=[protect_env]

            )

        ]

    }

)

async def audit_log(input_data, tool_use_id, context):

    """PostToolUse hook：记录每一个工具调用的审计信息"""

    tool_name = input_data.get("tool_name", "unknown")

    tool_input = input_data.get("tool_input", {})

    # 生产环境写入数据库或日志系统

    log_entry = {

        "tool": tool_name,

        "input_summary": str(tool_input)[:200], # 截断防过长

        "timestamp": context.get("timestamp"),

        "session_id": context.get("session_id"),

    }

    print(f"[AUDIT] {log_entry['timestamp']} | {tool_name} | {log_entry['input_summary'][:80]}")

    return {} # PostToolUse 不需要返回决策，它是纯观察者

options = ClaudeAgentOptions(

    hooks={

        "PostToolUse": [

            HookMatcher(matcher="Write|Edit|Bash", hooks=[audit_log])

        ]

    }

)

DANGEROUS_PATTERNS = [

    "rm -rf /", "DROP TABLE", "DROP DATABASE",

    "format C:", "dd if=/dev/zero",

    "> /dev/sda", "chmod 777 /",

]

async def block_dangerous_cmd(input_data, tool_use_id, context):

    command = input_data["tool_input"].get("command", "")

    for pattern in DANGEROUS_PATTERNS:

        if pattern in command:

            return {

                "hookSpecificOutput": {

                    "hookEventName": input_data["hook_event_name"],

                    "permissionDecision": "deny",

                    "permissionDecisionReason": f"安全拦截: 命令包含危险模式 '{pattern}'"

                }

            }

    return {}

options = ClaudeAgentOptions(

    hooks={

        "PreToolUse": [

            HookMatcher(matcher="Bash", hooks=[block_dangerous_cmd])

        ]

    }

)

async def main():

    async with ClaudeSDKClient(options=options) as client:

        # 主 Agent：分析 PR diff，把审查拆成并行子任务

        await client.query(

            "分析以下 PR 变更。把审查工作拆成 3 个并行子任务：\n"

            "1) 安全漏洞扫描（检查新增代码是否有 SQL 注入、XSS、路径遍历）\n"

            "2) 性能影响评估（循环复杂度、数据库查询次数、内存分配）\n"

            "3) 代码风格/最佳实践审查（命名规范、错误处理、类型安全）\n\n"

            "对每个子任务生成独立审查报告。"

        )

        async for msg in client.receive_response():

            if isinstance(msg, AssistantMessage):

                for block in msg.content:

                    if hasattr(block, "text"):

                        print(f"[主Agent] {block.text[:200]}")

            if isinstance(msg, ResultMessage):

                print(f"审查完成: {msg.subtype}")

asyncio.run(main())

results_store = {} # 简单的内存存储，生产环境用 Redis/DB

async def collect_subagent_result(input_data, tool_use_id, context):

    """SubagentStop 钩子：自动收集每个子Agent 的输出"""

    subagent_id = input_data.get("subagent_id", "unknown")

    result = input_data.get("output_summary", "")

    results_store[subagent_id] = {

        "summary": result[:500],

        "timestamp": context.get("timestamp"),

    }

    print(f"[汇总] 子Agent {subagent_id[:8]} 完成 | 输出长度: {len(result)} 字符")

    return {}

options = ClaudeAgentOptions(

    hooks={

        "SubagentStop": [HookMatcher(hooks=[collect_subagent_result])],

        "SubagentStart": [HookMatcher(hooks=[lambda *a: print(f"[启动] 子Agent 开始工作") and {}])],

    }

)