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(
# Auto-approve these tools
allowed_tools=["Read", "Grep", "Glob", "Agent"],
agents={
"code-reviewer": AgentDefinition(
# description tells Claude when to use this subagent
description="Expert code review specialist. Use for quality, security, and maintainability reviews.",
# prompt defines the subagent's behavior and expertise
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 restricts what the subagent can do (read-only here)
tools=["Read", "Grep", "Glob"],
# model overrides the default model for this subagent
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 access lets this subagent run test commands
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: {
// Auto-approve these tools
allowedTools: ["Read", "Grep", "Glob", "Agent"],
agents: {
"code-reviewer": {
// description tells Claude when to use this subagent
description:
"Expert code review specialist. Use for quality, security, and maintainability reviews.",
// prompt defines the subagent's behavior and expertise
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 restricts what the subagent can do (read-only here)
tools: ["Read", "Grep", "Glob"],
// model overrides the default model for this subagent
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 access lets this subagent run test commands
tools: ["Bash", "Read", "Grep"]
}
}
}
})) {
if ("result" in message) console.log(message.result);
}

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"],
)
},
),
):
# Check for subagent invocation. Match both names: older SDK
# versions emitted "Task", current versions emit "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')}")

# Check if this message is from within a subagent's context
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;

// Check for subagent invocation. Match both names: older SDK versions
// emitted "Task", current versions emit "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}`);
}
}

// Check if this message is from within a subagent's context
if (msg.parent_tool_use_id) {
console.log("  (running inside subagent)");
// On assistant and user messages, agent_id matches the task_id
// on that subagent's task_started and other task events
if (msg.agent_id) {
console.log(`  agent_id: ${msg.agent_id}`);
}
}

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:
"""Extract agentId from an Agent tool result's text content."""
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

# First invocation - run the endpoint-finder subagent
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),
):
# Capture session_id from ResultMessage (needed to resume this session)
if hasattr(message, "session_id"):
session_id = message.session_id
# Search tool results for the agentId trailer
for block in getattr(message, "content", None) or []:
if isinstance(block, ToolResultBlock):
agent_id = extract_agent_id(block) or agent_id
# Print the final result
if hasattr(message, "result"):
print(message.result)
except Exception as error:
# A single-shot query() raises after yielding an error result,
# so session_id and agent_id have already been captured by the loop above.
print(f"Session ended with an error: {error}")

# Second invocation - resume and ask follow-up
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"]
}
};

// Stringify content to search for agentId without traversing nested block types
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;

// First invocation - run the endpoint-finder subagent
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 }
})) {
// Capture session_id from ResultMessage (needed to resume this session)
if ("session_id" in message) sessionId = message.session_id;
// Search message content for the agentId (appears in Agent tool results)
const extractedId = extractAgentId(message);
if (extractedId) agentId = extractedId;
// Print the final result
if ("result" in message) console.log(message.result);
}
} catch (error) {
// A single-shot query() throws after yielding an error result,
// so sessionId and agentId have already been captured by the loop above.
console.error(`Session ended with an error: ${error}`);
}

// Second invocation - resume and ask follow-up
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.""",
# Read-only tools: no Edit, Write, or Bash access
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.`,
// Read-only tools: no Edit, Write, or Bash access
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 is merged on top of the inherited environment
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:
# A single-shot query() raises after yielding an error result,
# so the budget-capped result has already been printed above.
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 replaces the subprocess environment, so spread process.env to keep 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) {
// A single-shot query() throws after yielding an error result,
// so the budget-capped result has already been logged above.
console.error(`Session ended with an error: ${error}`);
}