Subagents

Subagents allow the main agent to delegate specialized tasks to focused, context-isolated agents.

Why Subagents?

• Focused context - Each subagent has a clear, specific purpose
• Isolation - Fresh todo list, no nested delegation
• Specialization - Expert instructions for specific tasks
• Reduced confusion - Less context = better performance

Defining Subagents

Python

from pydantic_deep import create_deep_agent, SubAgentConfig

subagents = [
    SubAgentConfig(
        name="code-reviewer",
        description="Reviews code for quality, security, and best practices",
        instructions="""
        You are an expert code reviewer. When reviewing code:

        1. Check for security vulnerabilities
        2. Look for performance issues
        3. Verify proper error handling
        4. Assess code readability

        Provide specific, actionable feedback.
        """,
    ),
    SubAgentConfig(
        name="test-writer",
        description="Generates pytest test cases for Python code",
        instructions="""
        You are a testing expert. Generate comprehensive tests:

        - Unit tests for individual functions
        - Edge cases and error conditions
        - Use pytest fixtures and parametrize
        - Include docstrings explaining each test
        """,
    ),
    SubAgentConfig(
        name="doc-writer",
        description="Writes documentation and docstrings",
        instructions="""
        You are a technical writer. Create clear documentation:

        - Use Google-style docstrings
        - Include examples in docstrings
        - Write clear README sections
        - Document edge cases and gotchas
        """,
    ),
]

agent = create_deep_agent(subagents=subagents)

How the Task Tool Works

The main agent can call the task tool:

Python

task(
    description="Review the authentication module for security issues",
    subagent_type="code-reviewer",
)

This:

1. Creates a new agent with the subagent's instructions
2. Clones dependencies with:
3. Same backend (shared files)
4. Empty todo list (isolated planning)
5. No nested subagents
6. Runs the subagent with the description as prompt
7. Returns the subagent's output to the main agent

Context Isolation

Subagents receive isolated context:

Python

def clone_for_subagent(self, max_depth: int = 0) -> DeepAgentDeps:
    """Create deps for a subagent.

    Args:
        max_depth: Allow nested subagents up to this depth (0 = no nesting)
    """
    return DeepAgentDeps(
        backend=self.backend,      # Shared - can read/write same files
        files=self.files,          # Shared reference
        todos=[],                  # Fresh - subagent plans independently
        subagents=self.subagents.copy() if max_depth > 0 else {},  # Nested delegation if allowed
        uploads=self.uploads,      # Shared uploads
    )

This prevents:

• Context bloat from accumulated todos
• Infinite recursion from nested delegation (unless explicitly allowed)
• Confusion from mixed responsibilities

General-Purpose Subagent

By default, a general-purpose subagent is included:

Python

agent = create_deep_agent(
    include_general_purpose_subagent=True,  # Default: True
)

This allows delegation for tasks without a specialized subagent:

Python

task(
    description="Research the best Python logging libraries",
    subagent_type="general-purpose",
)

Disable if you only want specific subagents:

Python

agent = create_deep_agent(
    subagents=subagents,
    include_general_purpose_subagent=False,
)

Custom Model per Subagent

Use different models for different subagents:

Python

subagents = [
    SubAgentConfig(
        name="code-reviewer",
        description="Reviews code (uses powerful model)",
        instructions="...",
        model="openai:gpt-4.1",
    ),
    SubAgentConfig(
        name="simple-formatter",
        description="Formats code (uses fast model)",
        instructions="...",
        model="anthropic:claude-3-haiku-20240307",
    ),
]

Custom Toolsets per Subagent

Subagents can have custom toolsets or agent configuration:

Python

from pydantic_ai.toolsets import FunctionToolset
from pydantic_ai.common_tools import BuitinTools

# Custom toolset
test_toolset = FunctionToolset[Any](id="test-tools")

@test_toolset.tool
async def run_tests(ctx, path: str) -> str:
    """Run pytest on the given path."""
    ...

subagents = [
    SubAgentConfig(
        name="test-writer",
        description="Writes and runs tests",
        instructions="...",
        toolsets=[test_toolset],  # Custom toolsets
    ),
    SubAgentConfig(
        name="researcher",
        description="Researches topics using web search",
        instructions="...",
        agent_kwargs={"builtin_tools": [BuitinTools.web_search]},  # Built-in tools
    ),
]

Example: Code Review Pipeline

Python

import asyncio
from pydantic_deep import create_deep_agent, DeepAgentDeps, StateBackend, SubAgentConfig

async def main():
    subagents = [
        SubAgentConfig(
            name="code-reviewer",
            description="Reviews code for issues",
            instructions="""
            Review code thoroughly. Check for:
            - Security issues
            - Performance problems
            - Error handling
            - Code style

            Format your review as markdown with sections.
            """,
        ),
        SubAgentConfig(
            name="test-writer",
            description="Generates pytest tests",
            instructions="""
            Write comprehensive pytest tests.
            Cover happy paths, edge cases, and error conditions.
            Use fixtures and parametrize decorators.
            """,
        ),
    ]

    agent = create_deep_agent(subagents=subagents)
    deps = DeepAgentDeps(backend=StateBackend())

    # Create some code to review
    deps.backend.write("/src/auth.py", '''
def authenticate(username, password):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    user = db.execute(query)
    if user and user.password == password:
        return True
    return False
    ''')

    result = await agent.run(
        """
        1. Review /src/auth.py for security issues
        2. Generate tests for the authenticate function
        3. Report findings
        """,
        deps=deps,
    )

    print(result.output)

asyncio.run(main())

Best Practices

1. Clear Descriptions

The description helps the main agent choose:

Python

# Good - clear when to use
SubAgentConfig(
    name="security-reviewer",
    description="Reviews code specifically for security vulnerabilities like SQL injection, XSS, and authentication issues",
    ...
)

# Bad - too vague
SubAgentConfig(
    name="reviewer",
    description="Reviews code",
    ...
)

2. Focused Instructions

Keep subagent instructions focused:

Python

# Good - specific focus
instructions="""
You are a security expert. Focus ONLY on:
- SQL injection
- XSS vulnerabilities
- Authentication issues
- Authorization flaws

Do NOT comment on code style or performance.
"""

# Bad - too broad
instructions="Review the code and check everything."

3. Output Format

Specify expected output format:

Python

instructions="""
...

Output your review in this format:
## Summary
[1-2 sentence overview]

## Critical Issues
- [Issue 1]
- [Issue 2]

## Recommendations
- [Recommendation 1]
"""

4. Limited Subagents

Use 3-5 focused subagents, not many generic ones:

Python

# Good - focused experts
subagents = [
    SubAgentConfig(name="security-reviewer", ...),
    SubAgentConfig(name="test-writer", ...),
    SubAgentConfig(name="doc-writer", ...),
]

# Bad - too many similar agents
subagents = [
    SubAgentConfig(name="python-reviewer", ...),
    SubAgentConfig(name="javascript-reviewer", ...),
    SubAgentConfig(name="typescript-reviewer", ...),
    SubAgentConfig(name="go-reviewer", ...),
    # ... 10 more language-specific reviewers
]

Dual-Mode Execution

Subagents support both synchronous (blocking) and asynchronous (background) execution. This is provided by subagents-pydantic-ai.

Execution Modes

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

Python

# Sync execution (default) - blocks until done
task(description="Quick code review", subagent_type="code-reviewer", mode="sync")

# Async execution - returns task handle immediately
task(description="Complex analysis", subagent_type="analyzer", mode="async")

# Auto mode - decides based on task complexity
task(
    description="Process large dataset",
    subagent_type="data-processor",
    mode="auto",
    complexity="complex",  # Hints for auto-mode decision
)

Task Management Tools

When using async mode, additional tools are available:

Tool	Description
check_task	Check status and get results of a background task
list_active_tasks	List all running/pending tasks
soft_cancel_task	Request graceful cancellation
hard_cancel_task	Force immediate cancellation

Subagent Communication

Subagents can ask questions to the parent agent using the ask_parent tool. This enables clarification without returning incomplete results.

Python

SubAgentConfig(
    name="researcher",
    description="Research with clarification",
    instructions="""
    You are a researcher. If you need clarification:
    - Use ask_parent() to ask the main agent
    - Wait for the response before proceeding
    - You can ask up to 3 questions per task
    """,
    can_ask_questions=True,   # Enable ask_parent tool
    max_questions=3,          # Limit questions per task
)

How it works:

1. Subagent encounters ambiguity and calls ask_parent("Which database should I query?")
2. Parent agent receives the question and formulates a response
3. Response is returned to the subagent
4. Subagent continues with the clarified information

Configuration options:

Parameter	Type	Default	Description
can_ask_questions	bool	False	Enable the ask_parent tool
max_questions	int	3	Maximum questions per task execution

Example with question handling:

Python

subagents = [
    SubAgentConfig(
        name="data-analyst",
        description="Analyzes data with clarification when needed",
        instructions="""
        Analyze the requested data. If unclear about:
        - Which columns to analyze
        - Time range to consider
        - Aggregation method to use

        Ask the parent agent for clarification using ask_parent().
        """,
        can_ask_questions=True,
        max_questions=5,
    ),
]

agent = create_deep_agent(subagents=subagents)

# When running, the data-analyst can ask:
# ask_parent("The dataset has 'date' and 'timestamp' columns. Which should I use for the time series?")

Best practices:

• Enable questions for complex analytical tasks
• Keep max_questions low (3-5) to prevent loops
• Include guidance in instructions about when to ask
• Subagent should ask before making assumptions

Next Steps

• Streaming - Monitor subagent progress
• Examples - More examples
• API Reference - SubAgentToolset API