Types API

SubAgentSpec, the validated YAML/JSON form of a config, is on its own Spec page.

SubAgentConfig

subagents_pydantic_ai.SubAgentConfig

Bases: _SubAgentConfigRequired

Configuration for a subagent.

Defines the name, description, and instructions for a subagent. Used by the toolset to create agent instances.

Required fields

name: Unique identifier for the subagent description: Brief description shown to parent agent instructions: System prompt for the subagent

Optional fields

model: LLM model to use (defaults to parent's default) agent: Pre-built agent instance. When provided, _compile_subagent uses this instead of creating a new Agent. Useful for passing agents created by frameworks like pydantic-deep. agent_factory: Callable that receives the SubAgentConfig and returns an agent instance. Called by _compile_subagent if agent is not provided. Signature: (config: SubAgentConfig) -> Agent. can_ask_questions: Whether subagent can ask parent questions max_questions: Maximum questions per task preferred_mode: Default execution mode preference for this subagent typical_complexity: Typical task complexity for this subagent typically_needs_context: Whether this subagent typically needs user context toolsets: Additional toolsets to register with the subagent agent_kwargs: Additional kwargs passed to Agent constructor (e.g., builtin_tools) context_files: List of context file paths in the backend. When used with pydantic-deep, these are loaded via ContextToolset and injected into this subagent's system prompt. Each subagent can have its own context files. extra: Generic extensibility dict for consumer libraries. subagents-pydantic-ai does not read this field — it's carried through for consumers like pydantic-deep to use freely. Example keys: memory, team, cost_budget. max_retries: Number of extra attempts after a transient failure (flaky gateway/network). Defaults to 3 — subagents are resilient out of the box; retries resume with the full message history so partial progress is not lost. Set 0 to disable retrying (legacy agent.run() opt-out path). retry_initial_delay: Seconds before the first retry (default 1.0). retry_max_delay: Cap for the backoff delay (default 30.0). retry_backoff_multiplier: Delay growth factor per attempt (default 2.0). retry_jitter: Randomise the backoff delay in [0, delay] to avoid a thundering herd (default True). retry_on: Custom predicate (exc) -> bool deciding whether an exception is transient. Defaults to the built-in classifier (ModelHTTPError 5xx/429/... and non-HTTP ModelAPIError). on_failure: Message returned to the parent as an ordinary tool result when this subagent fails, instead of raising ModelRetry. Use it to steer the parent ("summarise from what you already have") rather than letting it retry the delegation. contain_errors: Whether an unexpected subagent crash is contained. Defaults to True: the crash becomes a ModelRetry for the parent, logged with its traceback, so one failed delegation cannot abort the whole run. Set False to let crashes propagate. Control-flow signals (CallDeferred, ApprovalRequired, Skip*), UserError, and UsageLimitExceeded always propagate regardless.

Example with builtin_tools

Python

SubAgentConfig(
    name="researcher",
    description="Research agent with web search",
    instructions="You research topics using web search.",
    agent_kwargs={"builtin_tools": [BuitinTools.web_search]},
)

Example with per-subagent context

Python

SubAgentConfig(
    name="coder",
    description="Code writer",
    instructions="You write code following project rules.",
    context_files=["/agents/coder/AGENTS.md", "/CODING_RULES.md"],
)

Source code in src/subagents_pydantic_ai/types.py

class SubAgentConfig(_SubAgentConfigRequired, total=False):
    """Configuration for a subagent.

    Defines the name, description, and instructions for a subagent.
    Used by the toolset to create agent instances.

    Required fields:
        name: Unique identifier for the subagent
        description: Brief description shown to parent agent
        instructions: System prompt for the subagent

    Optional fields:
        model: LLM model to use (defaults to parent's default)
        agent: Pre-built agent instance. When provided, `_compile_subagent`
            uses this instead of creating a new `Agent`. Useful for passing
            agents created by frameworks like pydantic-deep.
        agent_factory: Callable that receives the SubAgentConfig and returns
            an agent instance. Called by `_compile_subagent` if `agent`
            is not provided. Signature: `(config: SubAgentConfig) -> Agent`.
        can_ask_questions: Whether subagent can ask parent questions
        max_questions: Maximum questions per task
        preferred_mode: Default execution mode preference for this subagent
        typical_complexity: Typical task complexity for this subagent
        typically_needs_context: Whether this subagent typically needs user context
        toolsets: Additional toolsets to register with the subagent
        agent_kwargs: Additional kwargs passed to Agent constructor (e.g., builtin_tools)
        context_files: List of context file paths in the backend.
            When used with pydantic-deep, these are loaded via ContextToolset
            and injected into this subagent's system prompt. Each subagent
            can have its own context files.
        extra: Generic extensibility dict for consumer libraries.
            subagents-pydantic-ai does not read this field — it's carried
            through for consumers like pydantic-deep to use freely.
            Example keys: `memory`, `team`, `cost_budget`.
        max_retries: Number of extra attempts after a transient failure
            (flaky gateway/network). Defaults to `3` — subagents are
            resilient out of the box; retries resume with the full
            message history so partial progress is not lost. Set `0`
            to disable retrying (legacy `agent.run()` opt-out path).
        retry_initial_delay: Seconds before the first retry (default 1.0).
        retry_max_delay: Cap for the backoff delay (default 30.0).
        retry_backoff_multiplier: Delay growth factor per attempt
            (default 2.0).
        retry_jitter: Randomise the backoff delay in `[0, delay]` to
            avoid a thundering herd (default `True`).
        retry_on: Custom predicate `(exc) -> bool` deciding whether an
            exception is transient. Defaults to the built-in classifier
            (`ModelHTTPError` 5xx/429/... and non-HTTP `ModelAPIError`).
        on_failure: Message returned to the parent as an ordinary tool result
            when this subagent fails, instead of raising `ModelRetry`. Use it
            to steer the parent ("summarise from what you already have")
            rather than letting it retry the delegation.
        contain_errors: Whether an unexpected subagent crash is contained.
            Defaults to `True`: the crash becomes a `ModelRetry` for the
            parent, logged with its traceback, so one failed delegation cannot
            abort the whole run. Set `False` to let crashes propagate.
            Control-flow signals (`CallDeferred`, `ApprovalRequired`,
            `Skip*`), `UserError`, and `UsageLimitExceeded` always propagate
            regardless.

    Example with builtin_tools:
        ```python
        SubAgentConfig(
            name="researcher",
            description="Research agent with web search",
            instructions="You research topics using web search.",
            agent_kwargs={"builtin_tools": [BuitinTools.web_search]},
        )
        ```

    Example with per-subagent context:
        ```python
        SubAgentConfig(
            name="coder",
            description="Code writer",
            instructions="You write code following project rules.",
            context_files=["/agents/coder/AGENTS.md", "/CODING_RULES.md"],
        )
        ```
    """

    model: NotRequired[str | Model]
    agent: NotRequired[Any]
    agent_factory: NotRequired[Callable[..., Any]]
    can_ask_questions: NotRequired[bool]
    max_questions: NotRequired[int]
    preferred_mode: NotRequired[Literal["sync", "async", "auto"]]
    typical_complexity: NotRequired[Literal["simple", "moderate", "complex"]]
    typically_needs_context: NotRequired[bool]
    toolsets: NotRequired[Sequence[AbstractToolset[Any]]]
    agent_kwargs: NotRequired[dict[str, Any]]
    context_files: NotRequired[list[str]]
    extra: NotRequired[dict[str, Any]]
    max_retries: NotRequired[int]
    retry_initial_delay: NotRequired[float]
    retry_max_delay: NotRequired[float]
    retry_backoff_multiplier: NotRequired[float]
    retry_jitter: NotRequired[bool]
    retry_on: NotRequired[Callable[[BaseException], bool]]
    on_failure: NotRequired[str]
    contain_errors: NotRequired[bool]

CompiledSubAgent

subagents_pydantic_ai.CompiledSubAgent dataclass

A pre-compiled subagent ready for use.

After processing SubAgentConfig, the toolset creates a CompiledSubAgent that includes the actual agent instance.

Attributes:

Name	Type	Description
name	str	Unique identifier for the subagent.
description	str	Brief description of the subagent's purpose.
agent	Any	The actual agent instance.
config	SubAgentConfig	The original configuration used to create this agent.

Source code in src/subagents_pydantic_ai/types.py

@dataclass
class CompiledSubAgent:
    """A pre-compiled subagent ready for use.

    After processing SubAgentConfig, the toolset creates a CompiledSubAgent
    that includes the actual agent instance.

    Attributes:
        name: Unique identifier for the subagent.
        description: Brief description of the subagent's purpose.
        agent: The actual agent instance.
        config: The original configuration used to create this agent.
    """

    name: str
    description: str
    config: SubAgentConfig
    agent: Any = None
    """The agent that runs when this subagent is delegated to.

    Deliberately untyped: consumers such as pydantic-deep supply their own agent
    objects rather than a `pydantic_ai.Agent`, so narrowing this would reject
    valid callers. Only `run`/`iter` are ever called on it.
    """

agent = None class-attribute instance-attribute

The agent that runs when this subagent is delegated to.

Deliberately untyped: consumers such as pydantic-deep supply their own agent objects rather than a pydantic_ai.Agent, so narrowing this would reject valid callers. Only run/iter are ever called on it.

TaskHandle

subagents_pydantic_ai.TaskHandle dataclass

Handle for managing a background task.

Returned when a task is started in async mode. Use this to check status, get results, or cancel the task.

Attributes:

Name	Type	Description
task_id	str	Unique identifier for the task
subagent_name	str	Name of the subagent executing the task
description	str	Task description
status	TaskStatus	Current task status
priority	TaskPriority	Task priority level
created_at	datetime	When the task was created
started_at	datetime | None	When execution started
completed_at	datetime | None	When execution finished
result	str | None	Task result (if completed)
error	str | None	Error message (if failed)
pending_question	str | None	Question waiting for answer (if any)
chat_trace_id	str | None	Chat trace ID for continuing this subagent conversation
run_id	str | None	Pydantic AI run ID for the subagent run
conversation_id	str | None	Pydantic AI conversation ID for the subagent run
traceparent	str | None	W3C traceparent for the subagent run span, if available
cost	Decimal | None	Total subagent model cost calculated from genai-prices
parent_run_id	str | None	run_id of the parent run that started this task. Tools use it to refuse cross-run access, and the capability uses it to cancel a run's tasks when that run ends.
deferred_requests	DeferredToolRequests | None	The tool calls a DEFERRED subagent is suspended on. Set only for that status, and the only place they survive: the delegation raises so the signal reaches the parent run, and a raise carries no result to read them off.

Source code in src/subagents_pydantic_ai/types.py

@dataclass
class TaskHandle:
    """Handle for managing a background task.

    Returned when a task is started in async mode. Use this to
    check status, get results, or cancel the task.

    Attributes:
        task_id: Unique identifier for the task
        subagent_name: Name of the subagent executing the task
        description: Task description
        status: Current task status
        priority: Task priority level
        created_at: When the task was created
        started_at: When execution started
        completed_at: When execution finished
        result: Task result (if completed)
        error: Error message (if failed)
        pending_question: Question waiting for answer (if any)
        chat_trace_id: Chat trace ID for continuing this subagent conversation
        run_id: Pydantic AI run ID for the subagent run
        conversation_id: Pydantic AI conversation ID for the subagent run
        traceparent: W3C traceparent for the subagent run span, if available
        cost: Total subagent model cost calculated from genai-prices
        parent_run_id: `run_id` of the parent run that started this task. Tools
            use it to refuse cross-run access, and the capability uses it to
            cancel a run's tasks when that run ends.
        deferred_requests: The tool calls a `DEFERRED` subagent is suspended on.
            Set only for that status, and the only place they survive: the
            delegation raises so the signal reaches the parent run, and a raise
            carries no result to read them off.
    """

    task_id: str
    subagent_name: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    priority: TaskPriority = TaskPriority.NORMAL
    created_at: datetime = field(default_factory=utcnow)
    started_at: datetime | None = None
    completed_at: datetime | None = None
    result: str | None = None
    error: str | None = None
    pending_question: str | None = None
    chat_trace_id: str | None = None
    usage: RunUsage | None = None
    """Token usage from the subagent run."""
    message_history: str | None = None
    run_id: str | None = None
    conversation_id: str | None = None
    traceparent: str | None = None
    trace_id: str | None = None
    span_id: str | None = None
    model_name: str | None = None
    provider_name: str | None = None
    provider_url: str | None = None
    provider_response_id: str | None = None
    provider_details: dict[str, Any] | None = None
    finish_reason: FinishReason | None = None
    cost: Decimal | None = None
    tool_call_counts: dict[str, int] = field(default_factory=dict)
    retry_count: int = 0
    """Number of transient-failure retries performed for this task."""
    parent_run_id: str | None = None
    deferred_requests: DeferredToolRequests | None = None

    @property
    def is_finished(self) -> bool:
        """Whether the task reached a terminal status."""
        return self.status in TERMINAL_STATUSES

    def finish(
        self,
        status: TaskStatus,
        *,
        result: str | None = None,
        error: str | None = None,
    ) -> bool:
        """Record a terminal outcome, first terminal transition winning.

        Idempotence is what makes `hard_cancel` safe. A task that already set
        `COMPLETED` and is running its `finally` block is still not
        `asyncio.Task.done()`, so a cancel arriving in that window used to
        overwrite the real result with `CANCELLED` and move `completed_at`.

        Args:
            status: The terminal status to record.
            result: Result text, for a completed task.
            error: Error text, for a failed or cancelled task.

        Returns:
            Whether this call recorded the outcome.
        """
        if self.is_finished:
            return False
        self.status = status
        self.result = result
        self.error = error
        self.completed_at = utcnow()
        return True

usage = None class-attribute instance-attribute

Token usage from the subagent run.

retry_count = 0 class-attribute instance-attribute

Number of transient-failure retries performed for this task.

is_finished property

Whether the task reached a terminal status.

finish(status, *, result=None, error=None)

Record a terminal outcome, first terminal transition winning.

Idempotence is what makes hard_cancel safe. A task that already set COMPLETED and is running its finally block is still not asyncio.Task.done(), so a cancel arriving in that window used to overwrite the real result with CANCELLED and move completed_at.

Parameters:

Name	Type	Description	Default
status	TaskStatus	The terminal status to record.	required
result	str | None	Result text, for a completed task.	None
error	str | None	Error text, for a failed or cancelled task.	None

Returns:

Type	Description
bool	Whether this call recorded the outcome.

Source code in src/subagents_pydantic_ai/types.py

def finish(
    self,
    status: TaskStatus,
    *,
    result: str | None = None,
    error: str | None = None,
) -> bool:
    """Record a terminal outcome, first terminal transition winning.

    Idempotence is what makes `hard_cancel` safe. A task that already set
    `COMPLETED` and is running its `finally` block is still not
    `asyncio.Task.done()`, so a cancel arriving in that window used to
    overwrite the real result with `CANCELLED` and move `completed_at`.

    Args:
        status: The terminal status to record.
        result: Result text, for a completed task.
        error: Error text, for a failed or cancelled task.

    Returns:
        Whether this call recorded the outcome.
    """
    if self.is_finished:
        return False
    self.status = status
    self.result = result
    self.error = error
    self.completed_at = utcnow()
    return True

TaskStatus

subagents_pydantic_ai.TaskStatus

Bases: _ValueStrEnum

Status of a background task.

Source code in src/subagents_pydantic_ai/types.py

class TaskStatus(_ValueStrEnum):
    """Status of a background task."""

    PENDING = "pending"
    """Task is queued but not started."""

    RUNNING = "running"
    """Task is currently executing."""

    WAITING_FOR_ANSWER = "waiting_for_answer"
    """Task is blocked waiting for parent response."""

    COMPLETED = "completed"
    """Task finished successfully."""

    FAILED = "failed"
    """Task failed with an error."""

    CANCELLED = "cancelled"
    """Task was cancelled."""

    RETRYING = "retrying"
    """Task hit a transient error and is waiting to retry."""

    DEFERRED = "deferred"
    """Task suspended for human approval or a deferred tool call.

    Distinct from `FAILED`: nothing went wrong, the subagent is waiting on a
    decision this library cannot make. Terminal here because resuming a
    suspension belongs to whoever holds the deferred requests -- see
    `TaskHandle.deferred_requests`.
    """

PENDING = 'pending' class-attribute instance-attribute

Task is queued but not started.

RUNNING = 'running' class-attribute instance-attribute

Task is currently executing.

WAITING_FOR_ANSWER = 'waiting_for_answer' class-attribute instance-attribute

Task is blocked waiting for parent response.

COMPLETED = 'completed' class-attribute instance-attribute

Task finished successfully.

FAILED = 'failed' class-attribute instance-attribute

Task failed with an error.

CANCELLED = 'cancelled' class-attribute instance-attribute

Task was cancelled.

RETRYING = 'retrying' class-attribute instance-attribute

Task hit a transient error and is waiting to retry.

DEFERRED = 'deferred' class-attribute instance-attribute

Task suspended for human approval or a deferred tool call.

Distinct from FAILED: nothing went wrong, the subagent is waiting on a decision this library cannot make. Terminal here because resuming a suspension belongs to whoever holds the deferred requests -- see TaskHandle.deferred_requests.

TaskPriority

subagents_pydantic_ai.TaskPriority

Bases: _ValueStrEnum

Priority levels for background tasks.

Source code in src/subagents_pydantic_ai/types.py

class TaskPriority(_ValueStrEnum):
    """Priority levels for background tasks."""

    LOW = "low"
    """Low priority task, can be deferred."""

    NORMAL = "normal"
    """Normal priority task (default)."""

    HIGH = "high"
    """High priority task, should be processed soon."""

    CRITICAL = "critical"
    """Critical priority task, process immediately."""

LOW = 'low' class-attribute instance-attribute

Low priority task, can be deferred.

NORMAL = 'normal' class-attribute instance-attribute

Normal priority task (default).

HIGH = 'high' class-attribute instance-attribute

High priority task, should be processed soon.

CRITICAL = 'critical' class-attribute instance-attribute

Critical priority task, process immediately.

ExecutionMode

subagents_pydantic_ai.ExecutionMode = Literal['sync', 'async', 'auto'] module-attribute

Execution mode for subagent tasks.

• sync: Execute synchronously, blocking until completion (default)
• async: Execute in background, return immediately with task handle
• auto: Automatically decide based on task characteristics

TaskCharacteristics

subagents_pydantic_ai.TaskCharacteristics dataclass

Characteristics that help decide execution mode.

These characteristics are used by decide_execution_mode to automatically select between sync and async execution based on task properties.

Attributes:

Name	Type	Description
estimated_complexity	Literal['simple', 'moderate', 'complex']	Expected task complexity level.
requires_user_context	bool	Whether task needs ongoing user interaction.
is_time_sensitive	bool	Whether quick response is important.
can_run_independently	bool	Whether task can complete without further input.
may_need_clarification	bool	Whether task might need clarifying questions.

Source code in src/subagents_pydantic_ai/types.py

@dataclass
class TaskCharacteristics:
    """Characteristics that help decide execution mode.

    These characteristics are used by `decide_execution_mode` to automatically
    select between sync and async execution based on task properties.

    Attributes:
        estimated_complexity: Expected task complexity level.
        requires_user_context: Whether task needs ongoing user interaction.
        is_time_sensitive: Whether quick response is important.
        can_run_independently: Whether task can complete without further input.
        may_need_clarification: Whether task might need clarifying questions.
    """

    estimated_complexity: Literal["simple", "moderate", "complex"] = "moderate"
    requires_user_context: bool = False
    is_time_sensitive: bool = False
    can_run_independently: bool = True
    may_need_clarification: bool = False

AgentMessage

subagents_pydantic_ai.AgentMessage dataclass

Message passed between agents via the message bus.

Attributes:

Name	Type	Description
type	MessageType	The message type (task_assigned, question, etc.)
sender	str	ID of the sending agent
receiver	str	ID of the receiving agent
payload	Any	Message-specific data
task_id	str	Associated task ID for correlation
id	str	Unique message identifier for tracing/debugging
timestamp	datetime	When the message was created
correlation_id	str | None	ID for request-response correlation

Source code in src/subagents_pydantic_ai/types.py

@dataclass
class AgentMessage:
    """Message passed between agents via the message bus.

    Attributes:
        type: The message type (task_assigned, question, etc.)
        sender: ID of the sending agent
        receiver: ID of the receiving agent
        payload: Message-specific data
        task_id: Associated task ID for correlation
        id: Unique message identifier for tracing/debugging
        timestamp: When the message was created
        correlation_id: ID for request-response correlation
    """

    type: MessageType
    sender: str
    receiver: str
    payload: Any
    task_id: str
    id: str = field(default_factory=_generate_message_id)
    timestamp: datetime = field(default_factory=utcnow)
    correlation_id: str | None = None

MessageType

subagents_pydantic_ai.MessageType

Bases: _ValueStrEnum

Types of messages that can be sent between agents.

Source code in src/subagents_pydantic_ai/types.py

class MessageType(_ValueStrEnum):
    """Types of messages that can be sent between agents."""

    TASK_ASSIGNED = "task_assigned"
    """A new task has been assigned to a subagent."""

    TASK_UPDATE = "task_update"
    """Progress update from a running task."""

    TASK_COMPLETED = "task_completed"
    """Task finished successfully."""

    TASK_FAILED = "task_failed"
    """Task failed with an error."""

    QUESTION = "question"
    """Subagent is asking the parent a question."""

    ANSWER = "answer"
    """Parent's response to a subagent question."""

    CANCEL_REQUEST = "cancel_request"
    """Request to cancel a task (soft cancel)."""

    CANCEL_FORCED = "cancel_forced"
    """Immediate cancellation (hard cancel)."""

TASK_ASSIGNED = 'task_assigned' class-attribute instance-attribute

A new task has been assigned to a subagent.

TASK_UPDATE = 'task_update' class-attribute instance-attribute

Progress update from a running task.

TASK_COMPLETED = 'task_completed' class-attribute instance-attribute

Task finished successfully.

TASK_FAILED = 'task_failed' class-attribute instance-attribute

Task failed with an error.

QUESTION = 'question' class-attribute instance-attribute

Subagent is asking the parent a question.

ANSWER = 'answer' class-attribute instance-attribute

Parent's response to a subagent question.

CANCEL_REQUEST = 'cancel_request' class-attribute instance-attribute

Request to cancel a task (soft cancel).

CANCEL_FORCED = 'cancel_forced' class-attribute instance-attribute

Immediate cancellation (hard cancel).

ToolsetFactory

subagents_pydantic_ai.ToolsetFactory = Callable[[Any], 'Sequence[AbstractToolset[Any]]'] module-attribute

Factory function that creates toolsets for a subagent.

Takes the subagent's deps as input and returns the toolsets to register. The deps parameter is Any because its type belongs to the application, not this library. The return type is a Sequence so a factory annotated with a concrete toolset type (list[FunctionToolset[MyDeps]]) still satisfies it -- list is invariant, Sequence is not.

Example

Python

def my_toolset_factory(deps: MyDeps) -> list[FunctionToolset[MyDeps]]:
    return [
        create_file_toolset(deps.backend),
        create_todo_toolset(),
    ]

UsageLimitsFactory

subagents_pydantic_ai.UsageLimitsFactory = Callable[[RunContext[Any], SubAgentConfig], 'UsageLimits | None'] module-attribute

Factory function that resolves usage limits for a delegated subagent task.

Called once per delegated task with the parent run context and selected subagent config. Return None to run that task without explicit limits.

AskUserCallback

subagents_pydantic_ai.AskUserCallback = Callable[[str], Awaitable[str]] module-attribute

Callback invoked when a subagent calls ask_parent in sync mode.

Receives the subagent's question and must return the answer. Typically wired to a human-in-the-loop UI, a CLI input() prompt, or a pre-canned answerer for tests.

Example

```python async def ask_user(question: str) -> str: return input(f"Subagent asks: {question}

")

Text Only

toolset = create_subagent_toolset(
    subagents=subagents,
    ask_user=ask_user,
)
```

decide_execution_mode

subagents_pydantic_ai.decide_execution_mode(characteristics, config, force_mode=None)

Decide whether to run sync or async based on task characteristics.

This function implements the auto-mode selection logic. It considers: 1. Explicit force_mode override 2. Config-level preferred_mode 3. Task characteristics

Parameters:

Name	Type	Description	Default
characteristics	TaskCharacteristics	Task characteristics that influence the decision.	required
config	SubAgentConfig	Subagent configuration with optional preferences.	required
force_mode	ExecutionMode | None	Override mode (if specified and not "auto").	None

Returns:

Type	Description
Literal['sync', 'async']	The resolved execution mode: either "sync" or "async".

Example

Python

characteristics = TaskCharacteristics(
    estimated_complexity="complex",
    can_run_independently=True,
)
config = SubAgentConfig(name="worker", ...)
mode = decide_execution_mode(characteristics, config)
# mode will be "async" for complex independent tasks

Source code in src/subagents_pydantic_ai/types.py

def decide_execution_mode(
    characteristics: TaskCharacteristics,
    config: SubAgentConfig,
    force_mode: ExecutionMode | None = None,
) -> Literal["sync", "async"]:
    """Decide whether to run sync or async based on task characteristics.

    This function implements the auto-mode selection logic. It considers:
    1. Explicit force_mode override
    2. Config-level preferred_mode
    3. Task characteristics

    Args:
        characteristics: Task characteristics that influence the decision.
        config: Subagent configuration with optional preferences.
        force_mode: Override mode (if specified and not "auto").

    Returns:
        The resolved execution mode: either "sync" or "async".

    Example:
        ```python
        characteristics = TaskCharacteristics(
            estimated_complexity="complex",
            can_run_independently=True,
        )
        config = SubAgentConfig(name="worker", ...)
        mode = decide_execution_mode(characteristics, config)
        # mode will be "async" for complex independent tasks
        ```
    """
    # Explicit override takes precedence
    if force_mode and force_mode != "auto":
        return force_mode

    # Config-level preference
    config_preference = config.get("preferred_mode", "auto")
    if config_preference != "auto":
        return config_preference

    # Always sync if needs user context or clarification likely
    if characteristics.requires_user_context:
        return "sync"
    if characteristics.may_need_clarification and characteristics.is_time_sensitive:
        return "sync"

    # Prefer async for complex, independent tasks
    if characteristics.estimated_complexity == "complex" and characteristics.can_run_independently:
        return "async"

    # Simple tasks - sync is fine
    if characteristics.estimated_complexity == "simple":
        return "sync"

    # Default to async for moderate complexity if can run independently
    if characteristics.can_run_independently:
        return "async"

    return "sync"

---

Usage Examples

Creating a SubAgentConfig

Python

from subagents_pydantic_ai import SubAgentConfig

config = SubAgentConfig(
    name="researcher",
    description="Researches topics",
    instructions="You are a research assistant.",
    model="openai:gpt-4o",
    can_ask_questions=True,
    max_questions=3,
    preferred_mode="async",
    typical_complexity="complex",
)

Working with TaskHandle

Python

from subagents_pydantic_ai import TaskHandle, TaskStatus

# TaskHandle is returned by async tasks
handle: TaskHandle = ...

# Check status
if handle.status == TaskStatus.COMPLETED:
    print(f"Result: {handle.result}")
elif handle.status == TaskStatus.WAITING_FOR_ANSWER:
    print(f"Question: {handle.pending_question}")
elif handle.status == TaskStatus.FAILED:
    print(f"Error: {handle.error}")

Using decide_execution_mode

Python

from subagents_pydantic_ai import (
    decide_execution_mode,
    TaskCharacteristics,
    SubAgentConfig,
)

characteristics = TaskCharacteristics(
    estimated_complexity="complex",
    requires_user_context=False,
    can_run_independently=True,
)

config = SubAgentConfig(
    name="worker",
    description="...",
    instructions="...",
)

mode = decide_execution_mode(characteristics, config)
# Returns "async" for complex, independent tasks

Message Types

Python

from subagents_pydantic_ai import MessageType, AgentMessage

# Create a question message
message = AgentMessage(
    type=MessageType.QUESTION,
    sender="subagent-123",
    receiver="parent-456",
    payload={"question": "Which database?"},
    task_id="task-789",
)