AI参谋长代理:使用Claude Code SDK构建全面代理的指南

本教程面向有编程基础的开发者,深入讲解如何使用Claude Code SDK构建企业级AI代理。通过一个为初创公司CEO服务的AI参谋长案例,系统介绍了CLAUDE.md记忆、Bash工具、输出风格、计划模式、斜杠命令、钩子和子代理等核心功能。教程不仅展示代码,还剖析了每个功能的设计原理、适用场景和架构决策,并提供了完整的实践步骤、常见错误排查和生产优化建议,帮助开发者掌握构建复杂多代理系统的工程实践。

技术背景

随着大语言模型能力的增强,开发者需要将AI集成到实际业务工作流中,而不仅仅是简单的对话。Claude Code SDK提供了一种编程式、无头的方式,使开发者能够构建自动化、可扩展的AI代理系统。本教程针对的痛点包括:如何管理代理的长期记忆、如何执行外部脚本、如何适应不同受众、如何安全地规划复杂任务、如何复用提示词、如何添加审计和护栏,以及如何实现多代理协作。这些功能解决了从原型到生产部署的关键问题,适用于需要自动化决策支持、数据分析、报告生成和合规审计的企业场景。主要限制包括:SDK的无头模式与交互式CLI的差异(如计划模式的退出机制)、文件系统设置加载的显式配置要求,以及代理对数据源的自然偏好可能影响上下文使用。

核心原理

Claude Code SDK的核心原理是通过编程接口封装Claude Code的全部能力,实现功能对等。代理通过ClaudeSDKClient与模型交互,支持流式响应和工具调用。关键机制包括:

架构分析

系统架构设计遵循模块化和可扩展原则,核心组件包括:

完整实战教程

from dotenv import load_dotenv
from utils.agent_visualizer import (
    display_agent_response,
    print_activity,
    reset_activity_context,
    visualize_conversation,
)

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient

load_dotenv()

# Define the model to use throughout this notebook
# Using Opus 4.6 for its superior planning and reasoning capabilities
MODEL = "claude-opus-4-6"
print(f"📋 Notebook configured to use: {MODEL}")

引言

在笔记本00中,我们构建了一个简单的研究代理。在本笔记本中,我们将逐步介绍Claude Code SDK的关键功能,用于构建全面的代理。对于每个引入的功能,我们将解释:

如果你熟悉Claude Code,你会注意到SDK带来了功能对等,并使你能够以编程的无头方式利用Claude Code的所有能力。

场景

在整个笔记本中,我们将为一家拥有50名员工、刚刚筹集了1000万美元A轮融资的初创公司构建一个AI参谋长。CEO需要数据驱动的洞察,以在积极增长与财务可持续性之间取得平衡。

我们最终的参谋长代理将:

基本特性

功能 0:通过 CLAUDE.md 实现记忆

是什么:CLAUDE.md 文件充当代理的持久记忆和指令。当项目目录中存在该文件时,Claude Code 会在初始化代理时自动读取并整合此上下文。

为什么:无需在每次交互中重复提供项目上下文、团队偏好或标准,你可以在 CLAUDE.md 中一次性定义它们。这确保了行为的一致性,并通过避免冗余解释来减少 token 使用量。

如何操作:

重要行为说明:当 CLAUDE.md 和详细数据文件(如 CSV)同时可用时,代理可能更倾向于读取更细粒度的数据源以提供精确答案。这是预期行为——代理自然会寻求权威数据。为确保代理使用高层的 CLAUDE.md 上下文,请使用明确的提示指令(参见下面的示例)。这教会我们一个重要经验:CLAUDE.md 提供的是*上下文和指导*,而非对数据源的硬性约束。

messages = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        cwd="chief_of_staff_agent",  # Points to subdirectory with our CLAUDE.md
        setting_sources=["project"],
    )
) as agent:
    await agent.query("What's our current runway?")
    async for msg in agent.receive_response():
        print_activity(msg)
        messages.append(msg)

# Display the response with HTML rendering
display_agent_response(messages)
# With this prompt, the agent should use CLAUDE.md values: ~$500K burn, 20 months runway

理解 Agent 的数据源偏好

刚才发生了什么:

通过在提示词中添加 ,我们引导 Agent 依赖 CLAUDE.md 上下文,而不是从 CSV 文件中寻找更细粒度的数据。

关键见解:

  1. CLAUDE.md 作为上下文,而非约束:当你设置 cwd 时,CLAUDE.md 文件会作为背景上下文加载。然而,Agent 会自然地寻找可用的最权威数据源。如果存在详细的 CSV 文件,Agent 可能会为了精确性而优先选择它们。
  1. 提示工程很重要:提示词中的“来自上下文的高层财务数字”向 Agent 表明,你希望从 CLAUDE.md 中获取简化的执行摘要($500K 消耗,20 个月跑道),而不是来自 financial_data/burn_rate.csv 的精确逐月数据($525K 总消耗,$235K 净消耗)。
  1. 架构设计选择:这种行为在生产系统中实际上是可取的——你希望 Agent 找到最佳数据源。CLAUDE.md 应包含:
  1. 现实世界模式:将 CLAUDE.md 视为“入职文档”,为 Agent 提供方向,而详细文件则是 Agent 在需要精确性时可以查询的“源系统”。

功能 1:用于执行 Python 脚本的 Bash 工具

是什么:Bash 工具允许你的智能体(除其他功能外)直接运行 Python 脚本,从而能够访问程序性知识、执行复杂计算、进行数据分析以及实现其他超出智能体原生能力的集成。

为什么:我们的 Chief of Staff 可能需要处理数据文件、运行财务模型或基于这些数据生成可视化。这些都是使用 Bash 工具的绝佳场景。

怎么做:将你的 Python 脚本放在智能体可以访问的位置,并添加一些上下文说明它们是什么以及如何调用。如果脚本是为你的 Chief of Staff 智能体准备的,请将此上下文添加到其 CLAUDE.md 文件中;如果脚本是为某个子智能体准备的,请将所述上下文添加到其 MD 文件中(稍后会有更多细节)。在本教程中,我们在 chief_of_staff_agent/scripts 中添加了五个示例:

  1. hiring_impact.py:计算新工程招聘如何影响烧钱率、跑道和现金状况。对于 financial-analyst 子智能体来说,这是模拟招聘场景与每月 50 万美元烧钱率和 20 个月跑道的关键。
  2. talent_scorer.py:使用加权标准对候选人的技术技能、经验、文化契合度和薪资期望进行评分。这是 recruiter 子智能体根据 TechStart 的高级工程师基准(18 万至 22 万美元)对工程候选人进行排名的核心工具。
  3. simple_calculation.py:执行快速财务计算,如跑道、烧钱率和季度指标。这是 Chief of Staff 无需复杂建模即可获得即时指标的实用脚本。
  4. financial_forecast.py:根据当前 240 万美元 ARR 以每月 15% 的速度增长,模拟 ARR 增长情景(基准/乐观/悲观)。对于 financial-analyst 预测 B 轮准备情况并验证 3000 万美元融资目标至关重要。
  5. decision_matrix.py:为战略选择(如 SmartDev 收购或办公室扩张)创建加权决策矩阵。帮助 Chief of Staff 系统性地评估涉及多个利益相关者和标准的复杂决策。
messages = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        allowed_tools=["Bash", "Read"],
        cwd="chief_of_staff_agent",  # Points to subdirectory where our agent is defined
    )
) as agent:
    await agent.query(
        "Use your simple calculation script with a total runway of 2904829 and a monthly burn of 121938."
    )
    async for msg in agent.receive_response():
        print_activity(msg)
        messages.append(msg)

# Display the response with HTML rendering
display_agent_response(messages)

功能 2:输出风格

是什么:输出风格允许你针对不同的受众使用不同的输出风格。每种风格都在一个 Markdown 文件中定义。

为什么:你的智能体可能会被不同专业水平的人使用,或者他们可能有不同的优先级。你的输出风格可以帮助区分这些群体,而无需创建单独的智能体。

如何操作:

重要提示:输出风格会修改 Claude Code 底层的系统提示,省略掉专注于软件工程的部分,并让你在软件工程工作之外针对特定用例获得更多控制。
SDK 配置说明:与斜杠命令(在功能 4 中介绍)类似,输出风格存储在文件系统的 .claude/output-styles/ 目录中。为了让 SDK 加载这些文件,你必须在 ClaudeAgentOptions 中包含 setting_sources=["project"]。settings 参数告诉 SDK 使用*哪种*风格,但 setting_sources 是实际*加载*风格定义所必需的。此要求是在调试后续部分时发现的,并且适用于所有基于文件系统的设置。
messages_executive = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        cwd="chief_of_staff_agent",
        settings='{"outputStyle": "executive"}',
        # IMPORTANT: setting_sources must include "project" to load output styles from .claude/output-styles/
        # Without this, the SDK does NOT load filesystem settings (output styles, slash commands, etc.)
        setting_sources=["project"],
    )
) as agent:
    await agent.query("Tell me in two sentences about your writing output style.")
    async for msg in agent.receive_response():
        print_activity(msg)
        messages_executive.append(msg)

messages_technical = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        cwd="chief_of_staff_agent",
        settings='{"outputStyle": "technical"}',
        setting_sources=["project"],
    )
) as agent:
    await agent.query("Tell me in two sentences about your writing output style.")
    async for msg in agent.receive_response():
        print_activity(msg)
        messages_technical.append(msg)
# Display executive style response
display_agent_response(messages_executive)
# Technical output style - detailed, implementation-focused
display_agent_response(messages_technical, title="Technical Style")

功能 3:计划模式——不执行操作的策略规划

是什么:计划模式指示代理创建详细的执行计划,而不执行任何操作。代理分析需求、提出解决方案并概述步骤,但不修改文件、执行命令或进行更改。

为什么:复杂的任务受益于预先规划,以减少错误、便于审查并改善协调。在规划阶段之后,代理将有一条贯穿执行过程的主线。

如何操作:只需设置 permission_mode="plan"

计划持久化:由于计划是用于审查和决策的宝贵工件,我们将演示如何捕获并将其保存为持久的 Markdown 文件。这使利益相关者能够在批准执行之前审查计划。

注意:此功能在 Claude Code 中表现出色,但仍需完全适配使用 SDK 的无头应用程序。也就是说,代理会尝试调用其 ExitPlanMode() 工具,该工具仅在交互模式下相关。在这种情况下,您可以发送后续查询,并设置 continue_conversation=True,让代理在上下文中执行其计划。
# =============================================================================
# Plan Mode Helper Functions
# =============================================================================
# These utilities handle the various ways an agent might output its plan.
# Since agents can output plans via direct text, Write tool, or Claude's
# internal plan directory, we need robust extraction from multiple sources.

import glob as glob_module
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any


def extract_plan_from_xml(text: str | None, min_length: int = 200) -> str | None:
    """
    Extract content between <plan> tags from text.

    Args:
        text: The text to search for plan content
        min_length: Minimum character count for valid plan (prevents empty matches)

    Returns:
        Extracted plan content, or None if not found/too short
    """
    if not text:
        return None
    match = re.search(r"<plan>(.*?)</plan>", text, re.DOTALL)
    if match:
        extracted = match.group(1).strip()
        if len(extracted) > min_length:
            return extracted
    return None


def extract_plan_from_messages(
    plan_content: list[str], min_fallback_length: int = 500
) -> tuple[str | None, str | None]:
    """
    Try to extract plan from captured message stream content.

    Args:
        plan_content: List of text blocks captured during streaming
        min_fallback_length: Minimum length for fallback (no XML tags)

    Returns:
        Tuple of (plan_text, source_description)
    """
    combined_text = "\n\n".join(plan_content)

    # First try: XML tags
    plan = extract_plan_from_xml(combined_text)
    if plan:
        return plan, "message stream"

    # Fallback: Use raw content if substantial
    if len(combined_text.strip()) > min_fallback_length:
        return combined_text.strip(), "full message content (fallback)"

    return None, None


def extract_plan_from_write_tool(
    write_contents: list[str], min_fallback_length: int = 500
) -> tuple[str | None, str | None]:
    """
    Try to extract plan from captured Write tool calls.

    Args:
        write_contents: List of content strings from Write tool calls
        min_fallback_length: Minimum length for fallback (no XML tags)

    Returns:
        Tuple of (plan_text, source_description)
    """
    for content in write_contents:
        # Try XML extraction first
        plan = extract_plan_from_xml(content)
        if plan:
            return plan, "Write tool capture"

        # Fallback: substantial content without tags
        if content and len(content.strip()) > min_fallback_length:
            return content.strip(), "Write tool capture (no XML tags)"

    return None, None


def extract_plan_from_claude_dir(
    max_age_seconds: int = 300, min_fallback_length: int = 500
) -> tuple[str | None, str | None]:
    """
    Check Claude's internal plan directory for recently created plans.

    Args:
        max_age_seconds: Maximum age of plan file to consider (default: 5 minutes)
        min_fallback_length: Minimum length for fallback (no XML tags)

    Returns:
        Tuple of (plan_text, source_description)
    """
    claude_plans_dir = os.path.expanduser("~/.claude/plans")

    if not os.path.exists(claude_plans_dir):
        return None, None

    # Find most recent plan file
    plan_files = sorted(
        glob_module.glob(os.path.join(claude_plans_dir, "*.md")),
        key=os.path.getmtime,
        reverse=True,
    )

    if not plan_files:
        return None, None

    most_recent = plan_files[0]
    file_age = datetime.now().timestamp() - os.path.getmtime(most_recent)

    if file_age > max_age_seconds:
        return None, None

    with open(most_recent) as f:
        content = f.read()

    filename = os.path.basename(most_recent)

    # Try XML extraction first
    plan = extract_plan_from_xml(content)
    if plan:
        return plan, f"Claude plan file ({filename})"

    # Fallback: substantial content without tags
    if len(content.strip()) > min_fallback_length:
        return content.strip(), f"Claude plan file ({filename}, no XML tags)"

    return None, None


def save_plan_to_file(
    plan_content: str,
    plan_source: str,
    model_name: str,
    prompt_summary: str,
    output_dir: str = "chief_of_staff_agent/plans",
    title: str = "Agent Plan: Engineering Restructure for AI Focus",
) -> Path:
    """
    Save extracted plan to a timestamped markdown file.

    Args:
        plan_content: The plan text to save
        plan_source: Description of where plan was extracted from
        model_name: The model used to generate the plan
        prompt_summary: Brief description of the original prompt
        output_dir: Directory to save plan files
        title: Title for the plan document

    Returns:
        Path to the saved plan file
    """
    plans_dir = Path(output_dir)
    plans_dir.mkdir(exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    plan_file = plans_dir / f"plan_{timestamp}.md"

    with open(plan_file, "w") as f:
        f.write(f"# {title}\n\n")
        f.write(f"**Created:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write(f"**Prompt:** {prompt_summary}\n")
        f.write(f"**Model:** {model_name}\n")
        f.write(f"**Plan Source:** {plan_source}\n\n")
        f.write("---\n\n")
        f.write(plan_content)
        f.write("\n\n---\n\n")
        f.write("*This plan was generated in plan mode and has not been executed.*\n")

    return plan_file


def capture_message_content(
    msg: Any,
    plan_content: list[str],
    write_tool_content: list[str],
    write_tool_paths: list[str],
) -> None:
    """
    Process a streaming message and capture relevant plan content.

    This function extracts content from three potential sources:
    1. Text blocks in message content
    2. Write tool call parameters
    3. Final result attribute

    Args:
        msg: The message object from the agent stream
        plan_content: List to append text content to
        write_tool_content: List to append Write tool content to
        write_tool_paths: List to append Write tool file paths to
    """
    # Source 1: Text blocks from message content
    if hasattr(msg, "content"):
        for block in msg.content:
            if hasattr(block, "text"):
                plan_content.append(block.text)

            # Source 2: Write tool calls
            if hasattr(block, "type") and block.type == "tool_use":
                if hasattr(block, "name") and block.name == "Write":
                    if hasattr(block, "input") and isinstance(block.input, dict):
                        if "content" in block.input:
                            write_tool_content.append(block.input["content"])
                        if "file_path" in block.input:
                            write_tool_paths.append(block.input["file_path"])

    # Source 3: Final result
    if hasattr(msg, "result") and msg.result:
        plan_content.append(msg.result)


print("✅ Plan Mode helper functions loaded")
# =============================================================================
# Plan Mode Configuration
# =============================================================================

# Note: MODEL is defined in cell-0 as "claude-opus-4-6"
# Opus excels at complex planning tasks

# The prompt is carefully crafted to:
# 1. Provide explicit context (since Opus prefers explicit information)
# 2. Request XML-tagged output for reliable extraction
# 3. Prevent file-writing so we can capture the plan programmatically

PLAN_PROMPT = """Restructure our engineering team for AI focus.

**CONTEXT (from CLAUDE.md):**
You are the Chief of Staff for TechStart Inc, a 50-person B2B SaaS startup that raised $10M Series A.
- Current engineering team: 25 people (Backend: 12, Frontend: 8, DevOps: 5)
- Monthly burn rate: ~$500K, Runway: 20 months
- Senior Engineer compensation: $180K-$220K + equity

**CRITICAL OUTPUT INSTRUCTIONS:**

1. **DO NOT use the Write tool** - Output your plan directly in your response text
2. **DO NOT save to any files** - I will handle saving the plan myself
3. **Wrap your ENTIRE plan inside `<plan> </plan>` XML tags** in your response

**Required Format:**
<plan>
[Your complete restructuring plan here - include all sections, timelines, budgets, and recommendations]
</plan>

**IMPORTANT:**
- The plan content MUST appear directly in your response between the XML tags
- Do NOT use Write, Edit, or any file-saving tools
- You may research and analyze before outputting, but the final plan must be in your response text
- Include: team structure, hiring recommendations, timeline, budget impact, and success metrics
- Use the company context provided above - do NOT ask clarifying questions"""

print(f"📋 Plan Mode configured with model: {MODEL}")
print(f"📝 Prompt length: {len(PLAN_PROMPT):,} characters")
# =============================================================================
# Execute Plan Mode Agent
# =============================================================================
# Run the agent with plan mode enabled. The agent will create a detailed plan
# but won't execute any actions. We capture content from multiple sources
# to handle different agent behaviors.

# Initialize capture lists
messages = []
plan_content = []  # Text from message stream
write_tool_content = []  # Content from Write tool calls
write_tool_paths = []  # Paths from Write tool calls

# Run the agent in plan mode
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        permission_mode="plan",
        cwd="chief_of_staff_agent",
    )
) as agent:
    await agent.query(PLAN_PROMPT)
    async for msg in agent.receive_response():
        print_activity(msg)
        messages.append(msg)

        # Capture content from this message
        capture_message_content(msg, plan_content, write_tool_content, write_tool_paths)

print(f"\n✅ Agent completed. Captured {len(plan_content)} content blocks.")
# =============================================================================
# Extract and Save the Plan
# =============================================================================
# Try multiple sources in priority order to find the plan content.
# This handles different agent behaviors robustly.

final_plan = None
plan_source = None

# Priority 1: Message stream (preferred - direct from agent response)
final_plan, plan_source = extract_plan_from_messages(plan_content)

# Priority 2: Write tool captures (if agent saved despite instructions)
if not final_plan and write_tool_content:
    final_plan, plan_source = extract_plan_from_write_tool(write_tool_content)

# Priority 3: Claude's internal plan directory (safety net)
if not final_plan:
    final_plan, plan_source = extract_plan_from_claude_dir()

# Report results
if final_plan:
    print(f"✅ Plan extracted from: {plan_source}")
    print(f"   Plan length: {len(final_plan):,} characters")

    # Save to file
    plan_file = save_plan_to_file(
        plan_content=final_plan,
        plan_source=plan_source,
        model_name=MODEL,
        prompt_summary="Restructure our engineering team for AI focus.",
    )
    print(f"\n📁 Plan saved to: {plan_file}")
else:
    error_msg = "Could not extract plan content from any source!\n"
    error_msg += "   Sources checked: message stream, Write tool, ~/.claude/plans/"
    if write_tool_paths:
        error_msg += f"\n   Write tool attempted to save to: {write_tool_paths}"
    print(f"❌ ERROR: {error_msg}")
    raise RuntimeError(f"Plan extraction failed: {error_msg}")
# Display the plan result with styled HTML
display_agent_response(messages, title="Engineering Restructure Plan")

执行已保存的计划

如上所述,代理将在创建计划后停止。保存的计划文件作为利益相关者的审查工件。

审查后执行计划:

  1. 在 chief_of_staff_agent/plans/plan_*.md 中查看已保存的计划
  2. 如果批准,发送带有 continue_conversation=True 的新查询,并移除 permission_mode="plan" 以执行

此工作流实现了“计划 → 审查 → 批准 → 执行”的循环,非常适合组织重组或重大基础设施变更等高风险的决策。

计划持久化如何工作

在上面的代码中,我们实现了一个健壮的多源计划捕获机制,用于处理 Plan 模式代理可能输出计划的各种方式:

挑战:

当使用 permission_mode="plan" 时,代理可能以不同方式输出计划:

  1. 消息流中的直接文本输出(理想情况)
  2. Write 工具保存到 ~/.claude/plans/(Claude 的内部计划系统)
  3. Write 工具保存到自定义路径

我们的捕获机制通过基于优先级的回退系统处理所有三种场景:

来源优先级(按顺序):

  1. 消息流(首选)
  1. Write 工具捕获
  1. Claude 的内部计划目录
  1. 完整内容回退

关键实现细节:

def extract_plan_from_text(text):
    """Extract content between <plan> tags, return None if not found or empty."""
    match = re.search(r'<plan>(.*?)</plan>', text, re.DOTALL)
    if match:
        extracted = match.group(1).strip()
        # Validate minimum content length (a real plan should be substantial)
        if len(extracted) > 200:
            return extracted
    return None

为什么内容验证很重要:

直接输出的提示工程:

提示明确指示代理:

这种方法为你提供:

让我们查看保存的计划:

# Display the saved plan with markdown rendering
from IPython.display import Markdown, display

# Show the plan with proper markdown formatting
display(Markdown(f"## 📋 Saved Plan Preview\n\n{final_plan}"))

print(f"\n📁 Full plan with metadata saved to: {plan_file}")

高级特性

功能 4:自定义斜杠命令

注意:斜杠命令对用户来说是语法糖,并非新的智能体能力

是什么:自定义斜杠命令是预定义的提示词模板,用户可以通过简写语法(例如 /budget-impact)触发。这些是面向用户的快捷方式,而非智能体能力。可以把它们想象成键盘快捷键,展开后变成完整、精心编写的提示词。

为什么:你的 Chief of Staff 将处理重复出现的高管问题。用户无需反复输入复杂的提示词,而是可以使用已经过验证的提示词。这提高了一致性和标准化程度。

怎么做:

关键 SDK 配置:使用 SDK 时,必须在 ClaudeAgentOptions 中设置 setting_sources=["project"],斜杠命令才能生效。默认情况下,SDK 以隔离模式运行,不会加载文件系统设置(斜杠命令、CLAUDE.md、子代理、钩子等)。这与交互式使用 Claude Code 时自动加载这些设置不同。
# User types: "/slash-command-test this is a test"
# -> behind the scenes EXPANDS to the prompt in .claude/commands/slash-command-test.md
# In this case the expanded prompt says to simply reverse the sentence word wise

messages = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        cwd="chief_of_staff_agent",
        # IMPORTANT: setting_sources must include "project" to load slash commands from .claude/commands/
        # Without this, the SDK does NOT load filesystem settings (slash commands, CLAUDE.md, etc.)
        setting_sources=["project"],
    )
) as agent:
    await agent.query("/slash-command-test this is a test")
    async for msg in agent.receive_response():
        print_activity(msg)
        messages.append(msg)
display_agent_response(messages, title="Slash Command Result")

功能 5:钩子 - 自动化确定性操作

是什么:钩子是 Python 脚本,你可以设置它们在特定工具调用之前(pre)或之后(post)等事件中自动执行。钩子以确定性方式运行,非常适合用于验证和审计跟踪。

为什么:想象一下这样的场景:你想确保你的代理有一些护栏(例如,防止危险操作),或者你想要一个审计跟踪。钩子与代理结合使用非常理想,既能给代理足够的自由来完成任务,同时又能确保代理以安全的方式行事。

如何操作:

SDK 配置说明:在 .claude/settings.local.json 中配置的钩子需要 setting_sources=["project", "local"]。SDK 区分三种设置来源:
- "project" → .claude/settings.json(版本控制,团队共享)
- "local" → .claude/settings.local.json(gitignored,本地设置,如钩子)
- "user" → ~/.claude/settings.json(全局用户设置)
由于我们的钩子在 settings.local.json 中,我们必须将 "local" 包含在 setting_sources 中。

示例:用于合规性的报告跟踪

一个钩子,用于记录对财务报告的写入/编辑操作,以供审计和合规目的。

该钩子定义在 chief_of_staff_agent/.claude/hooks/report-tracker.py 中,强制执行它的逻辑在 chief_of_staff_agent/.claude/settings.local.json 中:

"hooks": {
"PostToolUse": [
{
"matcher": "Write",
"hooks": [
{
"type": "command",
"command": "$CLAUDE_PROJECT_DIR/.claude/hooks/report-tracker.py"
}
]
},
{
"matcher": "Edit",
"hooks": [
{
"type": "command",
"command": "$CLAUDE_PROJECT_DIR/.claude/hooks/report-tracker.py"
}
]
}
]
}
messages = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        cwd="chief_of_staff_agent",
        allowed_tools=["Bash", "Write", "Edit", "MultiEdit"],
        # IMPORTANT: setting_sources must include BOTH "project" AND "local" to load hooks
        # - "project" loads .claude/settings.json (shared settings, CLAUDE.md, slash commands)
        # - "local" loads .claude/settings.local.json (where hooks are configured)
        setting_sources=["project", "local"],
    )
) as agent:
    await agent.query(
        "Create a quick Q2 financial forecast report with our current burn rate and runway projections. Save it to our /output_reports folder."
    )
    async for msg in agent.receive_response():
        print_activity(msg)
        messages.append(msg)

# The hook will track this in audit/report_history.json
display_agent_response(messages, title="Q2 Financial Forecast")

如果你现在导航到 ./chief_of_staff_agent/audit/report_history.json,你会发现它已经记录了该代理创建和/或修改报告的操作。生成的报告本身可以在 ./chief_of_staff_agent/output_reports/ 中找到。

功能 6:通过任务工具实现子代理

是什么:任务工具使您的代理能够将专业工作委派给其他子代理。这些子代理各自拥有自己的指令、工具和专业知识。

为什么:添加子代理开启了许多可能性:

  1. 专业化:每个子代理都是其领域的专家
  2. 独立上下文:子代理拥有自己的对话历史和工具
  3. 并行化:多个子代理可以同时处理不同方面的工作。

如何操作:

可视化增强:我们的 print_activity() 和 visualize_conversation() 工具已得到增强,可以清晰显示子代理操作:

# Reset the subagent tracking context before starting a new query
# This ensures clean state for activity display
reset_activity_context()

messages = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        allowed_tools=["Task"],  # this enables our Chief agent to invoke subagents
        system_prompt="Delegate financial questions to the financial-analyst subagent. Do not try to answer these questions yourself.",
        cwd="chief_of_staff_agent",
        setting_sources=["project", "local"],
    )
) as agent:
    await agent.query("Should we hire 5 engineers? Analyze the financial impact.")
    async for msg in agent.receive_response():
        print_activity(msg)
        messages.append(msg)

display_agent_response(messages, title="Hiring Impact Analysis")
visualize_conversation(messages)

在这里,当我们的主代理决定使用子代理时,它将:

  1. 调用 Task 工具,并传入如下参数:
  2. {
    "description": "分析招聘影响",
    "prompt": "分析招聘5名工程师的财务影响...",
    "subagent_type": "financial-analyst"
    }
    
  3. Task 工具在单独的上下文中执行子代理
  4. 将结果返回给主 Chief of Staff 代理以继续处理

综合运用

现在,让我们把之前看到的所有功能整合在一起。我们将让我们的智能体评估招聘3名高级工程师的财务影响,并将其见解写入 output_reports/hiring_decision.md。这将展示上述所有功能:

为了准备好这一切,我们将代理循环封装在一个 Python 文件中,类似于我们在上一个笔记本中所做的。请查看 chief_of_staff_agent 子目录中的 agent.py 文件。

总而言之,我们的 send_query() 函数接受4个参数(提示词、继续对话、权限模式和输出风格),其他所有内容都在代理文件中设置,即:系统提示词、最大轮数、允许的工具和工作目录。

为了更好地可视化这一切是如何组合在一起的,请查看这些 Claude 为我们制作的流程和架构图 :)

from chief_of_staff_agent.agent import send_query

reset_activity_context()

result, messages = await send_query(
    "/budget-impact hiring 3 senior engineers. Save your insights by updating the 'hiring_decision.md' file in /output_reports or creating a new file there",
    # permission_mode="plan", # Enable this to use planning mode
    output_style="executive",
)
visualize_conversation(messages)

结论

我们已经演示了 Claude Code SDK 如何使您能够构建具有企业级功能的复杂多智能体系统。从使用 Bash 工具的基本脚本执行开始,我们逐步引入了高级功能,包括使用 CLAUDE.md 的持久记忆、针对不同受众的自定义输出风格、战略规划模式、方便用户的斜杠命令、用于护栏的合规性钩子,以及用于专门任务的子智能体协调。

通过结合这些功能,我们创建了一个能够处理复杂高管决策工作流的 AI 幕僚长。该系统将财务分析委派给专门的子智能体,通过钩子维护审计跟踪,适应不同利益相关者的沟通风格,并提供由数据驱动分析支持的可操作见解。

这一高级智能体模式和多智能体编排的基础为您构建生产级企业系统做好了准备。在下一个笔记本中,我们将探讨如何通过模型上下文协议(MCP)服务器将我们的智能体连接到外部服务,从而大幅扩展其超越内置工具的能力。

下一步:02_The_observability_agent.ipynb - 了解如何通过 MCP 使用自定义集成和外部数据源扩展您的智能体。

常见问题

常见错误及排查方法:

  1. 斜杠命令或输出风格不生效:确保setting_sources包含"project",否则SDK不会加载文件系统设置。
  2. 钩子不触发:钩子配置在settings.local.json,需在setting_sources中同时包含"project"和"local"。
  3. 计划模式代理不退出:代理会调用ExitPlanMode()工具,在无头模式下需发送后续查询并设置continue_conversation=True来执行计划。
  4. 代理忽略CLAUDE.md而读取CSV:这是预期行为,代理偏好精确数据。可通过明确提示(如“使用来自上下文的高层财务数字”)引导。
  5. 计划提取失败:检查是否使用了XML标签,确保内容长度超过200字符;若代理使用Write工具,需捕获Write调用;检查~/.claude/plans/目录。
  6. 子代理未调用:确保allowed_tools包含"Task",并在系统提示中明确委派策略。
  7. 生产环境优化:

企业应用扩展

本教程展示了Claude Code SDK的核心功能,但生产级系统还需考虑以下扩展:

总结

本教程面向有编程基础的开发者,深入讲解如何使用Claude Code SDK构建企业级AI代理。通过一个为初创公司CEO服务的AI参谋长案例,系统介绍了CLAUDE.md记忆、Bash工具、输出风格、计划模式、斜杠命令、钩子和子代理等核心功能。教程不仅展示代码,还剖析了每个功能的设计原理、适用场景和架构决策,并提供了完整的实践步骤、常见错误排查和生产优化建议,帮助开发者掌握构建复杂多代理系统的工程实践。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程