Skip to main content
The Subagent Delegator provides primitives for spawning subagents with scoped permissions and context, enabling complex multi-agent workflows.

Overview

Subagent delegation allows a primary agent to:
  • Spawn specialized subagents for specific tasks
  • Control concurrency and resource limits
  • Scope permissions per subagent
  • Collect and aggregate results

Quick Start

from praisonaiagents import Agent

# Quick Start
agent = Agent(instructions="You are a helpful assistant")
agent.start("Hello!")
from praisonaiagents.agents.delegator import SubagentDelegator, DelegationConfig
import asyncio

# Create delegator
config = DelegationConfig(
    max_concurrent_subagents=3,
    max_total_subagents=10,
    default_timeout_seconds=300
)
delegator = SubagentDelegator(config=config)

# Delegate a task
async def run():
    result = await delegator.delegate(
        agent_name="explorer",
        objective="Find all authentication-related files",
        context={"workspace": "/path/to/project"}
    )
    
    if result.success:
        print(f"Result: {result.result}")
    else:
        print(f"Error: {result.error}")

asyncio.run(run())

Available Agents

Get list of available agents for delegation:
agents = delegator.get_available_agents()
# ['explorer', 'coder', 'planner', 'reviewer', 'debugger']

# Get description
desc = delegator.get_agent_description("explorer")
print(desc)
# "Fast read-only agent for codebase investigation..."

Built-in Agent Profiles

AgentModeDescription
generalPrimaryGeneral-purpose coding assistant
coderAllFocused code implementation
plannerSubagentTask planning and decomposition
reviewerSubagentCode review and quality
explorerSubagentRead-only codebase investigation
debuggerSubagentDebugging and troubleshooting

Explorer Agent

The explorer agent is a specialized read-only agent for codebase investigation:
result = await delegator.delegate(
    agent_name="explorer",
    objective="Find where authentication is implemented",
    context={
        "workspace": "/path/to/project",
        "thoroughness": "very thorough"
    }
)
The explorer agent:
  • Uses only read-only tools (read_file, grep, glob, list_files)
  • Cannot modify files or execute commands
  • Provides structured reports with file locations and insights

Parallel Delegation

Delegate multiple tasks in parallel:
results = await delegator.delegate_parallel([
    ("explorer", "Find authentication files"),
    ("explorer", "Find database models"),
    ("explorer", "Find API endpoints"),
])

for result in results:
    print(f"{result.agent_name}: {result.success}")
With context:
results = await delegator.delegate_parallel([
    ("explorer", "Find auth files", {"workspace": "/project"}),
    ("reviewer", "Review security", {"files": ["auth.py"]}),
])

Configuration

from praisonaiagents.agents.delegator import DelegationConfig

config = DelegationConfig(
    # Concurrency limits
    max_concurrent_subagents=3,    # Max running at once
    max_total_subagents=10,        # Max total spawned
    
    # Timeouts
    default_timeout_seconds=300.0, # Default timeout
    max_timeout_seconds=600.0,     # Maximum allowed timeout
    
    # Resource limits
    max_steps_per_subagent=50,     # Max steps per subagent
    max_tokens_per_subagent=50000, # Max tokens per subagent
    
    # Permissions
    inherit_permissions=True,      # Inherit from parent
    allow_nested_delegation=False, # Allow subagents to delegate
    
    # Behavior
    auto_cancel_on_parent_cancel=True,
    collect_results=True,
)

Task Management

Cancel a Task

# Cancel specific task
success = await delegator.cancel_task(task_id)

# Cancel all running tasks
cancelled_count = await delegator.cancel_all()

Check Task Status

status = delegator.get_task_status(task_id)
# DelegationStatus.RUNNING, COMPLETED, FAILED, etc.

# Get all running tasks
running = delegator.get_running_tasks()

Statistics

stats = delegator.get_stats()
print(f"Total tasks: {stats['total_tasks']}")
print(f"Running: {stats['running_tasks']}")
print(f"Status counts: {stats['status_counts']}")

Custom Agent Factory

Provide a custom agent factory for more control:
from praisonaiagents import Agent

def my_agent_factory(agent_name: str) -> Agent:
    if agent_name == "explorer":
        return Agent(
            name="explorer",
            instructions="You are a codebase explorer...",
            tools=[read_file, grep, glob],
            llm="gpt-4o-mini"
        )
    # ... other agents

delegator = SubagentDelegator(
    config=config,
    agent_factory=my_agent_factory
)

Callbacks

Register callbacks for task completion:
def on_task_complete(result):
    print(f"Task {result.task_id} completed")
    print(f"Agent: {result.agent_name}")
    print(f"Success: {result.success}")
    print(f"Duration: {result.duration_seconds}s")

delegator = SubagentDelegator(
    config=config,
    on_task_complete=on_task_complete
)

Convenience Function

Quick delegation without creating a delegator:
from praisonaiagents.agents.delegator import delegate_to_agent

result = await delegate_to_agent(
    agent_name="explorer",
    objective="Find all Python files",
    context={"workspace": "/project"},
    timeout_seconds=60
)

Best Practices

  1. Use appropriate agents - Match agent capabilities to task requirements
  2. Set timeouts - Prevent runaway subagents
  3. Limit concurrency - Don’t overwhelm resources
  4. Scope permissions - Use read-only agents when possible
  5. Handle failures - Check result.success and handle errors