《AI Agent 高级编程》5.3 读 Claude Agent SDK：subagent 与 hook 调度（含 TS SDK 对照）

/tmp/agentbook/venv3/bin/python -c "import importlib.metadata as m; print(m.version('claude-agent-sdk'))"
/tmp/agentbook/venv3/bin/python -c "import claude_agent_sdk as c; print(c.__file__)"

0.2.165
/tmp/agentbook/venv3/lib/python3.14/site-packages/claude_agent_sdk/__init__.py

# claude_agent_sdk/query.py:11
async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]:
    """Query Claude Code for one-shot or unidirectional streaming interactions."""

# claude_agent_sdk/types.py:107
@dataclass
class AgentDefinition:
    """Agent definition configuration."""
    description: str
    prompt: str
    tools: list[str] | None = None
    disallowedTools: list[str] | None = None
    model: str | None = None
    skills: list[str] | None = None
    memory: Literal["user", "project", "local"] | None = None
    mcpServers: list[str | dict[str, Any]] | None = None
    initialPrompt: str | None = None
    maxTurns: int | None = None
    background: bool | None = None
    effort: EffortLevel | int | None = None
    permissionMode: PermissionMode | None = None

# claude_agent_sdk/types.py:2291
agents: dict[str, AgentDefinition] | None = None
"""Programmatically define custom subagents invokable via the Agent tool."""

# claude_agent_sdk/client.py:197
# Convert agents to dict format for initialize request
agents_dict: dict[str, dict[str, Any]] | None = None
if self.options.agents:
    agents_dict = {
        name: {k: v for k, v in asdict(agent_def).items() if v is not None}
        for name, agent_def in self.options.agents.items()
    }

# claude_agent_sdk/_internal/query.py:344
if self._agents:
    request["agents"] = self._agents

# claude_agent_sdk/types.py:605
@dataclass
class HookMatcher:
    matcher: str | None = None       # 工具名或 "Write|Edit" 这种模式
    hooks: list[HookCallback] = field(default_factory=list)
    timeout: float | None = None     # 该 matcher 内所有 hook 的超时（秒）

# claude_agent_sdk/_internal/query.py:318
hooks_config: dict[str, Any] = {}
if self.hooks:
    for event, matchers in self.hooks.items():
        if matchers:
            hooks_config[event] = []
            for matcher in matchers:
                callback_ids = []
                for callback in matcher.get("hooks", []):
                    callback_id = f"hook_{self.next_callback_id}"
                    self.next_callback_id += 1
                    self.hook_callbacks[callback_id] = callback
                    callback_ids.append(callback_id)
                hook_matcher_config: dict[str, Any] = {
                    "matcher": matcher.get("matcher"),
                    "hookCallbackIds": callback_ids,
                }
                ...
                hooks_config[event].append(hook_matcher_config)

# claude_agent_sdk/_internal/query.py:640
elif subtype == "hook_callback":
    hook_callback_request: SDKHookCallbackRequest = request_data
    callback_id = hook_callback_request["callback_id"]
    callback = self.hook_callbacks.get(callback_id)
    if not callback:
        raise Exception(f"No hook callback found for ID: {callback_id}")

    hook_output = await callback(
        request_data.get("input"),
        request_data.get("tool_use_id"),
        {"signal": None},  # TODO: Add abort signal support
    )
    response_data = _convert_hook_output_for_cli(hook_output)

# claude_agent_sdk/types.py:404
class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str

import inspect, dataclasses
import claude_agent_sdk as c

print(inspect.signature(c.query))
print([f.name for f in dataclasses.fields(c.AgentDefinition)])
print([f.name for f in dataclasses.fields(c.HookMatcher)])
opts = [f.name for f in dataclasses.fields(c.ClaudeAgentOptions)]
print('agents' in opts, 'hooks' in opts, 'can_use_tool' in opts)

(*, prompt: str | AsyncIterable[dict[str, Any]], options: ClaudeAgentOptions | None = None, transport: Transport | None = None) -> AsyncIterator[Message]
['description', 'prompt', 'tools', 'disallowedTools', 'model', 'skills', 'memory', 'mcpServers', 'initialPrompt', 'maxTurns', 'background', 'effort', 'permissionMode']
['matcher', 'hooks', 'timeout']
True True True

// node_modules/@anthropic-ai/claude-agent-sdk/sdk.d.ts:38
export declare type AgentDefinition = {
    description: string;
    tools?: string[];
    disallowedTools?: string[];
    prompt: string;
    model?: string;
    mcpServers?: AgentMcpServerSpec[];
    skills?: string[];
    ...
};

// node_modules/@anthropic-ai/claude-agent-sdk/sdk.d.ts:968
export declare type HookCallback = (input: HookInput, toolUseID: string | undefined, options: {
    signal: AbortSignal;
}) => Promise<HookJSONOutput>;

export declare interface HookCallbackMatcher {
    matcher?: string;
    hooks: HookCallback[];
    timeout?: number;
}

export declare type HookEvent = 'PreToolUse' | 'PostToolUse' | ... | 'SubagentStop' | ...;

import { query } from '@anthropic-ai/claude-agent-sdk';
import type {
  AgentDefinition, HookCallback, HookCallbackMatcher, Options,
} from '@anthropic-ai/claude-agent-sdk';

const auditBash: HookCallback = async (input, toolUseID) => {
  if (input.hook_event_name === 'PreToolUse') {
    console.log('hook sees', input.tool_name, 'toolUseID=', toolUseID);
  }
  return { continue: true };
};

const bashGuard: HookCallbackMatcher = {
  matcher: 'Bash',
  hooks: [auditBash],
  timeout: 10,
};

const reviewer: AgentDefinition = {
  description: 'Reviews a diff for correctness',
  prompt: 'You review code and report defects.',
  tools: ['Read', 'Grep'],
  model: 'haiku',
};

const options: Options = {
  hooks: { PreToolUse: [bashGuard] },
  agents: { reviewer },
};

const build = (prompt: string) => query({ prompt, options });
console.log(typeof build, Object.keys(options.agents ?? {}));

function [ 'reviewer' ]