Skip to content

vstorm-co/subagents-pydantic-ai

Repository files navigation

Subagents for Pydantic AI

Subagents for Pydantic AI

Declarative multi-agent orchestration.
Delegate to specialist sub-agents — sync, async, or auto — with token tracking and cancellation.

Docs · PyPI · Install · Ecosystem · Deep Agents

PyPI version PyPI Downloads GitHub Stars Python 3.10+ License: MIT Coverage Status CI Pydantic AI

Sync / async / auto  •  Nested subagents  •  Runtime agent creation  •  Background tasks  •  Token tracking


Part of Pydantic Deep Agents — the open-source Claude Code alternative & Python agent framework. Use this library standalone, or get everything wired together in one create_deep_agent() call.

Subagents for Pydantic AI adds multi-agent delegation to any Pydantic AI agent. Spawn specialist subagents that run synchronously (blocking), asynchronously (background), or let the system auto-select the best mode — with built-in token tracking and cancellation.

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. SubAgentCapability handles 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)                    │
└─────────────────────────────────────────────────────────┘

Vstorm OSS Ecosystem

This library is one piece of a broader open-source toolkit for production AI agents — all built on Pydantic AI.

Project Description Stars
Pydantic Deep Agents The full agent framework and terminal assistant — bundles every library below into one create_deep_agent() call. Stars
pydantic-ai-backend Sandboxed execution & file tools — State / Local / Docker / Daytona backends + console toolset. Stars
👉 subagents-pydantic-ai Declarative multi-agent orchestration — sync / async / auto, with token tracking. Stars
summarization-pydantic-ai Unlimited context for long-running agents — summarization or sliding window. Stars
pydantic-ai-shields Drop-in guardrails — cost caps, prompt-injection defense, PII & secret redaction, tool blocking. Stars
pydantic-ai-todo Task planning with subtasks, dependencies, and cycle detection. Stars
full-stack-ai-agent-template Zero to production AI app in 30 minutes — FastAPI + Next.js 15, RAG, 6 AI frameworks. Stars

Want it all wired together? Pydantic Deep Agents ships every library above integrated — planning, filesystem, subagents, memory, context management, and guardrails — behind a single function call. Browse everything at oss.vstorm.co.

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.

Star History

If this library saved you from wiring an agent harness by hand — give it a ⭐. It's the single biggest thing that helps the project grow.

Star History


License

MIT — see LICENSE


Need help shipping AI agents in production?

We're Vstorm — an Applied Agentic AI Engineering Consultancy
with 30+ production agent implementations. Pydantic Deep Agents is what we build them with.

Talk to us



Made with care by Vstorm

About

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.

Topics

Resources

License

Contributing

Stars

Watchers

Forks

Contributors