Skip to content

SubAgentCapability API

subagents_pydantic_ai.capability

Subagent capability for pydantic-ai agents.

Provides a SubAgentCapability that integrates the subagent toolset + dynamic instructions via the pydantic-ai capabilities API.

Example
Python
from pydantic_ai import Agent
from subagents_pydantic_ai import SubAgentCapability, SubAgentConfig

agent = Agent(
    "openai:gpt-4.1",
    capabilities=[SubAgentCapability(
        default_model="openai:gpt-4.1",
        subagents=[
            SubAgentConfig(
                name="researcher",
                description="Researches topics",
                instructions="You are a research assistant.",
            ),
        ],
    )],
)

SubAgentCapability dataclass

Bases: AbstractCapability[Any]

Capability that provides subagent delegation tools and dynamic instructions.

Combines the subagent toolset (task, check_task, answer_subagent, etc.) with dynamic system prompt injection listing available subagents.

Example
Python
from pydantic_ai import Agent
from subagents_pydantic_ai import SubAgentCapability, SubAgentConfig

cap = SubAgentCapability(
    default_model="openai:gpt-4.1",
    subagents=[
        SubAgentConfig(
            name="researcher",
            description="Researches topics",
            instructions="You are a research assistant.",
        ),
    ],
)
agent = Agent("openai:gpt-4.1", capabilities=[cap])

Attributes:

Name Type Description
subagents list[SubAgentConfig] | None

List of subagent configurations.

default_model Any

Default model for subagents that name none, for the general-purpose subagent, and for a dynamic call that names no model. There is no implicit default: leave it unset and each of those is refused rather than run on a model the library picked, which resolves whatever provider credential the process environment happens to hold.

include_general_purpose bool

Include general-purpose subagent. Needs default_model or default_agent_factory to build it from — see Raises. A default_agent_factory is what builds it when present, so a consumer that resolves a model, a credential and a budget per caller gets a delegate it can account for.

max_nesting_depth int

Max depth for nested subagents.

toolsets_factory ToolsetFactory | None

Factory for subagent toolsets.

registry DynamicAgentRegistry | None

Dynamic agent registry.

descriptions dict[str, str] | None

Custom tool descriptions override.

usage_limits UsageLimits | UsageLimitsFactory | None

Optional static or per-task usage limits for delegated subagent runs.

delegation_configuration DelegationConfiguration

Select default, persisted, combined, or one-shot-only delegation entry points. A mode that hides a tool rejects that tool's configuration rather than ignoring it — see Raises.

allowed_models list[str] | None

Optional model allow-list for dynamic specialists.

capabilities_map dict[str, CapabilityFactory] | None

Optional capability factories for dynamic specialists.

default_agent_factory AgentFactory | None

Optional custom agent factory for dynamic specialists, and for the general-purpose subagent when include_general_purpose is on. Do not attach an ask_parent toolset in the factory; the toolset injects it at run time when needed.

max_agents int

Maximum number of persistent dynamic agents, applied to the registry the toolset creates for create_agent. Ignored when registry is set — that registry keeps its own max_agents.

max_chat_traces int

Maximum number of subagent conversations kept for chat_trace_id continuation, least-recently-used evicted first.

max_task_handles int

Maximum number of finished task handles retained for status queries. Evicted usage still counts toward get_total_usage().

max_result_chars int | None

Character budget for a completed task's result in the wait_tasks listing. Truncated results carry an explicit marker pointing at check_task, which always returns the full text. Pass None to never truncate.

ask_user AskUserCallback | None

Callback backing ask_parent. Required for a sync-mode subagent to ask its parent anything: the parent's run loop is blocked inside the delegation, so answer_subagent cannot be reached until it returns.

event_stream_handler EventStreamHandler[Any] | None

Streams every delegation's events as they happen, including for specialists created at run time. An agent supplied as SubAgentConfig["agent"] keeps its own handler if it has one.

event_stream_handler_factory EventStreamHandlerFactory | None

The same, resolved per delegation from the parent run context, the subagent config and the task id — which is what lets a handler label the events of a fan-out. Mutually exclusive with event_stream_handler.

cancel_grace_seconds float

How long wrap_run waits for a cancelled background task to unwind before logging it and returning.

Raises:

Type Description
ValueError

From create_subagent_toolset when the configuration is self-contradictory — an invalid delegation_configuration, one whose hidden tools make subagents, registry, allowed_models, capabilities_map, or default_agent_factory unreachable, or both an event_stream_handler and an event_stream_handler_factory — and when a subagent has no model to run on: include_general_purpose with neither default_model nor default_agent_factory, or a subagents entry naming no model and supplying no agent or agent_factory while default_model is unset.

Source code in src/subagents_pydantic_ai/capability.py
Python
@dataclass
class SubAgentCapability(AbstractCapability[Any]):
    """Capability that provides subagent delegation tools and dynamic instructions.

    Combines the subagent toolset (task, check_task, answer_subagent, etc.)
    with dynamic system prompt injection listing available subagents.

    Example:
        ```python
        from pydantic_ai import Agent
        from subagents_pydantic_ai import SubAgentCapability, SubAgentConfig

        cap = SubAgentCapability(
            default_model="openai:gpt-4.1",
            subagents=[
                SubAgentConfig(
                    name="researcher",
                    description="Researches topics",
                    instructions="You are a research assistant.",
                ),
            ],
        )
        agent = Agent("openai:gpt-4.1", capabilities=[cap])
        ```

    Attributes:
        subagents: List of subagent configurations.
        default_model: Default model for subagents that name none, for the
            general-purpose subagent, and for a dynamic call that names no model.
            There is no implicit default: leave it unset and each of those is
            refused rather than run on a model the library picked, which resolves
            whatever provider credential the process environment happens to hold.
        include_general_purpose: Include general-purpose subagent. Needs
            `default_model` or `default_agent_factory` to build it from — see
            `Raises`. A `default_agent_factory` is what builds it when present,
            so a consumer that resolves a model, a credential and a budget per
            caller gets a delegate it can account for.
        max_nesting_depth: Max depth for nested subagents.
        toolsets_factory: Factory for subagent toolsets.
        registry: Dynamic agent registry.
        descriptions: Custom tool descriptions override.
        usage_limits: Optional static or per-task usage limits for delegated
            subagent runs.
        delegation_configuration: Select default, persisted, combined, or
            one-shot-only delegation entry points. A mode that hides a tool
            rejects that tool's configuration rather than ignoring it — see
            `Raises`.
        allowed_models: Optional model allow-list for dynamic specialists.
        capabilities_map: Optional capability factories for dynamic specialists.
        default_agent_factory: Optional custom agent factory for dynamic
            specialists, and for the general-purpose subagent when
            `include_general_purpose` is on. Do not attach an `ask_parent` toolset
            in the factory; the toolset injects it at run time when needed.
        max_agents: Maximum number of persistent dynamic agents, applied to the
            registry the toolset creates for `create_agent`. Ignored when
            `registry` is set — that registry keeps its own `max_agents`.
        max_chat_traces: Maximum number of subagent conversations kept for
            `chat_trace_id` continuation, least-recently-used evicted first.
        max_task_handles: Maximum number of finished task handles retained for
            status queries. Evicted usage still counts toward `get_total_usage()`.
        max_result_chars: Character budget for a completed task's result in the
            `wait_tasks` listing. Truncated results carry an explicit marker
            pointing at `check_task`, which always returns the full text. Pass
            `None` to never truncate.
        ask_user: Callback backing `ask_parent`. Required for a sync-mode subagent
            to ask its parent anything: the parent's run loop is blocked inside the
            delegation, so `answer_subagent` cannot be reached until it returns.
        event_stream_handler: Streams every delegation's events as they happen,
            including for specialists created at run time. An agent supplied as
            `SubAgentConfig["agent"]` keeps its own handler if it has one.
        event_stream_handler_factory: The same, resolved per delegation from the
            parent run context, the subagent config and the task id — which is
            what lets a handler label the events of a fan-out. Mutually exclusive
            with `event_stream_handler`.
        cancel_grace_seconds: How long `wrap_run` waits for a cancelled
            background task to unwind before logging it and returning.

    Raises:
        ValueError: From `create_subagent_toolset` when the configuration is
            self-contradictory — an invalid `delegation_configuration`, one
            whose hidden tools make `subagents`, `registry`, `allowed_models`,
            `capabilities_map`, or `default_agent_factory` unreachable, or both
            an `event_stream_handler` and an `event_stream_handler_factory` —
            and when a subagent has no model to run on: `include_general_purpose`
            with neither `default_model` nor `default_agent_factory`, or a
            `subagents` entry naming no `model` and supplying no `agent` or
            `agent_factory` while `default_model` is unset.
    """

    subagents: list[SubAgentConfig] | None = None
    default_model: Any = None
    include_general_purpose: bool = True
    max_nesting_depth: int = 0
    toolsets_factory: ToolsetFactory | None = None
    registry: DynamicAgentRegistry | None = None
    descriptions: dict[str, str] | None = None
    usage_limits: UsageLimits | UsageLimitsFactory | None = None
    delegation_configuration: DelegationConfiguration = "default"
    allowed_models: list[str] | None = None
    capabilities_map: dict[str, CapabilityFactory] | None = None
    default_agent_factory: AgentFactory | None = None
    max_agents: int = 10
    max_chat_traces: int = 100
    max_task_handles: int = 500
    max_result_chars: int | None = 2000
    ask_user: AskUserCallback | None = None
    ask_timeout_seconds: float = DEFAULT_ASK_TIMEOUT_SECONDS
    contain_errors: bool = True
    event_stream_handler: EventStreamHandler[Any] | None = None
    event_stream_handler_factory: EventStreamHandlerFactory | None = None
    cancel_grace_seconds: float = DEFAULT_CANCEL_GRACE_SECONDS
    _toolset: SubAgentToolset = field(init=False, repr=False)

    def __post_init__(self) -> None:
        """Create the underlying subagent toolset."""
        self._toolset = create_subagent_toolset(
            subagents=self.subagents,
            default_model=self.default_model,
            toolsets_factory=self.toolsets_factory,
            include_general_purpose=self.include_general_purpose,
            max_nesting_depth=self.max_nesting_depth,
            id="subagents",
            registry=self.registry,
            descriptions=self.descriptions,
            usage_limits=self.usage_limits,
            delegation_configuration=self.delegation_configuration,
            allowed_models=self.allowed_models,
            capabilities_map=self.capabilities_map,
            default_agent_factory=self.default_agent_factory,
            max_agents=self.max_agents,
            max_chat_traces=self.max_chat_traces,
            max_task_handles=self.max_task_handles,
            max_result_chars=self.max_result_chars,
            ask_user=self.ask_user,
            ask_timeout_seconds=self.ask_timeout_seconds,
            contain_errors=self.contain_errors,
            event_stream_handler=self.event_stream_handler,
            event_stream_handler_factory=self.event_stream_handler_factory,
            cancel_grace_seconds=self.cancel_grace_seconds,
        )

    @classmethod
    def get_serialization_name(cls) -> str:
        """Return name for AgentSpec YAML/JSON serialization."""
        return "SubAgentCapability"

    @property
    def task_manager(self) -> TaskManager:
        """The task manager behind the toolset, for observability."""
        return self._toolset.task_manager

    def get_toolset(self) -> AbstractToolset[Any] | None:
        """Return the subagent toolset with all registered tools."""
        return self._toolset

    async def wrap_run(
        self,
        ctx: RunContext[Any],
        *,
        handler: WrapRunHandler,
    ) -> AgentRunResult[Any]:
        """Run the agent, then stop every background task this run started.

        A background delegation is an `asyncio.Task` the parent run does not await.
        Without this finalizer it keeps executing after the run returns, against
        deps the application has already torn down, and one blocked in `ask_parent`
        waits for an answer that can never arrive.
        """
        try:
            return await handler()
        finally:
            await self._toolset.cancel_run_tasks(ctx.run_id)

    def get_instructions(self) -> Any:
        """Return dynamic instructions listing available subagents."""
        configs = list(self.subagents) if self.subagents else []

        def _instructions(ctx: RunContext[Any]) -> str:
            if self.delegation_configuration == "oneshot_only":
                return (
                    "## One-Shot Delegation\n\n"
                    "Use the `delegate` tool to create an ephemeral specialist "
                    "and run a task in one call."
                )
            return get_subagent_system_prompt(configs)

        return _instructions

task_manager property

The task manager behind the toolset, for observability.

__post_init__()

Create the underlying subagent toolset.

Source code in src/subagents_pydantic_ai/capability.py
Python
def __post_init__(self) -> None:
    """Create the underlying subagent toolset."""
    self._toolset = create_subagent_toolset(
        subagents=self.subagents,
        default_model=self.default_model,
        toolsets_factory=self.toolsets_factory,
        include_general_purpose=self.include_general_purpose,
        max_nesting_depth=self.max_nesting_depth,
        id="subagents",
        registry=self.registry,
        descriptions=self.descriptions,
        usage_limits=self.usage_limits,
        delegation_configuration=self.delegation_configuration,
        allowed_models=self.allowed_models,
        capabilities_map=self.capabilities_map,
        default_agent_factory=self.default_agent_factory,
        max_agents=self.max_agents,
        max_chat_traces=self.max_chat_traces,
        max_task_handles=self.max_task_handles,
        max_result_chars=self.max_result_chars,
        ask_user=self.ask_user,
        ask_timeout_seconds=self.ask_timeout_seconds,
        contain_errors=self.contain_errors,
        event_stream_handler=self.event_stream_handler,
        event_stream_handler_factory=self.event_stream_handler_factory,
        cancel_grace_seconds=self.cancel_grace_seconds,
    )

get_serialization_name() classmethod

Return name for AgentSpec YAML/JSON serialization.

Source code in src/subagents_pydantic_ai/capability.py
Python
@classmethod
def get_serialization_name(cls) -> str:
    """Return name for AgentSpec YAML/JSON serialization."""
    return "SubAgentCapability"

get_toolset()

Return the subagent toolset with all registered tools.

Source code in src/subagents_pydantic_ai/capability.py
Python
def get_toolset(self) -> AbstractToolset[Any] | None:
    """Return the subagent toolset with all registered tools."""
    return self._toolset

wrap_run(ctx, *, handler) async

Run the agent, then stop every background task this run started.

A background delegation is an asyncio.Task the parent run does not await. Without this finalizer it keeps executing after the run returns, against deps the application has already torn down, and one blocked in ask_parent waits for an answer that can never arrive.

Source code in src/subagents_pydantic_ai/capability.py
Python
async def wrap_run(
    self,
    ctx: RunContext[Any],
    *,
    handler: WrapRunHandler,
) -> AgentRunResult[Any]:
    """Run the agent, then stop every background task this run started.

    A background delegation is an `asyncio.Task` the parent run does not await.
    Without this finalizer it keeps executing after the run returns, against
    deps the application has already torn down, and one blocked in `ask_parent`
    waits for an answer that can never arrive.
    """
    try:
        return await handler()
    finally:
        await self._toolset.cancel_run_tasks(ctx.run_id)

get_instructions()

Return dynamic instructions listing available subagents.

Source code in src/subagents_pydantic_ai/capability.py
Python
def get_instructions(self) -> Any:
    """Return dynamic instructions listing available subagents."""
    configs = list(self.subagents) if self.subagents else []

    def _instructions(ctx: RunContext[Any]) -> str:
        if self.delegation_configuration == "oneshot_only":
            return (
                "## One-Shot Delegation\n\n"
                "Use the `delegate` tool to create an ephemeral specialist "
                "and run a task in one call."
            )
        return get_subagent_system_prompt(configs)

    return _instructions