SDK：首席幕僚 Agent
构建全面的个人助理 Agent
🌐 查看英文原文 | 源码 Notebook
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}")📋 Notebook configured to use: claude-opus-4-601 - 参谋长 Agent
简介
在 notebook 00 中，我们构建了一个简单的研究 Agent。在本 notebook 中，我们将逐步介绍构建综合 Agent 的关键 Claude Code SDK 功能。对于每个引入的功能，我们将解释： - 是什么：该功能是什么 - 为什么：该功能能做什么以及为什么要使用它 - 怎么做：展示如何使用它的最小实现
如果你熟悉 Claude Code，你会注意到 SDK 如何实现功能对等，并使你能够以编程方式利用 Claude Code 的所有功能。
场景
在本 notebook 中，我们将为一家刚筹集了 1000 万美元 A 轮融资的 50 人创业公司构建一个 AI 参谋长。CEO 需要数据驱动的洞察来平衡激进增长与财务可持续性。
我们最终的参谋长 Agent 将： - 为不同领域协调专业化子 Agent - 从多个来源聚合洞察 - 提供带有可操作建议的高管摘要
基础功能
功能 0：使用 CLAUDE.md 实现记忆
是什么：CLAUDE.md 文件作为 Agent 的持久记忆和指令。当存在于项目目录中时，Claude Code 会在初始化 Agent 时自动读取并纳入此上下文。
为什么：你可以在 CLAUDE.md 中一次性定义项目上下文、团队偏好或标准，而无需在每次交互中重复提供。这确保了行为的一致性，并通过避免冗余解释来减少 token 使用。
怎么做： - 在工作目录中放置 CLAUDE.md 文件——在我们的示例中：chief_of_staff_agent/CLAUDE.md - 将 ClaudeSDKClient 的 cwd 参数设置为指向你的 CLAUDE.md 文件所在目录 - 当你希望 Agent 优先使用高级上下文而非详细数据文件时，使用明确的提示来引导 Agent
重要行为说明：当 CLAUDE.md 和详细数据文件（如 CSV）同时可用时，Agent 可能倾向于读取更细粒度的数据源以提供精确答案。这是预期行为——Agent 自然会寻求权威数据。为确保 Agent 使用 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🤖 Thinking...<IPython.core.display.HTML object>理解 Agent 的数据源偏好
刚才发生了什么： 通过在提示中添加，我们引导 Agent 依赖 CLAUDE.md 上下文，而不是从 CSV 文件中寻求更细粒度的数据。
关键洞察： 1. CLAUDE.md 是上下文，而非约束：当你设置 cwd 时，CLAUDE.md 文件作为背景上下文加载。然而，Agent 会自然地寻求最权威的数据源。如果存在详细的 CSV 文件，Agent 可能会倾向于使用它们以获得精确度。
提示工程很重要：“来自上下文的高级财务数据”这一表述向 Agent 表明，你想要的是 CLAUDE.md 中的简化高管摘要（50 万美元消耗、20 个月资金 runway），而不是 financial_data/burn_rate.csv 中的精确逐月数据（53 万美元总消耗、23.5 万美元净消耗）。
架构设计选择：这种行为在生产系统中实际上是理想的——你希望 Agent 找到最佳数据源。CLAUDE.md 应包含：
- 高级上下文和策略
- 公司信息和标准
- 指向详细数据存放位置的指针
- 何时使用高级数据 vs. 详细数据的指南
现实世界模式：将 CLAUDE.md 视为帮助 Agent 定位的”入职文档”，而详细文件是 Agent 在需要精确数据时可以查询的”源系统”。
功能 1：用于 Python 脚本执行的 Bash 工具
是什么：Bash 工具允许你的 Agent（除其他功能外）直接运行 Python 脚本，从而获得对过程性知识、复杂计算、数据分析和其他集成的访问能力，这些超出了 Agent 的原生能力。
为什么：我们的参谋长可能需要处理数据文件、运行财务模型或基于这些数据生成可视化。这些都是使用 Bash 工具的好场景。
怎么做：将 Python 脚本放在 Agent 可以访问的位置，并添加一些关于它们是什么以及如何调用的上下文。如果脚本是为参谋长 Agent 准备的，将此上下文添加到其 CLAUDE.md 文件中；如果它们是为某个子 Agent 准备的，将所述上下文添加到对应的 MD 文件中（更多细节稍后介绍）。在本教程中，我们在 chief_of_staff_agent/scripts 中添加了五个示例： 1. hiring_impact.py：计算新工程师如何影响消耗率、runway 和现金状况。financial-analyst 子 Agent 用来根据 50 万美元月消耗和 20 个月 runway 建模招聘场景的必备工具。 2. talent_scorer.py：使用加权标准对候选人的技术技能、经验、文化契合度和薪资期望进行评分。recruiter 子 Agent 用以根据 TechStart 的 18-22 万美元高级工程师基准对工程候选人排名的核心工具。 3. simple_calculation.py：对 runway、消耗率和季度指标进行快速财务计算。参谋长无需复杂建模即可获取即时指标的实用脚本。 4. financial_forecast.py：在当前 240 万美元 ARR 以 15% 月环比增长的基础上建模 ARR 增长场景（基准/乐观/悲观）。financial-analyst 用来评估 B 轮融资准备情况并验证 3000 万美元融资目标的关键工具。 5. decision_matrix.py：为 SmartDev 收购或办公室扩张等战略选择创建加权决策矩阵。帮助参谋长系统性地评估涉及多个利益相关者和标准的复杂决策。
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)🤖 Using: Glob()
🤖 Using: Glob()
🤖 Using: Glob()
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: Read()
✓ Tool completed
🤖 Thinking...
🤖 Using: Bash()
✓ Tool completed
🤖 Thinking...<IPython.core.display.HTML object>功能 2：输出样式
是什么：输出样式允许你为不同的受众使用不同的输出样式。每种样式在 markdown 文件中定义。
为什么：你的 Agent 可能被不同专业水平的人使用，或者他们可能有不同的优先级。输出样式可以帮助区分这些细分群体，而无需创建单独的 Agent。
怎么做： - 在 chief_of_staff_agent/.claude/output-styles/ 中为每种样式配置一个 markdown 文件。例如，查看 .claude/output-styles/executive.md 中的高管输出样式。输出样式用包含两个字段（name 和 description）的简单 frontmatter 定义。注意：确保 frontmatter 中的 name 与文件名完全匹配（区分大小写）
重要：输出样式会修改 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)🤖 Thinking...
🤖 Thinking...# Display executive style response
display_agent_response(messages_executive)<IPython.core.display.HTML object># Technical output style - detailed, implementation-focused
display_agent_response(messages_technical, title="Technical Style")<IPython.core.display.HTML object>功能 3：计划模式 - 无需执行的策略规划
是什么：计划模式指示 Agent 创建详细的执行计划而不执行任何操作。Agent 分析需求、提出解决方案并列出步骤，但不修改文件、执行命令或进行更改。
为什么：复杂任务受益于前期规划，以减少错误、实现审查并改善协调。在规划阶段之后，Agent 将有一个主线贯穿其整个执行过程。
怎么做：只需设置 permission_mode="plan"
计划持久化：由于计划是审查和决策的宝贵产物，我们将演示如何捕获并将其保存为持久化 markdown 文件。这使得利益相关者可以在批准执行之前审查计划。
注意：此功能在 Claude Code 中表现出色，但仍然需要完全适应使用 SDK 的无头应用。即，Agent 会尝试调用其
ExitPlanMode()工具，该工具仅在交互模式下相关。在这种情况下，你可以发送一个带有continue_conversation=True的后续查询，让 Agent 在上下文中执行其计划。
# =============================================================================
# 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 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")📋 Plan Mode configured with model: claude-opus-4-6
📝 Prompt length: 1,180 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.")🤖 Thinking...
🤖 Using: ExitPlanMode()
✓ Tool completed
🤖 Thinking...
✅ Agent completed. Captured 3 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}")✅ Plan extracted from: message stream
Plan length: 6,783 characters
📁 Plan saved to: chief_of_staff_agent/plans/plan_20251204_152737.md# Display the plan result with styled HTML
display_agent_response(messages, title="Engineering Restructure Plan")<IPython.core.display.HTML object>执行已保存的计划
如上所述，Agent 在创建计划后会停止。保存的计划文件作为利益相关者的审查产物。
审查后执行计划： 1. 审查 chief_of_staff_agent/plans/plan_*.md 中保存的计划 2. 如果批准，发送一个带有 continue_conversation=True 的新查询，并移除 permission_mode="plan" 以执行
此工作流实现了”规划 → 审查 → 批准 → 执行”循环，非常适合组织重组或重大基础设施变更等高风险决策。
计划持久化的工作原理
在上面的代码中，我们实现了一个健壮的多来源计划捕获机制，可以处理计划模式 Agent 输出计划的各种方式：
挑战： 使用 permission_mode="plan" 时，Agent 可能以不同方式输出计划： 1. 消息流中的直接文本输出（理想情况） 2. 使用 Write 工具保存到 ~/.claude/plans/（Claude 的内部计划系统） 3. 使用 Write 工具保存到自定义路径
我们的捕获机制通过基于优先级的回退系统处理所有三种情况：
来源优先级（按顺序）：
- 消息流（首选）
- 在流式传输期间从
msg.content捕获文本块 - 提取
<plan></plan>XML 标签之间的内容 - 这是最干净的方法，因为内容直接来自响应
- 在流式传输期间从
- Write 工具捕获
- 监控消息流中的 Write 工具调用
- 提取正在写入的
content参数 - 当 Agent 尽管有提示指令仍决定保存时很有用
- Claude 的内部计划目录
- 检查
~/.claude/plans/中最近创建的计划文件（5 分钟内） - 读取并从最新文件中提取内容
- 在其他方法失败时作为安全保障
- 检查
- 完整内容回退
- 如果未找到 XML 标签但存在大量内容（>500 字符），直接使用它
- 防止生成空计划文件，同时保留部分信息
关键实现细节：
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为什么内容验证很重要： - 之前的版本如果在提取”成功”但没有内容时，可能会产生空计划文件 - 我们现在要求 XML 标签内容至少有 200 个字符 - 这防止了正则表达式匹配空内容或无意义内容时的误报
用于直接输出的提示工程： 提示明确指示 Agent： - 不要使用 Write 工具 - 防止文件系统绕行 - 直接在响应中输出 - 确保内容通过消息流传递 - 使用 XML 标签 - 从可能冗长的响应中实现清晰提取
这种方法为你提供： - 可靠性：无论 Agent 行为如何，计划都会被捕获 - 透明度：保存的文件指示使用了哪个来源 - 审计追踪：所有计划的历史记录，带有时间戳和来源元数据 - 调试：提取失败时有清晰的错误消息
让我们查看保存的计划：
# 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}")<IPython.core.display.Markdown object>
📁 Full plan with metadata saved to: chief_of_staff_agent/plans/plan_20251204_152737.md高级功能
功能 4：自定义斜杠命令
注意：斜杠命令是用户的语法糖，不是新的 Agent 能力
是什么：自定义斜杠命令是预定义的提示模板，用户可以用简写语法触发（例如 /budget-impact）。这些是面向用户的快捷方式，不是 Agent 能力。可以将它们想象为展开为完整、精心编写的提示的键盘快捷键。
为什么：你的参谋长将处理重复的高管问题。用户可以使用已经审查过的提示，而不是反复输入复杂的提示。这提高了一致性和标准化。
怎么做： - 在 .claude/commands/ 中定义一个 markdown 文件。例如，我们在 .claude/commands/slash-command-test.md 中定义了一个。注意命令是如何定义的：包含两个字段（name、description）的 frontmatter 和展开的提示，可以选择包含查询中传入的参数。 - 你可以使用 $ARGUMENTS（完整参数字符串）或 $1、$2 等（位置参数）为提示添加参数 - 用户在提示中使用斜杠命令
关键 SDK 配置：使用 SDK 时，你必须在
ClaudeAgentOptions中设置setting_sources=["project"]才能使斜杠命令工作。默认情况下，SDK 在隔离模式下运行，不会加载文件系统设置（斜杠命令、CLAUDE.md、子 Agent、hooks 等）。这与交互式使用 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)🤖 Thinking...display_agent_response(messages, title="Slash Command Result")<IPython.core.display.HTML object>功能 5：Hooks - 自动化的确定性操作
是什么：Hooks 是 Python 脚本，你可以设置它们在特定工具调用之前（pre）或之后（post）等事件中自动执行。Hooks 以确定性方式运行，使它们非常适合验证和审计追踪。
为什么：想象一些场景，你想确保你的 Agent 有一些护栏（例如，防止危险操作）或者你想拥有审计追踪。Hooks 与 Agent 配合理想，允许它们有足够的自由来完成任务，同时仍确保 Agent 以安全的方式运行。
怎么做： - 在 .claude/hooks/ 中定义 hook 脚本 -> _什么_行为应该在 hook 被触发时执行 - 在 .claude/settings.local.json 中定义 hook 配置 -> _何时_应该触发 hook - 在本例中，我们的 hooks 配置为监视特定工具调用（Bash、Write、Edit） - 当这些工具被调用时，hook 脚本在工具完成后运行（PostToolUse）
SDK 配置说明：在
.claude/settings.local.json中配置的 hooks 需要setting_sources=["project", "local"]。SDK 区分三种设置来源： -"project"→.claude/settings.json（版本控制、团队共享） -"local"→.claude/settings.local.json（gitignore、本地设置如 hooks） -"user"→~/.claude/settings.json（全局用户设置）由于我们的 hooks 在
settings.local.json中，我们必须在setting_sources中包含"local"。
示例：用于合规性的报告追踪
一个用于记录对财务报告的 Write/Edit 操作的 hook，用于审计和合规目的。 hook 定义在 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")🤖 Using: Bash()
🤖 Using: Bash()
🤖 Using: Bash()
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Using: Read()
🤖 Using: Read()
🤖 Using: Read()
🤖 Using: Bash()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: Write()
✓ Tool completed
🤖 Thinking...<IPython.core.display.HTML object>如果你现在导航到 ./chief_of_staff_agent/audit/report_history.json，你会发现它已经记录了 Agent 创建和/或修改了你的报告。生成的报告本身可以在 ./chief_of_staff_agent/output_reports/ 找到。
功能 6：通过 Task 工具实现子 Agent
是什么：Task 工具使你的 Agent 能够将专业工作委托给其他子 Agent。这些子 Agent 各自拥有自己的指令、工具和专业知识。
为什么：添加子 Agent 开启了很多可能性： 1. 专业化：每个子 Agent 是其领域的专家 2. 独立上下文：子 Agent 有自己的对话历史和工具 3. 并行化：多个子 Agent 可以同时处理不同方面的工作。
怎么做： - 将 "Task" 添加到 allowed_tools - 使用系统提示指示你的 Agent 如何委托任务（你也可以更通用地在 CLAUDE.md 中定义） - 在 .claude/agents/ 中为每个 Agent 创建一个 markdown 文件。例如，查看 .claude/agents/financial-analyst.md 并注意如何用如此简单直观的 markdown 文件定义（子）Agent：包含三个字段（name、description 和 tools）的 frontmatter 和其系统提示。description 对于主参谋长 Agent 了解何时调用每个子 Agent 很有用。
可视化增强：我们的 print_activity() 和 visualize_conversation() 工具已经增强，可以清晰地显示子 Agent 操作： - 🚀 表示正在委托给子 Agent（带有子 Agent 名称） - 📎 表示子 Agent 正在使用的工具（缩进以显示视觉层级） - 视觉分隔符清晰标记子 Agent 委托和完成边界 - 任务描述和提示显示在对话时间线中
# 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")🤖 Thinking...
🚀 Delegating to subagent: financial-analyst
└─ Task: Analyze 5 engineer hiring impact
✓ Tool completed
📎 [financial-analyst] Using: Bash()
📎 [financial-analyst] Using: Read()
📎 [financial-analyst] Using: Read()
✓ Tool completed
✓ Tool completed
✓ Tool completed
📎 [financial-analyst] Using: Read()
📎 [financial-analyst] Using: Bash()
✓ Tool completed
✓ Tool completed
📎 [financial-analyst] Using: Bash()
✓ Tool completed
📎 [financial-analyst] Using: Bash()
✓ Tool completed
📎 [financial-analyst] Using: Bash()
✓ Tool completed
📎 [financial-analyst] Using: Bash()
✓ Tool completed
✓ Tool completed
📎 [financial-analyst] Thinking...<IPython.core.display.HTML object>visualize_conversation(messages)<IPython.core.display.HTML object>在这里，当我们的主 Agent 决定使用子 Agent 时，它将： 1. 调用 Task 工具，参数如下： json { "description": "Analyze hiring impact", "prompt": "Analyze the financial impact of hiring 5 engineers...", "subagent_type": "financial-analyst" } 2. Task 工具在独立上下文中执行子 Agent 3. 将结果返回给主参谋长 Agent 继续处理
综合应用
现在让我们将看到的所有内容整合起来。我们将要求 Agent 确定招聘 3 名高级工程师的财务影响，并将其洞察写入 output_reports/hiring_decision.md。这演示了上面看到的所有功能： - Bash 工具：用于执行 hiring_impact.py 脚本来确定招聘新工程师的影响 - 记忆：读取目录中的 CLAUDE.md 作为上下文，以了解当前预算、runway、收入和其他相关信息 - 输出样式：不同的输出样式，定义在 chief_of_staff_agent/.claude/output-styles - 自定义斜杠命令：使用快捷方式 /budget-impact，它展开为 chief_of_staff_agent/.claude/commands 中定义的完整提示 - 子 Agent：我们的 /budget_impact 命令引导参谋长 Agent 调用 chief_of_staff_agent/.claude/agents 中定义的 financial-analyst 子 Agent - Hooks：Hooks 定义在 chief_of_staff_agent/.claude/hooks 中，并在 chief_of_staff_agent/.claude/settings.local.json 中配置 - 如果我们的某个 Agent 正在更新财务报告，hook 应该在 chief_of_staff_agent/audit/report_history.json 日志文件中记录此编辑/写入活动 - 如果财务分析子 Agent 调用 hiring_impact.py 脚本，这将被记录在 chief_of_staff_agent/audit/tool_usage_log.json 日志文件中
- 计划模式：如果你希望参谋长在采取任何行动之前制定一个计划供你批准，请取消注释下面的注释行
为了让这一切准备就绪，我们将 Agent 循环封装在一个 python 文件中，类似于我们在前一个 notebook 中所做的。请查看 chief_of_staff_agent 子目录中的 agent.py 文件。
总而言之，我们的 send_query() 函数接受 4 个参数（prompt、continue_conversation、permission_mode 和 output_style），其他所有内容都在 agent 文件中设置，即：系统提示、最大轮数、允许的工具和工作目录。
为了更好地可视化这一切是如何整合在一起的，请查看这些 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",
)🤖 Thinking...
🤖 Using: Glob()
🤖 Using: Glob()
✓ Tool completed
✓ Tool completed
🤖 Using: Read()
🤖 Using: Read()
🤖 Using: Read()
🤖 Using: Read()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: Task()
✓ Tool completed
🤖 Using: Bash()
🤖 Using: Read()
🤖 Using: Read()
🤖 Using: Read()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: Write()
✓ Tool completed
🤖 Thinking...visualize_conversation(messages)<IPython.core.display.HTML object>总结
我们演示了 Claude Code SDK 如何使你能够构建具有企业级功能的复杂多 Agent 系统。从使用 Bash 工具进行基本脚本执行开始，我们逐步引入了高级功能，包括使用 CLAUDE.md 的持久记忆、为不同受众定制的输出样式、策略规划模式、方便用户的斜杠命令、用于护栏的合规 hooks，以及用于专业化任务的子 Agent 协调。
通过结合这些功能，我们创建了一个能够处理复杂高管决策工作流的 AI 参谋长。该系统将财务分析委托给专业化子 Agent，通过 hooks 维护审计追踪，为不同利益相关者调整沟通风格，并提供由数据驱动分析支持的可操作洞察。
这一高级 Agent 模式和多 Agent 编排的基础为你构建生产就绪的企业系统做好了准备。在下一个 notebook 中，我们将探索如何通过 Model Context Protocol (MCP) 服务器将我们的 Agent 连接到外部服务，大幅扩展其超越内置工具的能力。
下一步：02_The_observability_agent.ipynb - 了解如何通过 MCP 扩展你的 Agent 的自定义集成和外部数据源。
中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。