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, registry=None, descriptions=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 | Model	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
descriptions	dict[str, str] | None	Optional mapping of tool name to custom description. Keys are tool names (task, check_task, answer_subagent, list_active_tasks, wait_tasks, soft_cancel_task, hard_cancel_task). When provided, the custom description replaces the built-in default.	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(  # noqa: C901
    subagents: list[SubAgentConfig] | None = None,
    default_model: str | Model = "openai:gpt-4.1",
    toolsets_factory: ToolsetFactory | None = None,
    include_general_purpose: bool = True,
    max_nesting_depth: int = 0,
    id: str | None = None,
    registry: Any | None = None,
    descriptions: dict[str, 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".
        descriptions: Optional mapping of tool name to custom description.
            Keys are tool names (task, check_task, answer_subagent,
            list_active_tasks, wait_tasks, soft_cancel_task, hard_cancel_task).
            When provided, the custom description replaces the built-in default.

    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])
        ```
    """
    _descs = descriptions or {}

    # 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 dynamic task description with available subagents
    subagent_list = "\n".join(f"- {name}: {c.description}" for name, c in compiled.items())
    task_description = _descs.get(
        "task",
        TASK_TOOL_DESCRIPTION.rstrip() + f"\n\nAvailable subagent types:\n{subagent_list}",
    )

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

    @toolset.tool(description=task_description)
    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:
        """Delegate a task to a specialized subagent.

        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.
        """
        # Validate subagent_type — check static compiled dict first, then dynamic registry
        if subagent_type in compiled:
            subagent = compiled[subagent_type]
        elif (
            registry is not None
            and hasattr(registry, "get_compiled")
            and registry.get_compiled(subagent_type)
        ):
            subagent = registry.get_compiled(subagent_type)
        else:
            # Build available list from both sources
            available_names = list(compiled.keys())
            if registry is not None and hasattr(registry, "list_agents"):
                available_names.extend(registry.list_agents())
            available = ", ".join(available_names)
            return f"Error: Unknown subagent '{subagent_type}'. Available: {available}"

        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)

        # Build runtime toolsets from factory if provided
        runtime_toolsets = toolsets_factory(subagent_deps) if toolsets_factory else None

        # 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,
                extra_toolsets=runtime_toolsets,
            )
        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,
                extra_toolsets=runtime_toolsets,
                priority=priority,
            )

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

        Args:
            ctx: The run context.
            task_id: The task ID returned when the task was started.
        """
        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(description=_descs.get("answer_subagent", ANSWER_SUBAGENT_DESCRIPTION))
    async def answer_subagent(
        ctx: RunContext[SubAgentDepsProtocol],
        task_id: str,
        answer: str,
    ) -> str:
        """Answer a question from a subagent.

        Args:
            ctx: The run context.
            task_id: The task ID of the waiting subagent.
            answer: Your answer to the subagent's question.
        """
        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})"

        # Resolve the answer future that ask_parent is waiting on
        future = task_manager.get_answer_future(task_id)
        if future is not None and not future.done():
            future.set_result(answer)
            return f"Answer sent to task '{task_id}'"

        return "Error: Could not send answer - subagent is no longer waiting"

    @toolset.tool(description=_descs.get("list_active_tasks", LIST_ACTIVE_TASKS_DESCRIPTION))
    async def list_active_tasks(
        ctx: RunContext[SubAgentDepsProtocol],
    ) -> str:
        """List all active background tasks."""
        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(description=_descs.get("wait_tasks", WAIT_TASKS_DESCRIPTION))
    async def wait_tasks(
        ctx: RunContext[SubAgentDepsProtocol],
        task_ids: list[str],
        timeout: float = 300.0,
    ) -> str:
        """Wait for multiple background tasks to complete.

        Args:
            ctx: The run context.
            task_ids: List of task IDs to wait for.
            timeout: Maximum seconds to wait (default 300s / 5 minutes).
        """
        # Collect asyncio.Task objects for the requested task_ids
        tasks_to_await: list[tuple[str, asyncio.Task[Any]]] = []
        for tid in task_ids:
            t = task_manager.tasks.get(tid)
            if t is not None and not t.done():
                tasks_to_await.append((tid, t))

        # Wait for all with timeout
        if tasks_to_await:
            aws = [t for _, t in tasks_to_await]
            try:
                await asyncio.wait_for(
                    asyncio.gather(*aws, return_exceptions=True),
                    timeout=timeout,
                )
            except asyncio.TimeoutError:
                pass  # Report what we have so far

        # Collect results
        lines: list[str] = []
        for tid in task_ids:
            handle = task_manager.get_handle(tid)
            if handle is None:
                lines.append(f"- {tid}: not found")
                continue
            status = handle.status
            if status == "completed":
                result_preview = (handle.result or "")[:2000]
                lines.append(f"- {tid} ({handle.subagent_name}): COMPLETED\n{result_preview}")
            elif status == "failed":
                lines.append(f"- {tid} ({handle.subagent_name}): FAILED - {handle.error}")
            else:
                lines.append(f"- {tid} ({handle.subagent_name}): {status}")

        return "Task results:\n" + "\n\n".join(lines)

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

        Args:
            ctx: The run context.
            task_id: The task to cancel.
        """
        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(description=_descs.get("hard_cancel_task", HARD_CANCEL_TASK_DESCRIPTION))
    async def hard_cancel_task(
        ctx: RunContext[SubAgentDepsProtocol],
        task_id: str,
    ) -> str:
        """Immediately cancel a background task.

        Args:
            ctx: The run context.
            task_id: The task to cancel.
        """
        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"

    # Expose task_manager for external monitoring (e.g., push notifications)
    toolset.task_manager = task_manager  # type: ignore[attr-defined]

    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 | Model	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 | Model = "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,
        )

        # Collect toolsets
        agent_toolsets: list[Any] = []
        if toolsets_factory:
            agent_toolsets.extend(toolsets_factory(ctx.deps))
        elif capabilities and capabilities_map:
            for cap_name in capabilities:
                cap_factory = capabilities_map[cap_name]
                agent_toolsets.extend(cap_factory(ctx.deps))

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

            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("Use the `task` tool to delegate work to these subagents:")
    lines.append("")

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

        # Add hint if agent cannot ask questions
        if config.get("can_ask_questions") is False:
            lines[-1] += " *(cannot ask clarifying questions)*"

    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 Custom Tool Descriptions

Override default tool descriptions for better LLM behavior:

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",
    },
)

Available tool names: task, check_task, answer_subagent, list_active_tasks, wait_tasks, soft_cancel_task, hard_cancel_task.

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,
        ),
    ],
)