subagents-pydantic-ai 0.0.1

Subagent toolset for pydantic-ai with dual-mode execution and dynamic agent creation

subagents-pydantic-ai

Subagent delegation toolset for pydantic-ai with dual-mode execution.

Looking for a complete agent framework? Check out pydantic-deep - a full-featured agent framework with planning, subagents, and skills system.

Need file operations? Check out pydantic-ai-backend - file storage and sandbox backends for pydantic-ai agents.

Documentation

Full Documentation - Installation, concepts, examples, and API reference.

Architecture

%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#6366f1', 'primaryTextColor': '#fff', 'primaryBorderColor': '#4f46e5', 'lineColor': '#94a3b8', 'secondaryColor': '#22c55e', 'tertiaryColor': '#1e293b', 'background': '#0f172a', 'mainBkg': '#1e293b', 'textColor': '#e2e8f0', 'nodeTextColor': '#e2e8f0'}}}%%
flowchart LR
    subgraph parent [" Parent Agent "]
        direction TB
        PA["🤖 pydantic-ai Agent"]
        TS["🔧 Subagent Toolset"]
        PA --> TS
    end

    subgraph tools [" Tools "]
        direction TB
        T1["task()"]
        T2["check_task()"]
        T3["answer_subagent()"]
        T4["cancel_task()"]
    end

    subgraph agents [" Specialized Subagents "]
        direction TB
        S1["🔍 researcher"]
        S2["✍️ writer"]
        S3["💻 coder"]
        S4["🔧 general"]
    end

    TS --> tools
    tools -->|"sync / async"| agents

Installation

pip install subagents-pydantic-ai

Or with uv:

uv add subagents-pydantic-ai

Quick Start

from dataclasses import dataclass, field
from typing import Any
from pydantic_ai import Agent
from subagents_pydantic_ai import create_subagent_toolset, SubAgentConfig

# Dependencies must implement SubAgentDepsProtocol
@dataclass
class Deps:
    subagents: dict[str, Any] = field(default_factory=dict)

    def clone_for_subagent(self, max_depth: int = 0) -> "Deps":
        return Deps(subagents={} if max_depth <= 0 else self.subagents.copy())

# Define specialized subagents
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.",
    ),
]

# Create toolset and agent
toolset = create_subagent_toolset(subagents=subagents)
agent = Agent(
    "openai:gpt-4o",
    deps_type=Deps,
    toolsets=[toolset],
    system_prompt="You can delegate tasks to specialized subagents.",
)

# Run the agent
result = agent.run_sync(
    "Research Python async patterns and write a blog post about it",
    deps=Deps(),
)
print(result.output)

Dual-Mode Execution

%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#6366f1', 'actorBkg': '#6366f1', 'actorTextColor': '#fff', 'actorLineColor': '#94a3b8', 'signalColor': '#e2e8f0', 'signalTextColor': '#e2e8f0', 'labelBoxBkgColor': '#1e293b', 'labelBoxBorderColor': '#475569', 'labelTextColor': '#e2e8f0', 'loopTextColor': '#e2e8f0', 'noteBkgColor': '#334155', 'noteTextColor': '#e2e8f0', 'noteBorderColor': '#475569', 'activationBkgColor': '#4f46e5', 'sequenceNumberColor': '#fff'}}}%%
sequenceDiagram
    participant P as 🤖 Parent
    participant S as 🔧 Subagent

    rect rgba(99, 102, 241, 0.2)
        Note over P,S: 🔄 Sync Mode (default)
        P->>+S: task(mode="sync")
        S-->>S: working...
        S->>-P: ✅ result
    end

    rect rgba(34, 197, 94, 0.2)
        Note over P,S: ⚡ Async Mode
        P->>+S: task(mode="async")
        S-->>P: 🎫 task_id
        Note over P: continues working...
        S-->>S: working in background...
        P->>S: check_task(id)
        S->>-P: ✅ result
    end

Sync Mode (Default)

Block until the subagent completes:

# The agent uses task() with mode="sync" (default)
# - Quick tasks
# - When result is needed immediately
# - Back-and-forth communication

Async Mode

Run in background, continue working:

# The agent uses task() with mode="async"
# - Long-running research
# - Parallel tasks
# - When immediate result isn't needed

Give Subagents Tools

Provide toolsets to your subagents:

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 at runtime:

from subagents_pydantic_ai import (
    create_subagent_toolset,
    create_agent_factory_toolset,
    DynamicAgentRegistry,
)

registry = DynamicAgentRegistry()

# Main agent can create new specialized agents on-the-fly
agent = Agent(
    "openai:gpt-4o",
    deps_type=Deps,
    toolsets=[
        create_subagent_toolset(),
        create_agent_factory_toolset(
            registry=registry,
            allowed_models=["openai:gpt-4o", "openai:gpt-4o-mini"],
            max_agents=5,
        ),
    ],
)

Available Tools

Tool	Description
task	Delegate a task to a subagent (sync or async)
check_task	Check status of a background task
answer_subagent	Answer a question from a subagent
list_active_tasks	List all running background tasks
soft_cancel_task	Request cooperative cancellation
hard_cancel_task	Immediately cancel a task

Related Projects

• pydantic-ai - Agent framework by Pydantic
• pydantic-deep - Full agent framework (uses this library)
• pydantic-ai-backend - File storage and sandbox backends
• pydantic-ai-todo - Task planning toolset

Development

git clone https://github.com/vstorm-co/subagents-pydantic-ai.git
cd subagents-pydantic-ai
make install
make test

License

MIT License - see LICENSE for details.