SDK 里的子智能体

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition


async def main():
    async for message in query(
        prompt="Review the authentication module for security issues",
        options=ClaudeAgentOptions(
            # 自动批准这些工具
            allowed_tools=["Read", "Grep", "Glob", "Agent"],
            agents={
                "code-reviewer": AgentDefinition(
                    # description 告诉 Claude 何时使用这个子智能体
                    description="Expert code review specialist. Use for quality, security, and maintainability reviews.",
                    # prompt 定义子智能体的行为和专长
                    prompt="""You are a code review specialist with expertise in security, performance, and best practices.

When reviewing code:
- Identify security vulnerabilities
- Check for performance issues
- Verify adherence to coding standards
- Suggest specific improvements

Be thorough but concise in your feedback.""",
                    # tools 限制子智能体能做什么（这里是只读）
                    tools=["Read", "Grep", "Glob"],
                    # model 覆盖该子智能体的默认模型
                    model="sonnet",
                ),
                "test-runner": AgentDefinition(
                    description="Runs and analyzes test suites. Use for test execution and coverage analysis.",
                    prompt="""You are a test execution specialist. Run tests and provide clear analysis of results.

Focus on:
- Running test commands
- Analyzing test output
- Identifying failing tests
- Suggesting fixes for failures""",
                    # Bash 访问让这个子智能体能运行测试命令
                    tools=["Bash", "Read", "Grep"],
                ),
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Review the authentication module for security issues",
  options: {
    // 自动批准这些工具
    allowedTools: ["Read", "Grep", "Glob", "Agent"],
    agents: {
      "code-reviewer": {
        // description 告诉 Claude 何时使用这个子智能体
        description:
          "Expert code review specialist. Use for quality, security, and maintainability reviews.",
        // prompt 定义子智能体的行为和专长
        prompt: `You are a code review specialist with expertise in security, performance, and best practices.

When reviewing code:
- Identify security vulnerabilities
- Check for performance issues
- Verify adherence to coding standards
- Suggest specific improvements

Be thorough but concise in your feedback.`,
        // tools 限制子智能体能做什么（这里是只读）
        tools: ["Read", "Grep", "Glob"],
        // model 覆盖该子智能体的默认模型
        model: "sonnet"
      },
      "test-runner": {
        description:
          "Runs and analyzes test suites. Use for test execution and coverage analysis.",
        prompt: `You are a test execution specialist. Run tests and provide clear analysis of results.

Focus on:
- Running test commands
- Analyzing test output
- Identifying failing tests
- Suggesting fixes for failures`,
        // Bash 访问让这个子智能体能运行测试命令
        tools: ["Bash", "Read", "Grep"]
      }
    }
  }
})) {
  if ("result" in message) console.log(message.result);
}

"Use the code-reviewer agent to check the authentication module"

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition


# 返回 AgentDefinition 的工厂函数
# 这种模式让你能按运行时条件定制智能体
def create_security_agent(security_level: str) -> AgentDefinition:
    is_strict = security_level == "strict"
    return AgentDefinition(
        description="Security code reviewer",
        # 按严格程度定制提示
        prompt=f"You are a {'strict' if is_strict else 'balanced'} security reviewer...",
        tools=["Read", "Grep", "Glob"],
        # 关键点：高风险审查用更强的模型
        model="opus" if is_strict else "sonnet",
    )


async def main():
    # 智能体在查询时创建，所以每个请求可以用不同的设置
    async for message in query(
        prompt="Review this PR for security issues",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Grep", "Glob", "Agent"],
            agents={
                # 用你想要的配置调用工厂
                "security-reviewer": create_security_agent("strict")
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

import { query, type AgentDefinition } from "@anthropic-ai/claude-agent-sdk";

// 返回 AgentDefinition 的工厂函数
// 这种模式让你能按运行时条件定制智能体
function createSecurityAgent(securityLevel: "basic" | "strict"): AgentDefinition {
  const isStrict = securityLevel === "strict";
  return {
    description: "Security code reviewer",
    // 按严格程度定制提示
    prompt: `You are a ${isStrict ? "strict" : "balanced"} security reviewer...`,
    tools: ["Read", "Grep", "Glob"],
    // 关键点：高风险审查用更强的模型
    model: isStrict ? "opus" : "sonnet"
  };
}

// 智能体在查询时创建，所以每个请求可以用不同的设置
for await (const message of query({
  prompt: "Review this PR for security issues",
  options: {
    allowedTools: ["Read", "Grep", "Glob", "Agent"],
    agents: {
      // 用你想要的配置调用工厂
      "security-reviewer": createSecurityAgent("strict")
    }
  }
})) {
  if ("result" in message) console.log(message.result);
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition, ToolUseBlock


async def main():
    async for message in query(
        prompt="Use the code-reviewer agent to review this codebase",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "Agent"],
            agents={
                "code-reviewer": AgentDefinition(
                    description="Expert code reviewer.",
                    prompt="Analyze code quality and suggest improvements.",
                    tools=["Read", "Glob", "Grep"],
                )
            },
        ),
    ):
        # 检查子智能体调用。两个名字都要匹配：较旧的 SDK
        # 版本发出 "Task"，当前版本发出 "Agent"。
        if hasattr(message, "content") and message.content:
            for block in message.content:
                if isinstance(block, ToolUseBlock) and block.name in (
                    "Task",
                    "Agent",
                ):
                    print(f"Subagent invoked: {block.input.get('subagent_type')}")

        # 检查这条消息是否来自子智能体的上下文内部
        if hasattr(message, "parent_tool_use_id") and message.parent_tool_use_id:
            print("  (running inside subagent)")

        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Use the code-reviewer agent to review this codebase",
  options: {
    allowedTools: ["Read", "Glob", "Grep", "Agent"],
    agents: {
      "code-reviewer": {
        description: "Expert code reviewer.",
        prompt: "Analyze code quality and suggest improvements.",
        tools: ["Read", "Glob", "Grep"]
      }
    }
  }
})) {
  const msg = message as any;

  // 检查子智能体调用。两个名字都要匹配：较旧的 SDK 版本
  // 发出 "Task"，当前版本发出 "Agent"。
  for (const block of msg.message?.content ?? []) {
    if (block.type === "tool_use" && (block.name === "Task" || block.name === "Agent")) {
      console.log(`Subagent invoked: ${block.input.subagent_type}`);
    }
  }

  // 检查这条消息是否来自子智能体的上下文内部
  if (msg.parent_tool_use_id) {
    console.log("  (running inside subagent)");
  }

  if ("result" in message) {
    console.log(message.result);
  }
}

import asyncio
import re
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition, ToolResultBlock

AGENTS = {
    "endpoint-finder": AgentDefinition(
        description="Locates and catalogs API endpoints in a codebase.",
        prompt="You find and document API endpoints. Report each endpoint's path, method, and handler.",
        tools=["Read", "Grep", "Glob"],
    )
}


def extract_agent_id(block: ToolResultBlock) -> str | None:
    """从 Agent 工具结果的文本内容里提取 agentId。"""
    parts = block.content if isinstance(block.content, list) else [{"text": block.content}]
    for part in parts:
        if match := re.search(r"agentId:\s*([\w-]+)", part.get("text") or ""):
            return match.group(1)
    return None


async def main():
    agent_id = None
    session_id = None

    # 第一次调用——运行 endpoint-finder 子智能体
    try:
        async for message in query(
            prompt="Use the endpoint-finder agent to find all API endpoints in this codebase",
            options=ClaudeAgentOptions(allowed_tools=["Read", "Grep", "Glob", "Agent"], agents=AGENTS),
        ):
            # 从 ResultMessage 捕获 session_id（恢复该会话需要）
            if hasattr(message, "session_id"):
                session_id = message.session_id

            # 在工具结果里搜索 agentId 尾部
            for block in getattr(message, "content", None) or []:
                if isinstance(block, ToolResultBlock):
                    agent_id = extract_agent_id(block) or agent_id

            # 打印最终结果
            if hasattr(message, "result"):
                print(message.result)
    except Exception as error:
        # 单次 query() 在产出错误结果之后抛出，
        # 所以 session_id 和 agent_id 已经被上面的循环捕获了。
        print(f"Session ended with an error: {error}")

    # 第二次调用——恢复并提追问
    if agent_id and session_id:
        async for message in query(
            prompt=f"Resume agent {agent_id} and list the top 3 most complex endpoints",
            options=ClaudeAgentOptions(
                allowed_tools=["Read", "Grep", "Glob", "Agent"], agents=AGENTS, resume=session_id
            ),
        ):
            if hasattr(message, "result"):
                print(message.result)
    else:
        print("No agentId found in the first query, so there is no subagent to resume.")


asyncio.run(main())

import { query, type SDKMessage } from "@anthropic-ai/claude-agent-sdk";

const agents = {
  "endpoint-finder": {
    description: "Locates and catalogs API endpoints in a codebase.",
    prompt: "You find and document API endpoints. Report each endpoint's path, method, and handler.",
    tools: ["Read", "Grep", "Glob"]
  }
};

// 把内容字符串化以搜索 agentId，而不必遍历嵌套的块类型
function extractAgentId(message: SDKMessage): string | undefined {
  if (message.type !== "assistant" && message.type !== "user") return undefined;
  const content = JSON.stringify(message.message.content);
  const match = content.match(/agentId:\s*([\w-]+)/);
  return match?.[1];
}

let agentId: string | undefined;
let sessionId: string | undefined;

// 第一次调用——运行 endpoint-finder 子智能体
try {
  for await (const message of query({
    prompt: "Use the endpoint-finder agent to find all API endpoints in this codebase",
    options: { allowedTools: ["Read", "Grep", "Glob", "Agent"], agents }
  })) {
    // 从 ResultMessage 捕获 session_id（恢复该会话需要）
    if ("session_id" in message) sessionId = message.session_id;

    // 在消息内容里搜索 agentId（出现在 Agent 工具结果里）
    const extractedId = extractAgentId(message);
    if (extractedId) agentId = extractedId;

    // 打印最终结果
    if ("result" in message) console.log(message.result);
  }
} catch (error) {
  // 单次 query() 在产出错误结果之后抛出，
  // 所以 sessionId 和 agentId 已经被上面的循环捕获了。
  console.error(`Session ended with an error: ${error}`);
}

// 第二次调用——恢复并提追问
if (agentId && sessionId) {
  for await (const message of query({
    prompt: `Resume agent ${agentId} and list the top 3 most complex endpoints`,
    options: { allowedTools: ["Read", "Grep", "Glob", "Agent"], agents, resume: sessionId }
  })) {
    if ("result" in message) console.log(message.result);
  }
} else {
  console.log("No agentId found in the first query, so there is no subagent to resume.");
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition


async def main():
    async for message in query(
        prompt="Analyze the architecture of this codebase",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Grep", "Glob", "Agent"],
            agents={
                "code-analyzer": AgentDefinition(
                    description="Static code analysis and architecture review",
                    prompt="""You are a code architecture analyst. Analyze code structure,
identify patterns, and suggest improvements without making changes.""",
                    # 只读工具：没有 Edit、Write 或 Bash 访问
                    tools=["Read", "Grep", "Glob"],
                )
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Analyze the architecture of this codebase",
  options: {
    allowedTools: ["Read", "Grep", "Glob", "Agent"],
    agents: {
      "code-analyzer": {
        description: "Static code analysis and architecture review",
        prompt: `You are a code architecture analyst. Analyze code structure,
identify patterns, and suggest improvements without making changes.`,
        // 只读工具：没有 Edit、Write 或 Bash 访问
        tools: ["Read", "Grep", "Glob"]
      }
    }
  }
})) {
  if ("result" in message) console.log(message.result);
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
    try:
        async for message in query(
            prompt="Audit every service in this repo for unhandled promise rejections",
            options=ClaudeAgentOptions(
                allowed_tools=["Read", "Grep", "Glob", "Agent"],
                # env 合并在继承的环境之上
                env={
                    "CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH": "1",
                    "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "5",
                },
                max_budget_usd=5.0,
            ),
        ):
            if isinstance(message, ResultMessage):
                print(f"{message.subtype}: ${message.total_cost_usd}")
    except Exception as error:
        # 单次 query() 在产出错误结果之后抛出，
        # 所以受预算限制的结果已经在上面打印了。
        print(f"Session ended with an error: {error}")


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

try {
  for await (const message of query({
    prompt: "Audit every service in this repo for unhandled promise rejections",
    options: {
      allowedTools: ["Read", "Grep", "Glob", "Agent"],
      // env 替换子进程环境，所以展开 process.env 以保留 PATH
      env: {
        ...process.env,
        CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH: "1",
        CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS: "5",
      },
      maxBudgetUsd: 5,
    },
  })) {
    if (message.type === "result") {
      console.log(`${message.subtype}: $${message.total_cost_usd}`);
    }
  }
} catch (error) {
  // 单次 query() 在产出错误结果之后抛出，
  // 所以受预算限制的结果已经在上面记录了。
  console.error(`Session ended with an error: ${error}`);
}