Skip to content

Types API

SubAgentConfig

subagents_pydantic_ai.SubAgentConfig

Bases: TypedDict

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).

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
Python
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class SubAgentConfig(TypedDict, 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`).

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

    name: str
    description: str
    instructions: str
    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[list[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]]

SubAgentSpec

subagents_pydantic_ai.SubAgentSpec

Bases: BaseModel

Declarative subagent configuration for YAML/JSON specs.

A Pydantic model that mirrors SubAgentConfig (a TypedDict) but provides validation, defaults, and serialization support. This makes it suitable for loading subagent definitions from YAML or JSON files.

Attributes:

Name Type Description
name str

Unique identifier for the subagent.
description str

Brief description shown to the parent agent.
instructions str

System prompt for the subagent.
model str | None

LLM model identifier (e.g. openai:gpt-4.1). If None, the parent agent's default model is used.
can_ask_questions bool | None

Whether the subagent can ask the parent questions.
max_questions int | None

Maximum number of questions per task.
preferred_mode Literal['sync', 'async', 'auto'] | None

Default execution mode preference.
typical_complexity Literal['simple', 'moderate', 'complex'] | None

Typical task complexity for this subagent.
typically_needs_context bool | None

Whether this subagent typically needs user context.
context_files list[str] | None

List of context file paths to inject into system prompt.
extra dict[str, Any]

Generic extensibility dict for consumer libraries.
Source code in src/subagents_pydantic_ai/spec.py
Python
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class SubAgentSpec(BaseModel):
    """Declarative subagent configuration for YAML/JSON specs.

    A Pydantic model that mirrors `SubAgentConfig` (a TypedDict) but
    provides validation, defaults, and serialization support. This makes
    it suitable for loading subagent definitions from YAML or JSON files.

    Attributes:
        name: Unique identifier for the subagent.
        description: Brief description shown to the parent agent.
        instructions: System prompt for the subagent.
        model: LLM model identifier (e.g. `openai:gpt-4.1`).
            If None, the parent agent's default model is used.
        can_ask_questions: Whether the subagent can ask the parent questions.
        max_questions: Maximum number of questions per task.
        preferred_mode: Default execution mode preference.
        typical_complexity: Typical task complexity for this subagent.
        typically_needs_context: Whether this subagent typically needs user context.
        context_files: List of context file paths to inject into system prompt.
        extra: Generic extensibility dict for consumer libraries.
    """

    name: str
    description: str = ""
    instructions: str = ""
    model: str | None = None
    can_ask_questions: bool | None = None
    max_questions: int | None = None
    preferred_mode: Literal["sync", "async", "auto"] | None = None
    typical_complexity: Literal["simple", "moderate", "complex"] | None = None
    typically_needs_context: bool | None = None
    context_files: list[str] | None = None
    extra: dict[str, Any] = Field(default_factory=dict)

    def to_config(self) -> SubAgentConfig:
        """Convert to a SubAgentConfig TypedDict.

        Only includes fields that have been explicitly set (non-None values),
        except for `extra` which is included when non-empty.

        Returns:
            A SubAgentConfig dict suitable for `create_subagent_toolset()`.
        """
        config = SubAgentConfig(
            name=self.name,
            description=self.description,
            instructions=self.instructions,
        )

        if self.model is not None:
            config["model"] = self.model
        if self.can_ask_questions is not None:
            config["can_ask_questions"] = self.can_ask_questions
        if self.max_questions is not None:
            config["max_questions"] = self.max_questions
        if self.preferred_mode is not None:
            config["preferred_mode"] = self.preferred_mode
        if self.typical_complexity is not None:
            config["typical_complexity"] = self.typical_complexity
        if self.typically_needs_context is not None:
            config["typically_needs_context"] = self.typically_needs_context
        if self.context_files is not None:
            config["context_files"] = self.context_files
        if self.extra:
            config["extra"] = self.extra

        return config

    @classmethod
    def from_config(cls, config: SubAgentConfig) -> SubAgentSpec:
        """Create a SubAgentSpec from a SubAgentConfig dict.

        Args:
            config: A SubAgentConfig TypedDict.

        Returns:
            A new SubAgentSpec instance.
        """
        data: dict[str, Any] = {
            "name": config["name"],
            "description": config.get("description", ""),
            "instructions": config.get("instructions", ""),
        }

        # Map optional fields, converting Model objects to str for model field
        model_val = config.get("model")
        if model_val is not None:
            data["model"] = str(model_val)

        for field_name in (
            "can_ask_questions",
            "max_questions",
            "preferred_mode",
            "typical_complexity",
            "typically_needs_context",
            "context_files",
            "extra",
        ):
            if field_name in config:
                data[field_name] = config.get(field_name)

        return cls(**data)

to_config()

Convert to a SubAgentConfig TypedDict.

Only includes fields that have been explicitly set (non-None values), except for extra which is included when non-empty.

Returns:

Type Description
SubAgentConfig

A SubAgentConfig dict suitable for create_subagent_toolset().
Source code in src/subagents_pydantic_ai/spec.py
Python
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def to_config(self) -> SubAgentConfig:
    """Convert to a SubAgentConfig TypedDict.

    Only includes fields that have been explicitly set (non-None values),
    except for `extra` which is included when non-empty.

    Returns:
        A SubAgentConfig dict suitable for `create_subagent_toolset()`.
    """
    config = SubAgentConfig(
        name=self.name,
        description=self.description,
        instructions=self.instructions,
    )

    if self.model is not None:
        config["model"] = self.model
    if self.can_ask_questions is not None:
        config["can_ask_questions"] = self.can_ask_questions
    if self.max_questions is not None:
        config["max_questions"] = self.max_questions
    if self.preferred_mode is not None:
        config["preferred_mode"] = self.preferred_mode
    if self.typical_complexity is not None:
        config["typical_complexity"] = self.typical_complexity
    if self.typically_needs_context is not None:
        config["typically_needs_context"] = self.typically_needs_context
    if self.context_files is not None:
        config["context_files"] = self.context_files
    if self.extra:
        config["extra"] = self.extra

    return config

from_config(config) classmethod

Create a SubAgentSpec from a SubAgentConfig dict.

Parameters:

Name Type Description Default
config SubAgentConfig

A SubAgentConfig TypedDict.

 required

Returns:

Type Description
SubAgentSpec

A new SubAgentSpec instance.
Source code in src/subagents_pydantic_ai/spec.py
Python
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@classmethod
def from_config(cls, config: SubAgentConfig) -> SubAgentSpec:
    """Create a SubAgentSpec from a SubAgentConfig dict.

    Args:
        config: A SubAgentConfig TypedDict.

    Returns:
        A new SubAgentSpec instance.
    """
    data: dict[str, Any] = {
        "name": config["name"],
        "description": config.get("description", ""),
        "instructions": config.get("instructions", ""),
    }

    # Map optional fields, converting Model objects to str for model field
    model_val = config.get("model")
    if model_val is not None:
        data["model"] = str(model_val)

    for field_name in (
        "can_ask_questions",
        "max_questions",
        "preferred_mode",
        "typical_complexity",
        "typically_needs_context",
        "context_files",
        "extra",
    ):
        if field_name in config:
            data[field_name] = config.get(field_name)

    return cls(**data)

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 object | None

The actual agent instance.
config SubAgentConfig

The original configuration used to create this agent.
Source code in src/subagents_pydantic_ai/types.py
Python
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
@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: object | None = None  # Agent instance - typed as object to avoid circular imports

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)
Source code in src/subagents_pydantic_ai/types.py
Python
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
@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)
    """

    task_id: str
    subagent_name: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    priority: TaskPriority = TaskPriority.NORMAL
    created_at: datetime = field(default_factory=datetime.now)
    started_at: datetime | None = None
    completed_at: datetime | None = None
    result: str | None = None
    error: str | None = None
    pending_question: str | None = None
    usage: Any = None
    """Token usage from the subagent run (`RunUsage` from pydantic-ai)."""
    retry_count: int = 0
    """Number of transient-failure retries performed for this task."""

usage = None class-attribute instance-attribute

Token usage from the subagent run (RunUsage from pydantic-ai).

retry_count = 0 class-attribute instance-attribute

Number of transient-failure retries performed for this task.

TaskStatus

subagents_pydantic_ai.TaskStatus

Bases: str, Enum

Status of a background task.

Source code in src/subagents_pydantic_ai/types.py
Python
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class TaskStatus(str, Enum):
    """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."""

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.

TaskPriority

subagents_pydantic_ai.TaskPriority

Bases: str, Enum

Priority levels for background tasks.

Source code in src/subagents_pydantic_ai/types.py
Python
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class TaskPriority(str, Enum):
    """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
Python
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@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
Python
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@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=datetime.now)
    correlation_id: str | None = None

MessageType

subagents_pydantic_ai.MessageType

Bases: str, Enum

Types of messages that can be sent between agents.

Source code in src/subagents_pydantic_ai/types.py
Python
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class MessageType(str, Enum):
    """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], list[Any]] module-attribute

Factory function that creates toolsets for a subagent.

Takes deps as input and returns a list of toolsets to register.

Example
Python
def my_toolset_factory(deps: MyDeps) -> list[Any]:
    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
Python
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
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",
)