Toolset API

create_subagent_toolset

subagents_pydantic_ai.create_subagent_toolset(subagents=None, default_model='openai:gpt-4.1', toolsets_factory=None, include_general_purpose=True, max_nesting_depth=0, id=None)

Create a toolset for delegating tasks to subagents.

This is the main entry point for using the subagent system. It creates a toolset with tools for: - task: Delegate a task to a subagent (sync or async) - check_task: Check status of an async task - answer_subagent: Answer a question from a subagent - list_active_tasks: List all running background tasks - soft_cancel_task: Request cooperative cancellation - hard_cancel_task: Immediately cancel a task

Parameters:

Name	Type	Description	Default
subagents	list[SubAgentConfig] | None	List of subagent configurations. If None, only general-purpose subagent will be available.	None
default_model	str	Default model for subagents that don't specify one.	'openai:gpt-4.1'
toolsets_factory	ToolsetFactory | None	Factory function that creates toolsets for subagents. Called with deps when running a task.	None
include_general_purpose	bool	Whether to include the default general-purpose subagent. Set to False if you want only specialized subagents.	True
max_nesting_depth	int	Maximum depth for nested subagents. 0 means subagents cannot spawn their own subagents.	0
id	str | None	Optional toolset ID. Defaults to "subagents".	None

Returns:

Type	Description
FunctionToolset[Any]	FunctionToolset configured with subagent management tools.

Example

Python

from pydantic_ai import Agent
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,
    default_model="openai:gpt-4.1",
)

agent = Agent("openai:gpt-4.1", toolsets=[toolset])

Source code in src/subagents_pydantic_ai/toolset.py

def create_subagent_toolset(
    subagents: list[SubAgentConfig] | None = None,
    default_model: str = "openai:gpt-4.1",
    toolsets_factory: ToolsetFactory | None = None,
    include_general_purpose: bool = True,
    max_nesting_depth: int = 0,
    id: str | None = None,
) -> FunctionToolset[Any]:
    """Create a toolset for delegating tasks to subagents.

    This is the main entry point for using the subagent system. It creates
    a toolset with tools for:
    - `task`: Delegate a task to a subagent (sync or async)
    - `check_task`: Check status of an async task
    - `answer_subagent`: Answer a question from a subagent
    - `list_active_tasks`: List all running background tasks
    - `soft_cancel_task`: Request cooperative cancellation
    - `hard_cancel_task`: Immediately cancel a task

    Args:
        subagents: List of subagent configurations. If None, only
            general-purpose subagent will be available.
        default_model: Default model for subagents that don't specify one.
        toolsets_factory: Factory function that creates toolsets for subagents.
            Called with deps when running a task.
        include_general_purpose: Whether to include the default general-purpose
            subagent. Set to False if you want only specialized subagents.
        max_nesting_depth: Maximum depth for nested subagents. 0 means
            subagents cannot spawn their own subagents.
        id: Optional toolset ID. Defaults to "subagents".

    Returns:
        FunctionToolset configured with subagent management tools.

    Example:
        ```python
        from pydantic_ai import Agent
        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,
            default_model="openai:gpt-4.1",
        )

        agent = Agent("openai:gpt-4.1", toolsets=[toolset])
        ```
    """
    # Build subagent configs
    configs: list[SubAgentConfig] = list(subagents) if subagents else []
    if include_general_purpose:
        configs.append(_create_general_purpose_config())

    # Compile subagents
    compiled: dict[str, CompiledSubAgent] = {}
    for config in configs:
        compiled[config["name"]] = _compile_subagent(config, default_model)

    # Create shared state
    message_bus = InMemoryMessageBus()
    task_manager = TaskManager(message_bus=message_bus)

    # Build available subagents description for tool docstring
    subagent_list = "\n".join(f"- {name}: {c.description}" for name, c in compiled.items())

    toolset: FunctionToolset[Any] = FunctionToolset(id=id or "subagents")

    @toolset.tool
    async def task(
        ctx: RunContext[SubAgentDepsProtocol],
        description: str,
        subagent_type: str,
        mode: ExecutionMode = "sync",
        priority: TaskPriority = TaskPriority.NORMAL,
        complexity: Literal["simple", "moderate", "complex"] | None = None,
        requires_user_context: bool = False,
        may_need_clarification: bool = False,
    ) -> str:
        f"""Delegate a task to a specialized subagent.

        Choose the appropriate subagent_type based on the task requirements.
        The task will be executed according to the specified mode.

        Available subagents:
        {subagent_list}

        Args:
            ctx: The run context with dependencies.
            description: Detailed description of the task to perform.
            subagent_type: Name of the subagent to use.
            mode: Execution mode - "sync" (blocking), "async" (background), or "auto".
            priority: Task priority level (for async tasks).
            complexity: Override complexity estimate ("simple", "moderate", "complex").
            requires_user_context: Whether task needs ongoing user interaction.
            may_need_clarification: Whether task might need clarifying questions.

        Returns:
            In sync mode: The subagent's response.
            In async mode: Task handle information with task_id.
        """
        # Validate subagent_type
        if subagent_type not in compiled:
            available = ", ".join(compiled.keys())
            return f"Error: Unknown subagent '{subagent_type}'. Available: {available}"

        subagent = compiled[subagent_type]
        config = subagent.config
        agent = subagent.agent

        if agent is None:
            return f"Error: Subagent '{subagent_type}' is not properly initialized"

        # Create deps for subagent
        parent_deps = ctx.deps
        subagent_deps = parent_deps.clone_for_subagent(max_nesting_depth - 1)

        # Apply toolsets from factory if provided
        if toolsets_factory:
            runtime_toolsets = toolsets_factory(subagent_deps)
            for ts in runtime_toolsets:
                agent._register_toolset(ts)  # type: ignore[attr-defined]

        # Generate task ID
        task_id = str(uuid.uuid4())[:8]

        # Resolve mode if "auto"
        if mode == "auto":
            characteristics = TaskCharacteristics(
                estimated_complexity=complexity or config.get("typical_complexity", "moderate"),
                requires_user_context=requires_user_context
                or config.get("typically_needs_context", False),
                may_need_clarification=may_need_clarification,
            )
            resolved_mode = decide_execution_mode(characteristics, config)
        else:
            resolved_mode = mode

        if resolved_mode == "sync":
            return await _run_sync(
                agent=agent,
                config=config,
                description=description,
                deps=subagent_deps,
                task_id=task_id,
            )
        else:
            return await _run_async(
                agent=agent,
                config=config,
                description=description,
                deps=subagent_deps,
                task_id=task_id,
                task_manager=task_manager,
                message_bus=message_bus,
                priority=priority,
            )

    @toolset.tool
    async def check_task(
        ctx: RunContext[SubAgentDepsProtocol],
        task_id: str,
    ) -> str:
        """Check the status of a background task.

        Use this to check if an async task has completed and get its result.

        Args:
            ctx: The run context.
            task_id: The task ID returned when the task was started.

        Returns:
            Status information and result if completed.
        """
        handle = task_manager.get_handle(task_id)
        if handle is None:
            return f"Error: Task '{task_id}' not found"

        status_info = [
            f"Task: {task_id}",
            f"Subagent: {handle.subagent_name}",
            f"Status: {handle.status}",
            f"Description: {handle.description}",
        ]

        if handle.status == TaskStatus.COMPLETED:
            status_info.append(f"Result: {handle.result}")
        elif handle.status == TaskStatus.FAILED:
            status_info.append(f"Error: {handle.error}")
        elif handle.status == TaskStatus.WAITING_FOR_ANSWER:
            status_info.append(f"Question: {handle.pending_question}")
        elif handle.started_at:
            elapsed = (datetime.now() - handle.started_at).total_seconds()
            status_info.append(f"Running for: {elapsed:.1f}s")

        return "\n".join(status_info)

    @toolset.tool
    async def answer_subagent(
        ctx: RunContext[SubAgentDepsProtocol],
        task_id: str,
        answer: str,
    ) -> str:
        """Answer a question from a subagent.

        When a background task is waiting for an answer (status: WAITING_FOR_ANSWER),
        use this tool to provide the requested information.

        Args:
            ctx: The run context.
            task_id: The task ID of the waiting subagent.
            answer: Your answer to the subagent's question.

        Returns:
            Confirmation that the answer was sent.
        """
        handle = task_manager.get_handle(task_id)
        if handle is None:
            return f"Error: Task '{task_id}' not found"

        if handle.status != TaskStatus.WAITING_FOR_ANSWER:
            return f"Error: Task '{task_id}' is not waiting for an answer (status: {handle.status})"

        # Find the pending question message and answer it
        try:
            # Create answer message
            answer_msg = AgentMessage(
                type=MessageType.ANSWER,
                sender="parent",
                receiver=handle.subagent_name,
                payload=answer,
                task_id=task_id,
            )
            await message_bus.send(answer_msg)

            # Update handle status
            handle.status = TaskStatus.RUNNING
            handle.pending_question = None

            return f"Answer sent to task '{task_id}'"
        except KeyError:
            return "Error: Could not send answer - subagent not available"

    @toolset.tool
    async def list_active_tasks(
        ctx: RunContext[SubAgentDepsProtocol],
    ) -> str:
        """List all active background tasks.

        Returns:
            List of active task IDs and their status.
        """
        active_ids = task_manager.list_active_tasks()

        if not active_ids:
            return "No active background tasks."

        lines = ["Active background tasks:"]
        for tid in active_ids:
            handle = task_manager.get_handle(tid)
            if handle:  # pragma: no branch
                desc = handle.description[:50]
                lines.append(f"- {tid}: {handle.subagent_name} ({handle.status}) - {desc}...")

        return "\n".join(lines)

    @toolset.tool
    async def soft_cancel_task(
        ctx: RunContext[SubAgentDepsProtocol],
        task_id: str,
    ) -> str:
        """Request cooperative cancellation of a background task.

        The subagent will be notified and can clean up before stopping.
        Use this for graceful cancellation.

        Args:
            ctx: The run context.
            task_id: The task to cancel.

        Returns:
            Confirmation or error message.
        """
        success = await task_manager.soft_cancel(task_id)
        if success:
            return f"Cancellation requested for task '{task_id}'"
        return f"Error: Task '{task_id}' not found"

    @toolset.tool
    async def hard_cancel_task(
        ctx: RunContext[SubAgentDepsProtocol],
        task_id: str,
    ) -> str:
        """Immediately cancel a background task.

        The task will be forcefully stopped. Use this only when soft
        cancellation doesn't work or immediate stopping is required.

        Args:
            ctx: The run context.
            task_id: The task to cancel.

        Returns:
            Confirmation or error message.
        """
        success = await task_manager.hard_cancel(task_id)
        if success:
            return f"Task '{task_id}' has been cancelled"
        return f"Error: Task '{task_id}' not found"

    return toolset

create_agent_factory_toolset

subagents_pydantic_ai.create_agent_factory_toolset(registry, allowed_models=None, default_model='openai:gpt-4.1', max_agents=10, toolsets_factory=None, capabilities_map=None, id=None)

Create a toolset for dynamic agent creation.

This toolset provides tools for creating, listing, and removing agents at runtime. Created agents are stored in the provided registry and can be used with the main subagent toolset.

Parameters:

Name	Type	Description	Default
registry	DynamicAgentRegistry	Registry to store created agents.	required
allowed_models	list[str] | None	List of allowed model names. If None, any model is allowed.	None
default_model	str	Default model to use when not specified.	'openai:gpt-4.1'
max_agents	int	Maximum number of dynamic agents allowed.	10
toolsets_factory	ToolsetFactory | None	Factory to create toolsets for new agents. Takes priority over capabilities if both are provided.	None
capabilities_map	dict[str, CapabilityFactory] | None	Mapping of capability names to factory functions. E.g., {"filesystem": create_fs_toolset, "todo": create_todo_toolset}. Used when capabilities are specified in create_agent.	None
id	str | None	Optional toolset ID. Defaults to "agent_factory".	None

Returns:

Type	Description
FunctionToolset[Any]	FunctionToolset with agent management tools.

Example

Python

from pydantic_ai import Agent
from subagents_pydantic_ai import (
    create_agent_factory_toolset,
    DynamicAgentRegistry,
)

registry = DynamicAgentRegistry()

# With capabilities map
factory_toolset = create_agent_factory_toolset(
    registry=registry,
    allowed_models=["openai:gpt-4.1", "openai:gpt-4o-mini"],
    max_agents=5,
    capabilities_map={
        "filesystem": lambda deps: [create_fs_toolset(deps.backend)],
        "todo": lambda deps: [create_todo_toolset()],
    },
)

agent = Agent("openai:gpt-4.1", toolsets=[factory_toolset])

Source code in src/subagents_pydantic_ai/factory.py

def create_agent_factory_toolset(
    registry: DynamicAgentRegistry,
    allowed_models: list[str] | None = None,
    default_model: str = "openai:gpt-4.1",
    max_agents: int = 10,
    toolsets_factory: ToolsetFactory | None = None,
    capabilities_map: dict[str, CapabilityFactory] | None = None,
    id: str | None = None,
) -> FunctionToolset[Any]:
    """Create a toolset for dynamic agent creation.

    This toolset provides tools for creating, listing, and removing
    agents at runtime. Created agents are stored in the provided
    registry and can be used with the main subagent toolset.

    Args:
        registry: Registry to store created agents.
        allowed_models: List of allowed model names. If None, any model
            is allowed.
        default_model: Default model to use when not specified.
        max_agents: Maximum number of dynamic agents allowed.
        toolsets_factory: Factory to create toolsets for new agents.
            Takes priority over capabilities if both are provided.
        capabilities_map: Mapping of capability names to factory functions.
            E.g., {"filesystem": create_fs_toolset, "todo": create_todo_toolset}.
            Used when capabilities are specified in create_agent.
        id: Optional toolset ID. Defaults to "agent_factory".

    Returns:
        FunctionToolset with agent management tools.

    Example:
        ```python
        from pydantic_ai import Agent
        from subagents_pydantic_ai import (
            create_agent_factory_toolset,
            DynamicAgentRegistry,
        )

        registry = DynamicAgentRegistry()

        # With capabilities map
        factory_toolset = create_agent_factory_toolset(
            registry=registry,
            allowed_models=["openai:gpt-4.1", "openai:gpt-4o-mini"],
            max_agents=5,
            capabilities_map={
                "filesystem": lambda deps: [create_fs_toolset(deps.backend)],
                "todo": lambda deps: [create_todo_toolset()],
            },
        )

        agent = Agent("openai:gpt-4.1", toolsets=[factory_toolset])
        ```
    """
    # Update registry max_agents
    registry.max_agents = max_agents

    # Format allowed models for docstring
    models_desc = (
        f"Allowed models: {', '.join(allowed_models)}" if allowed_models else "Any model is allowed"
    )

    # Format available capabilities for docstring
    caps_desc = (
        f"Available capabilities: {', '.join(capabilities_map.keys())}"
        if capabilities_map
        else "No predefined capabilities available"
    )

    toolset: FunctionToolset[Any] = FunctionToolset(id=id or "agent_factory")

    @toolset.tool
    async def create_agent(
        ctx: RunContext[SubAgentDepsProtocol],
        name: str,
        description: str,
        instructions: str,
        model: str | None = None,
        capabilities: list[str] | None = None,
        can_ask_questions: bool = True,
    ) -> str:
        f"""Create a new specialized agent at runtime.

        Creates a new agent with the specified configuration. The agent
        will be available for delegation via the task tool.

        {models_desc}
        {caps_desc}

        Args:
            ctx: The run context.
            name: Unique name for the agent (letters, numbers, hyphens only).
            description: Brief description of what the agent does.
            instructions: System prompt / instructions for the agent.
            model: Model to use (optional, defaults to {default_model}).
            capabilities: List of capability names to enable (e.g., ["filesystem", "todo"]).
            can_ask_questions: Whether agent can ask parent questions.

        Returns:
            Confirmation message or error.
        """
        # Validate name
        if not name or not all(c.isalnum() or c == "-" for c in name):
            return "Error: Name must contain only letters, numbers, and hyphens"

        if registry.exists(name):
            return f"Error: Agent '{name}' already exists"

        # Validate model
        actual_model = model or default_model
        if allowed_models and actual_model not in allowed_models:
            allowed = ", ".join(allowed_models)
            return f"Error: Model '{actual_model}' is not allowed. Use one of: {allowed}"

        # Validate capabilities
        if capabilities and capabilities_map:
            invalid_caps = [c for c in capabilities if c not in capabilities_map]
            if invalid_caps:
                available = ", ".join(capabilities_map.keys())
                invalid = ", ".join(invalid_caps)
                return f"Error: Unknown capabilities: {invalid}. Available: {available}"

        # Create config
        config = SubAgentConfig(
            name=name,
            description=description,
            instructions=instructions,
            model=actual_model,
            can_ask_questions=can_ask_questions,
        )

        # Create agent
        try:
            agent: Agent[Any, str] = Agent(
                actual_model,
                system_prompt=instructions,
            )

            # Apply toolsets from factory if provided (takes priority)
            if toolsets_factory:
                runtime_toolsets = toolsets_factory(ctx.deps)
                for ts in runtime_toolsets:
                    agent._register_toolset(ts)  # type: ignore[attr-defined]
            # Otherwise, apply toolsets from capabilities
            elif capabilities and capabilities_map:
                for cap_name in capabilities:
                    cap_factory = capabilities_map[cap_name]
                    cap_toolsets = cap_factory(ctx.deps)
                    for ts in cap_toolsets:
                        agent._register_toolset(ts)  # type: ignore[attr-defined]

            registry.register(config, agent)

            caps_info = f"\nCapabilities: {', '.join(capabilities)}" if capabilities else ""
            return (
                f"Agent '{name}' created successfully.\n"
                f"Model: {actual_model}\n"
                f"Description: {description}{caps_info}\n"
                f"Use task(description, '{name}') to delegate tasks."
            )

        except ValueError as e:
            return f"Error: {e}"
        except Exception as e:
            return f"Error creating agent: {e}"

    @toolset.tool
    async def list_agents(
        ctx: RunContext[SubAgentDepsProtocol],
    ) -> str:
        """List all dynamically created agents.

        Returns:
            List of agent names and descriptions.
        """
        return registry.get_summary()

    @toolset.tool
    async def remove_agent(
        ctx: RunContext[SubAgentDepsProtocol],
        name: str,
    ) -> str:
        """Remove a dynamically created agent.

        The agent will no longer be available for task delegation.

        Args:
            ctx: The run context.
            name: Name of the agent to remove.

        Returns:
            Confirmation or error message.
        """
        if registry.remove(name):
            return f"Agent '{name}' has been removed."
        return f"Error: Agent '{name}' not found."

    @toolset.tool
    async def get_agent_info(
        ctx: RunContext[SubAgentDepsProtocol],
        name: str,
    ) -> str:
        """Get detailed information about a dynamic agent.

        Args:
            ctx: The run context.
            name: Name of the agent.

        Returns:
            Agent details or error message.
        """
        config = registry.get_config(name)
        if config is None:
            return f"Error: Agent '{name}' not found."

        info = [
            f"Agent: {name}",
            f"Description: {config['description']}",
            f"Model: {config.get('model', default_model)}",
            f"Can ask questions: {config.get('can_ask_questions', True)}",
            "",
            "Instructions:",
            config["instructions"][:500] + ("..." if len(config["instructions"]) > 500 else ""),
        ]

        return "\n".join(info)

    return toolset

SubAgentToolset

subagents_pydantic_ai.SubAgentToolset = create_subagent_toolset module-attribute

get_subagent_system_prompt

subagents_pydantic_ai.get_subagent_system_prompt(configs, include_dual_mode=True)

Generate system prompt section describing available subagents.

Parameters:

Name	Type	Description	Default
configs	list[SubAgentConfig]	List of subagent configurations.	required
include_dual_mode	bool	Whether to include dual-mode execution explanation.	True

Returns:

Type	Description
str	Formatted system prompt section.

Example

Python

configs = [
    SubAgentConfig(
        name="researcher",
        description="Researches topics",
        instructions="...",
    ),
]
prompt = get_subagent_system_prompt(configs)

Source code in src/subagents_pydantic_ai/prompts.py

def get_subagent_system_prompt(
    configs: list[SubAgentConfig],
    include_dual_mode: bool = True,
) -> str:
    """Generate system prompt section describing available subagents.

    Args:
        configs: List of subagent configurations.
        include_dual_mode: Whether to include dual-mode execution explanation.

    Returns:
        Formatted system prompt section.

    Example:
        ```python
        configs = [
            SubAgentConfig(
                name="researcher",
                description="Researches topics",
                instructions="...",
            ),
        ]
        prompt = get_subagent_system_prompt(configs)
        ```
    """
    lines = ["## Available Subagents", ""]
    lines.append("You can delegate tasks to the following specialized subagents:")
    lines.append("")

    for config in configs:
        name = config["name"]
        description = config["description"]
        lines.append(f"### {name}")
        lines.append(description)

        # Add hint if agent cannot ask questions
        if config.get("can_ask_questions") is False:
            lines.append("*Cannot ask clarifying questions*")
        lines.append("")

    if include_dual_mode:
        lines.append(DUAL_MODE_SYSTEM_PROMPT)

    return "\n".join(lines)

get_task_instructions_prompt

subagents_pydantic_ai.get_task_instructions_prompt(task_description, can_ask_questions=True, max_questions=None)

Generate the task instructions for a subagent.

Parameters:

Name	Type	Description	Default
task_description	str	The task to perform.	required
can_ask_questions	bool	Whether the subagent can ask the parent questions.	True
max_questions	int | None	Maximum number of questions allowed.	None

Returns:

Type	Description
str	Formatted task instructions.

Source code in src/subagents_pydantic_ai/prompts.py

def get_task_instructions_prompt(
    task_description: str,
    can_ask_questions: bool = True,
    max_questions: int | None = None,
) -> str:
    """Generate the task instructions for a subagent.

    Args:
        task_description: The task to perform.
        can_ask_questions: Whether the subagent can ask the parent questions.
        max_questions: Maximum number of questions allowed.

    Returns:
        Formatted task instructions.
    """
    lines = ["## Your Task", "", task_description, ""]

    if can_ask_questions:
        lines.append("## Asking Questions")
        lines.append("If you need clarification, use the `ask_parent` tool.")
        if max_questions is not None:
            lines.append(f"You may ask up to {max_questions} questions.")
        lines.append("Keep questions specific and essential.")
    else:
        lines.append("## Note")
        lines.append("Complete this task using your best judgment.")
        lines.append("You cannot ask the parent for clarification.")

    return "\n".join(lines)

---

Usage Example

Python

from subagents_pydantic_ai import create_subagent_toolset, SubAgentConfig

# Define subagents
subagents = [
    SubAgentConfig(
        name="researcher",
        description="Researches topics",
        instructions="You research topics thoroughly.",
    ),
    SubAgentConfig(
        name="writer",
        description="Writes content",
        instructions="You write clear content.",
    ),
]

# Create toolset
toolset = create_subagent_toolset(
    subagents=subagents,
    default_model="openai:gpt-4o",
    max_nesting_depth=1,
)

# Add to agent
from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    deps_type=Deps,
    toolsets=[toolset],
)

With Toolsets Factory

Python

from pydantic_ai_backends import create_console_toolset

def my_toolsets_factory(deps):
    return [create_console_toolset()]

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

With Dynamic Agent Creation

Python

from subagents_pydantic_ai import (
    create_subagent_toolset,
    create_agent_factory_toolset,
    DynamicAgentRegistry,
)

registry = DynamicAgentRegistry()

agent = Agent(
    "openai:gpt-4o",
    deps_type=Deps,
    toolsets=[
        create_subagent_toolset(subagents=subagents),
        create_agent_factory_toolset(
            registry=registry,
            allowed_models=["openai:gpt-4o", "openai:gpt-4o-mini"],
            max_agents=5,
        ),
    ],
)