subagents-pydantic-ai
Subagent Delegation framework for Pydantic AI, enabling nested subagents that can spawn their own specialists on-the-fly, with smart sync/async/auto mode selection, runtime agent creation, and clean multi-agent architecture. Adds specialization, parallel execution, and task cancellation.
Subagents for Pydantic AI
Multi-Agent Orchestration for Pydantic AI
Nested Subagents — subagents spawn their own subagents • Runtime Agent Creation — create specialists on-the-fly • Auto-Mode Selection — intelligent sync/async decision
Subagents for Pydantic AI adds multi-agent delegation to any Pydantic AI agent. Spawn specialized subagents that run synchronously (blocking), asynchronously (background), or let the system auto-select the best mode.
Full framework? Check out Pydantic Deep Agents - complete agent framework with planning, filesystem, subagents, and skills.
Use Cases
| What You Want to Build | How Subagents Help |
|---|---|
| Research Assistant | Delegate research to specialists, synthesize with a writer agent |
| Code Review System | Security agent, style agent, and performance agent work in parallel |
| Content Pipeline | Researcher → Analyst → Writer chain with handoffs |
| Data Processing | Spawn workers dynamically based on data volume |
| Customer Support | Route to specialized agents (billing, technical, sales) |
| Document Analysis | Extract, summarize, and categorize with focused agents |
Installation
pip install subagents-pydantic-ai
Or with uv:
uv add subagents-pydantic-ai
Quick Start
The recommended way to add subagent delegation is via the Capabilities API:
from pydantic_ai import Agent
from subagents_pydantic_ai import SubAgentCapability, SubAgentConfig
agent = Agent(
"openai:gpt-4.1",
capabilities=[SubAgentCapability(
subagents=[
SubAgentConfig(
name="researcher",
description="Researches topics and gathers information",
instructions="You are a research assistant. Investigate thoroughly.",
),
SubAgentConfig(
name="writer",
description="Writes content based on research",
instructions="You are a technical writer. Write clear, concise content.",
),
],
)],
)
result = await agent.run("Research Python async patterns and write a blog post about it")
SubAgentCapability automatically:
- Registers all delegation tools (
task,check_task,answer_subagent,list_active_tasks, etc.) - Injects dynamic system prompt listing available subagents
- Includes a general-purpose subagent by default
Alternative: Toolset API
For lower-level control:
from pydantic_ai import Agent
from subagents_pydantic_ai import create_subagent_toolset, SubAgentConfig
toolset = create_subagent_toolset(
subagents=[
SubAgentConfig(name="researcher", description="Researches topics", instructions="..."),
],
)
agent = Agent("openai:gpt-4.1", toolsets=[toolset])
Note: With the toolset API, you need to wire
get_subagent_system_prompt()manually.SubAgentCapabilityhandles this automatically.
Execution Modes
Choose how subagents execute their tasks:
| Mode | Description | Use Case |
|---|---|---|
sync |
Block until complete | Quick tasks, when result is needed immediately |
async |
Run in background | Long research, parallel tasks |
auto |
Smart selection | Let the system decide based on task characteristics |
Sync Mode (Default)
# Agent calls: task(description="...", subagent_type="researcher", mode="sync")
# Parent waits for result before continuing
Async Mode
# Agent calls: task(description="...", subagent_type="researcher", mode="async")
# Returns task_id immediately, agent continues working
# Later: check_task(task_id) to get result
Auto Mode
# Agent calls: task(description="...", subagent_type="researcher", mode="auto")
# System decides based on:
# - Task complexity (simple → sync, complex → async)
# - Independence (can run without user context → async)
# - Subagent preferences (from config)
Give Subagents Tools
Provide toolsets so subagents can interact with files, APIs, or other services:
from pydantic_ai_backends import create_console_toolset
def my_toolsets_factory(deps):
"""Factory that creates toolsets for subagents."""
return [
create_console_toolset(), # File operations
create_search_toolset(), # Web search
]
toolset = create_subagent_toolset(
subagents=subagents,
toolsets_factory=my_toolsets_factory,
)
Dynamic Agent Creation
Create agents on-the-fly and delegate to them seamlessly:
from subagents_pydantic_ai import (
create_subagent_toolset,
create_agent_factory_toolset,
DynamicAgentRegistry,
)
registry = DynamicAgentRegistry()
agent = Agent(
"openai:gpt-4o",
deps_type=Deps,
toolsets=[
# Pass registry so task() can resolve dynamically created agents
create_subagent_toolset(registry=registry),
create_agent_factory_toolset(
registry=registry,
allowed_models=["openai:gpt-4o", "openai:gpt-4o-mini"],
max_agents=5,
),
],
)
# Now the agent can:
# 1. create_agent(name="analyst", ...) — creates a new agent in registry
# 2. task(description="...", subagent_type="analyst") — delegates to it
Subagent Questions
Enable subagents to ask the parent for clarification:
SubAgentConfig(
name="analyst",
description="Analyzes data",
instructions="Ask for clarification when data is ambiguous.",
can_ask_questions=True,
max_questions=3,
)
The parent agent can then respond using answer_subagent(task_id, answer).
Available Tools
| Tool | Description |
|---|---|
task |
Delegate a task to a subagent (sync, async, or auto) |
check_task |
Check status and get result of a background task |
answer_subagent |
Answer a question from a blocked subagent |
list_active_tasks |
List all running background tasks |
soft_cancel_task |
Request cooperative cancellation |
hard_cancel_task |
Immediately cancel a task |
Declarative Configuration (YAML/JSON)
Define subagents in YAML or JSON files using SubAgentSpec:
# subagents.yaml
- name: researcher
description: Research assistant
instructions: You research topics thoroughly.
model: openai:gpt-4.1-mini
- name: coder
description: Code writer
instructions: You write clean Python code.
can_ask_questions: true
max_questions: 3
import yaml
from subagents_pydantic_ai import SubAgentSpec
# Load from YAML
with open("subagents.yaml") as f:
specs = [SubAgentSpec(**s) for s in yaml.safe_load(f)]
# Convert to SubAgentConfig dicts
configs = [spec.to_config() for spec in specs]
# Use with capability
agent = Agent("openai:gpt-4.1", capabilities=[
SubAgentCapability(subagents=configs),
])
Round-trip between specs and configs:
# Config -> Spec -> Config
spec = SubAgentSpec.from_config(existing_config)
config = spec.to_config()
Per-Subagent Configuration
SubAgentConfig(
name="coder",
description="Writes and reviews code",
instructions="Follow project coding rules.",
context_files=["/CODING_RULES.md"], # Loaded by consumer library
extra={"memory": "project", "cost_budget": 100}, # Custom metadata
)
Architecture
┌─────────────────────────────────────────────────────────┐
│ Parent Agent │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Subagent Toolset │ │
│ │ task() │ check_task() │ answer_subagent() │ │
│ └─────────────────────────────────────────────────┘ │
│ │ │
│ ┌───────────────┼───────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ researcher │ │ writer │ │ coder │ │
│ │ (sync) │ │ (async) │ │ (auto) │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
│ Message Bus (pluggable) │
└─────────────────────────────────────────────────────────┘
Related Projects
| Package | Description |
|---|---|
| Pydantic Deep Agents | Full agent framework (uses this library) |
| pydantic-ai-backend | File storage and Docker sandbox backends |
| pydantic-ai-todo | Task planning toolset |
| summarization-pydantic-ai | Context management processors |
| pydantic-ai | The foundation - agent framework by Pydantic |
Contributing
git clone https://github.com/vstorm-co/subagents-pydantic-ai.git
cd subagents-pydantic-ai
make install
make test # 100% coverage required
make all # lint + typecheck + test
See CONTRIBUTING.md for full guidelines.
License
MIT License - see LICENSE for details.
Yorumlar (0)
Yorum birakmak icin giris yap.
Yorum birakSonuc bulunamadi