Toolset¶

The create_subagent_toolset() function creates a toolset that adds delegation capabilities to your Pydantic AI agent.

Creating a Toolset¶

Python

from subagents_pydantic_ai import create_subagent_toolset, SubAgentConfig

subagents = [
    SubAgentConfig(
        name="researcher",
        description="Researches topics",
        instructions="You are a research assistant.",
    ),
]

toolset = create_subagent_toolset(subagents=subagents)

Factory Parameters¶

Parameter	Type	Default	Description
`subagents`	`list[SubAgentConfig]`	`[]`	List of subagent configurations
`default_model`	`str \\| None`	`None`	Default model for subagents
`toolsets_factory`	`Callable`	`None`	Factory to create toolsets for subagents
`max_nesting_depth`	`int`	`2`	Maximum subagent nesting depth
`general_purpose_config`	`SubAgentConfig \\| None`	Auto	Config for the "general" subagent
`descriptions`	`dict[str, str] \\| None`	`None`	Override default tool descriptions by tool name

Adding to an Agent¶

Python

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    deps_type=Deps,
    toolsets=[toolset],
    system_prompt="You can delegate tasks to specialized subagents.",
)

Custom Agent Resolution¶

When the toolset compiles a SubAgentConfig, the internal _compile_subagent function decides which agent instance to use. The resolution priority is:

config["agent"] -- a pre-built agent instance, used as-is.
config["agent_factory"] -- a (config: SubAgentConfig) -> Agent callable, invoked at compile time.
Default -- a new pydantic_ai.Agent is created from the config's model, instructions, toolsets, and agent_kwargs fields.

This means you can mix pre-built agents, factory-built agents, and default agents in the same subagent list:

Python

from pydantic_ai import Agent
from subagents_pydantic_ai import create_subagent_toolset, SubAgentConfig

toolset = create_subagent_toolset(
    subagents=[
        SubAgentConfig(
            name="custom",
            description="Uses a pre-built agent",
            instructions="...",
            agent=my_prebuilt_agent,
        ),
        SubAgentConfig(
            name="factory-built",
            description="Built via factory",
            instructions="...",
            agent_factory=lambda cfg: build_agent(cfg),
        ),
        SubAgentConfig(
            name="default",
            description="Default Agent created automatically",
            instructions="...",
        ),
    ],
)

See SubAgentConfig for full documentation of the agent and agent_factory fields.

Available Tools¶

The toolset provides these tools to your agent:

task¶

Delegate a task to a subagent.

Python

# The agent calls:
task(
    description="Research Python async patterns",
    subagent_type="researcher",
    mode="sync",  # or "async" or "auto"
)

Parameters:

Parameter	Type	Description
`description`	`str`	What the subagent should do
`subagent_type`	`str`	Name of the subagent to use
`mode`	`str`	`"sync"`, `"async"`, or `"auto"`

check_task¶

Check the status of a background task.

Python

# The agent calls:
check_task(task_id="abc123")

Returns: Status, result (if complete), or pending question (if waiting).

answer_subagent¶

Answer a question from a blocked subagent.

Python

# The agent calls:
answer_subagent(task_id="abc123", answer="Use PostgreSQL for this project")

send_message_to_subagent¶

Steer a running async subagent mid-flight, without cancelling it. Use this when you learn something new while a long task is in progress and want to redirect or narrow it — the subagent keeps all its partial progress.

Python

# The agent calls:
send_message_to_subagent(
    task_id="abc123",
    message="narrow the search to packages/sparta/ — it isn't in core/",
)

Parameters:

Parameter	Type	Description
`task_id`	`str`	Task ID of the running async subagent
`message`	`str`	Steering instruction to deliver

The message is folded into the subagent's next model request as an extra user instruction, so it adapts on its next step. This is unprompted parent -> child steering, distinct from answer_subagent (which only replies to a question the subagent already asked via ask_parent). It applies to async tasks that are still running; messages to a finished or unknown task return an error.

Note

Steering is delivered at model-request boundaries on the retry-driven run path, which is the default (max_retries > 0). If you explicitly set max_retries=0 on a subagent, it runs via the legacy single-shot path and steering messages stay queued instead of being applied.

list_active_tasks¶

List all running background tasks.

Python

# The agent calls:
list_active_tasks()

Returns: List of task IDs, subagent names, and statuses.

wait_tasks¶

Wait for one or more background tasks to finish.

Parameters:

Parameter	Type	Default	Description
`task_ids`	`list[str]`	—	Task IDs to wait on
`timeout`	`float`	`300.0`	Max seconds to wait
`mode`	`"all" \\| "any"`	`"all"`	When to return

A task is considered "finished" when it is completed, failed, or cancelled.

`mode="all"` (default)¶

Block until every task in task_ids is finished, or the timeout fires. Use when you need every result together before the next step (e.g. a final synthesis across all subagents).

Python

# Wait for both research subagents before writing the report
result = wait_tasks(task_ids=["abc123", "def456"], mode="all")

`mode="any"` — reactive orchestration¶

Return as soon as ONE task finishes. Use when the subagents are independent and you can act on each result as it arrives — this avoids stalling on the slowest task.

Python

# Dispatch 4 independent research tasks in parallel
ids = [task(...) for _ in range(4)]

remaining = list(ids)
while remaining:
    # Return as soon as one finishes — don't wait on the slowest
    result = wait_tasks(task_ids=remaining, mode="any")
    # ... react to the finished task (synthesize, dispatch follow-up, etc.)
    remaining = [tid for tid in remaining if check_task(tid).is_running]

The output includes a header like Task results (mode=any, 1/4 finished, 3 still running): so the orchestrator can see which tasks are still in flight and decide whether to keep waiting or do other work first.

soft_cancel_task¶

Request cooperative cancellation.

Python

# The agent calls:
soft_cancel_task(task_id="abc123")

The subagent will receive a cancellation request and should stop gracefully.

hard_cancel_task¶

Immediately cancel a task.

Python

# The agent calls:
hard_cancel_task(task_id="abc123")

Forces immediate termination.

Toolsets Factory¶

Provide tools to your subagents:

Python

from pydantic_ai_backends import create_console_toolset

def my_toolsets_factory(deps):
    """Create toolsets for each subagent."""
    return [
        create_console_toolset(),  # File operations
    ]

toolset = create_subagent_toolset(
    subagents=subagents,
    toolsets_factory=my_toolsets_factory,
)

The factory is called for each subagent with cloned dependencies.

Nesting Depth¶

Control how deep subagents can nest:

Python

# Allow subagents to have their own subagents (2 levels deep)
toolset = create_subagent_toolset(
    subagents=subagents,
    max_nesting_depth=2,
)

# No nesting - subagents can't delegate
toolset = create_subagent_toolset(
    subagents=subagents,
    max_nesting_depth=0,
)

General Purpose Subagent¶

By default, a "general" subagent is added for tasks that don't match specific subagents:

Python

# Customize the general subagent
toolset = create_subagent_toolset(
    subagents=subagents,
    general_purpose_config=SubAgentConfig(
        name="general",
        description="Handles miscellaneous tasks",
        instructions="You are a general-purpose assistant.",
    ),
)

# Disable the general subagent
toolset = create_subagent_toolset(
    subagents=subagents,
    general_purpose_config=None,
)

Custom Tool Descriptions¶

Override the default tool descriptions to better guide LLM behavior. This is useful when you want descriptions that are more specific to your use case:

Python

toolset = create_subagent_toolset(
    subagents=subagents,
    descriptions={
        "task": "Assign a task to a specialized subagent",
        "check_task": "Check the status of a delegated task",
        "list_active_tasks": "Show all currently running background tasks",
    },
)

Only the tool names you include in the dictionary are overridden; the rest keep their built-in defaults. Available tool names:

Tool Name	Description
`task`	Delegate a task to a subagent
`check_task`	Check status of a background task
`answer_subagent`	Answer a question from a blocked subagent
`list_active_tasks`	List all running background tasks
`wait_tasks`	Wait for background tasks to complete
`soft_cancel_task`	Request cooperative cancellation
`hard_cancel_task`	Immediately cancel a task

System Prompt¶

Add context about available subagents to your agent's system prompt:

Python

from subagents_pydantic_ai import SubAgentConfig, get_subagent_system_prompt

configs = [
    SubAgentConfig(
        name="researcher",
        description="Researches topics and gathers information",
        instructions="You are a research assistant.",
    ),
    SubAgentConfig(
        name="writer",
        description="Writes content based on research",
        instructions="You are a writer.",
    ),
]

# Generate prompt listing available subagents
prompt = get_subagent_system_prompt(configs)

The get_subagent_system_prompt function takes a list of SubAgentConfig dicts (not deps) and an optional include_dual_mode flag. It generates text like:

Text Only

## Available Subagents

Use the `task` tool to delegate work to these subagents:

- **researcher**: Researches topics and gathers information
- **writer**: Writes content based on research

Subagents configured with can_ask_questions=False are annotated with (cannot ask clarifying questions).

Next Steps¶

Types - Data structures and enums
Execution Modes - Sync vs async
Examples - Working examples

Toolset¶

Creating a Toolset¶

Factory Parameters¶

Adding to an Agent¶

Custom Agent Resolution¶

Available Tools¶

task¶

check_task¶

answer_subagent¶

send_message_to_subagent¶

list_active_tasks¶

wait_tasks¶

mode="all" (default)¶

mode="any" — reactive orchestration¶

soft_cancel_task¶

hard_cancel_task¶

Toolsets Factory¶

Nesting Depth¶

General Purpose Subagent¶

Custom Tool Descriptions¶

System Prompt¶

Next Steps¶

`mode="all"` (default)¶

`mode="any"` — reactive orchestration¶