Handoff Configuration

This page provides comprehensive documentation for configuring handoffs in PraisonAI, including handoff filters, delegation strategies, routing rules, and advanced orchestration patterns.

Handoff System Overview

The handoff system enables agents to delegate tasks to other agents based on expertise, availability, or specific conditions. This creates flexible multi-agent workflows with intelligent task routing.

Basic Handoff Configuration

from praisonaiagents import Agent, Handoff

# Basic handoff configuration
handoff = Handoff(
    name="research_handoff",
    target="research_agent",
    description="Hand off research tasks to specialized researcher",
    condition="task requires research"
)

agent = Agent(
    name="Coordinator",
    handoffs=[handoff],
    handoff_config={
        "strategy": "best_match",
        "timeout": 30,
        "fallback_behavior": "handle_self"
    }
)

YAML Configuration

Configure handoffs directly in YAML files using the nested handoff: block:

agents:
  coordinator:
    role: Coordinator
    handoff:
      to: [writer, reviewer]
      policy: summary
      timeout: 120
      max_depth: 5
      max_concurrent: 3
      detect_cycles: true

Handoff Policy Options

Policy	Description
`full`	Share complete conversation history
`summary`	Share summarized context (default)
`none`	No context sharing
`last_n`	Share last N messages

Handoff Filters

Filter Types and Configuration

# Expertise-based filter
expertise_filter = {
    "type": "expertise",
    "required_skills": ["python", "data_analysis"],
    "minimum_skill_level": 0.8,
    "match_all": True  # Require all skills
}

# Availability filter
availability_filter = {
    "type": "availability",
    "max_queue_size": 5,
    "max_active_tasks": 3,
    "estimated_duration": 300,  # 5 minutes
    "check_schedule": True
}

# Performance filter
performance_filter = {
    "type": "performance",
    "min_success_rate": 0.9,
    "max_avg_duration": 120,
    "recent_window": 100,  # Last 100 tasks
    "exclude_outliers": True
}

# Resource filter
resource_filter = {
    "type": "resource",
    "required_memory": 2048,  # MB
    "required_cpu": 2,        # cores
    "required_gpu": False,
    "check_availability": True
}

# Custom filter
custom_filter = {
    "type": "custom",
    "function": "evaluate_agent_compatibility",
    "parameters": {
        "task_complexity": "high",
        "domain": "finance"
    }
}

Composite Filters

# Combine multiple filters with AND/OR logic
composite_filter = {
    "type": "composite",
    "operator": "AND",
    "filters": [
        {
            "type": "expertise",
            "required_skills": ["machine_learning"],
            "minimum_skill_level": 0.7
        },
        {
            "type": "composite",
            "operator": "OR",
            "filters": [
                {
                    "type": "availability",
                    "max_queue_size": 3
                },
                {
                    "type": "performance",
                    "min_success_rate": 0.95
                }
            ]
        }
    ]
}

Dynamic Filter Functions

def dynamic_agent_filter(agent, task, context):
    """Custom filter function for complex logic"""
    # Check agent capabilities
    if not agent.has_capability(task.required_capability):
        return False
    
    # Check workload
    if agent.current_load > 0.8:
        return False
    
    # Check domain expertise
    domain_score = agent.get_domain_score(task.domain)
    if domain_score < context.get("min_domain_score", 0.7):
        return False
    
    # Check historical performance
    if task.priority == "high":
        success_rate = agent.get_success_rate(task.type)
        if success_rate < 0.9:
            return False
    
    return True

# Use dynamic filter
handoff_config = {
    "filters": [
        {
            "type": "dynamic",
            "function": dynamic_agent_filter,
            "context": {
                "min_domain_score": 0.8
            }
        }
    ]
}

Advanced Delegation Settings

Delegation Strategies

# Best match strategy - Select most suitable agent
best_match_config = {
    "strategy": "best_match",
    "scoring": {
        "expertise_weight": 0.4,
        "availability_weight": 0.3,
        "performance_weight": 0.3
    },
    "min_score": 0.7,
    "tie_breaker": "least_loaded"
}

# Round robin strategy - Distribute evenly
round_robin_config = {
    "strategy": "round_robin",
    "skip_unavailable": True,
    "sticky_sessions": True,  # Same agent for related tasks
    "session_duration": 3600  # 1 hour
}

# Load balancing strategy
load_balancing_config = {
    "strategy": "load_balance",
    "algorithm": "weighted_least_connections",
    "weights": {
        "agent_1": 3,
        "agent_2": 2,
        "agent_3": 1
    },
    "health_check_interval": 30
}

# Priority-based strategy
priority_config = {
    "strategy": "priority",
    "agent_priorities": {
        "expert_agent": 1,
        "senior_agent": 2,
        "junior_agent": 3
    },
    "escalation_enabled": True,
    "escalation_timeout": 60
}

Conditional Handoffs

# Task-based conditions
task_condition_handoff = Handoff(
    name="task_router",
    condition={
        "type": "task_properties",
        "rules": [
            {
                "if": {"task_type": "analysis", "complexity": "high"},
                "then": {"target": "senior_analyst", "priority": "high"}
            },
            {
                "if": {"task_type": "analysis", "complexity": "low"},
                "then": {"target": "junior_analyst", "priority": "normal"}
            },
            {
                "if": {"task_type": "research"},
                "then": {"target": "research_team", "strategy": "round_robin"}
            }
        ],
        "default": {"target": "general_agent"}
    }
)

# Context-based conditions
context_condition_handoff = Handoff(
    name="context_router",
    condition={
        "type": "context_evaluation",
        "evaluator": "evaluate_context",
        "rules": {
            "high_value_customer": {
                "target": "senior_support",
                "sla": 300  # 5 minute response
            },
            "technical_issue": {
                "target": "tech_support",
                "include_context": True
            },
            "billing_query": {
                "target": "billing_team",
                "secure_handoff": True
            }
        }
    }
)

Handoff Chains and Workflows

# Sequential handoff chain
sequential_handoff = {
    "type": "sequential",
    "chain": [
        {
            "agent": "validator",
            "action": "validate_input",
            "on_success": "next",
            "on_failure": "abort"
        },
        {
            "agent": "processor",
            "action": "process_data",
            "on_success": "next",
            "on_failure": "retry"
        },
        {
            "agent": "reviewer",
            "action": "review_output",
            "on_success": "complete",
            "on_failure": "escalate"
        }
    ],
    "timeout": 600,
    "preserve_context": True
}

# Parallel handoff
parallel_handoff = {
    "type": "parallel",
    "targets": ["analyzer_1", "analyzer_2", "analyzer_3"],
    "aggregation": "consensus",  # or "merge", "first", "all"
    "min_responses": 2,
    "timeout": 120,
    "continue_on_partial": True
}

# Conditional branching
branching_handoff = {
    "type": "branching",
    "condition": "evaluate_priority",
    "branches": {
        "urgent": {
            "target": "emergency_team",
            "escalate_after": 60
        },
        "high": {
            "target": "senior_team",
            "queue_position": "front"
        },
        "normal": {
            "target": "standard_team",
            "queue_position": "back"
        }
    }
}

Handoff Routing Rules

Static Routing

static_routing = {
    "routes": {
        "customer_support": ["support_agent_1", "support_agent_2"],
        "technical": ["tech_agent_1", "tech_agent_2", "tech_agent_3"],
        "sales": ["sales_agent_1"],
        "general": ["general_agent_1", "general_agent_2"]
    },
    "selection_method": "least_busy",
    "fallback_route": "general"
}

Dynamic Routing

def dynamic_router(task, agents, context):
    """Dynamic routing based on real-time conditions"""
    # Get task requirements
    required_skills = task.get_required_skills()
    urgency = task.get_urgency()
    
    # Score each agent
    scores = {}
    for agent in agents:
        score = 0
        
        # Skill match
        skill_match = agent.match_skills(required_skills)
        score += skill_match * 0.4
        
        # Availability
        availability = agent.get_availability()
        score += availability * 0.3
        
        # Performance history
        performance = agent.get_performance_score(task.type)
        score += performance * 0.3
        
        # Urgency bonus
        if urgency == "high" and agent.priority_capable:
            score += 0.2
        
        scores[agent.name] = score
    
    # Select best agent
    best_agent = max(scores, key=scores.get)
    return best_agent if scores[best_agent] > 0.6 else None

dynamic_routing_config = {
    "router_function": dynamic_router,
    "refresh_interval": 30,  # Re-evaluate every 30 seconds
    "cache_decisions": True,
    "cache_ttl": 60
}

Rule-Based Routing

rule_based_routing = {
    "rules": [
        {
            "name": "vip_customer_rule",
            "condition": "customer.tier == 'VIP'",
            "route": "senior_support",
            "priority": 1
        },
        {
            "name": "technical_issue_rule",
            "condition": "issue.category == 'technical' and issue.severity > 7",
            "route": "tech_specialist",
            "priority": 2
        },
        {
            "name": "language_rule",
            "condition": "customer.language != 'english'",
            "route": "multilingual_support",
            "priority": 3
        }
    ],
    "evaluation_order": "priority",  # or "sequential", "parallel"
    "stop_on_match": True
}

Tool Policy

HandoffConfig.tool_policy (HandoffToolPolicy) enforces tool boundaries when one agent hands off to another. Default mode is intersect (secure): the target receives only tools shared with the source, minus any blocked_tools.

from praisonaiagents import Agent, handoff, HandoffConfig, HandoffToolPolicy

config = HandoffConfig(
    tool_policy=HandoffToolPolicy(
        mode="intersect",
        blocked_tools=["execute_code"],
    ),
)

router = Agent(
    name="Router",
    handoffs=[handoff(automation_agent, config=config)],
)

The handoff() factory also accepts shorthand kwargs:

Parameter	Type	Default	Description
`tool_policy_mode`	`Optional[Literal["intersect","passthrough"]]`	`None`	Shorthand for `config.tool_policy.mode`
`blocked_tools`	`Optional[List[str]]`	`None`	Shorthand for `config.tool_policy.blocked_tools`

See Handoff Tool Policy for modes, patterns, and migration guidance.

Handoff Security and Validation

Tool policy (HandoffConfig.tool_policy) is the primary tool-level security mechanism during handoffs. The settings below cover transport, authorisation, and payload validation.

security_config = {
    "authentication": {
        "enabled": True,
        "method": "token",  # or "certificate", "mutual_tls"
        "token_ttl": 3600
    },
    
    "authorization": {
        "enabled": True,
        "check_permissions": True,
        "role_based": True,
        "resource_based": True
    },
    
    "data_handling": {
        "encrypt_in_transit": True,
        "sanitize_sensitive": True,
        "audit_trail": True,
        "data_retention": 90  # days
    },
    
    "validation": {
        "validate_schema": True,
        "validate_constraints": True,
        "max_payload_size": 1048576  # 1MB
    }
}

Performance Optimization

Caching and Optimization

optimization_config = {
    "caching": {
        "cache_routing_decisions": True,
        "cache_ttl": 300,
        "cache_size": 1000,
        "invalidation_strategy": "ttl"  # or "lru", "event_based"
    },
    
    "batching": {
        "enable_batching": True,
        "batch_size": 10,
        "batch_timeout": 1.0,
        "group_by": ["target_agent", "priority"]
    },
    
    "connection_pooling": {
        "pool_size": 20,
        "max_overflow": 10,
        "pool_timeout": 30,
        "recycle": 3600
    },
    
    "monitoring": {
        "track_latency": True,
        "track_success_rate": True,
        "alert_thresholds": {
            "latency_p95": 1000,  # ms
            "error_rate": 0.05
        }
    }
}

Complete Handoff Configuration Example

from praisonaiagents import Agent, Handoff

# Create comprehensive handoff configuration
coordinator = Agent(
    name="MasterCoordinator",
    handoffs=[
        Handoff(
            name="research_handoff",
            target="research_team",
            description="Complex research tasks",
            filters=[
                {
                    "type": "expertise",
                    "required_skills": ["research", "analysis"],
                    "minimum_skill_level": 0.8
                }
            ]
        ),
        Handoff(
            name="urgent_handoff",
            target="emergency_team",
            description="High-priority urgent tasks",
            condition={"urgency": "critical"},
            priority=1
        )
    ],
    handoff_config={
        # Strategy configuration
        "default_strategy": "best_match",
        "fallback_strategy": "round_robin",
        
        # Filter configuration
        "global_filters": [
            {
                "type": "availability",
                "max_queue_size": 10
            }
        ],
        
        # Routing configuration
        "routing": {
            "type": "hybrid",
            "static_routes": {
                "research": ["researcher_1", "researcher_2"],
                "analysis": ["analyst_1", "analyst_2"]
            },
            "dynamic_router": "smart_router_function"
        },
        
        # Performance settings
        "timeout": 60,
        "max_retries": 2,
        "retry_delay": 5,
        
        # Security settings
        "secure_handoffs": True,
        "encrypt_data": True,
        
        # Monitoring
        "track_metrics": True,
        "log_handoffs": True
    }
)

Environment Variables

# Handoff strategy
export PRAISONAI_HANDOFF_STRATEGY="best_match"
export PRAISONAI_HANDOFF_TIMEOUT="60"

# Filter settings
export PRAISONAI_HANDOFF_MIN_SCORE="0.7"
export PRAISONAI_HANDOFF_CHECK_AVAILABILITY="true"

# Routing settings
export PRAISONAI_HANDOFF_ROUTING="dynamic"
export PRAISONAI_HANDOFF_CACHE_ROUTES="true"

# Performance settings
export PRAISONAI_HANDOFF_BATCH_SIZE="10"
export PRAISONAI_HANDOFF_MAX_RETRIES="3"

# Security settings
export PRAISONAI_HANDOFF_SECURE="true"
export PRAISONAI_HANDOFF_ENCRYPT="true"

Best Practices

Use appropriate filters to ensure tasks are routed to capable agents
Implement fallback mechanisms for handling failures
Monitor handoff performance and adjust strategies accordingly
Cache routing decisions for frequently occurring patterns
Set reasonable timeouts to prevent indefinite waiting
Implement circuit breakers for unreliable agents
Use batching for high-volume scenarios
Maintain audit trails for compliance and debugging

​Handoff Configuration

​Handoff System Overview

​Basic Handoff Configuration

​YAML Configuration

​Handoff Policy Options

​Handoff Filters

​Filter Types and Configuration

​Composite Filters

​Dynamic Filter Functions

​Advanced Delegation Settings

​Delegation Strategies

​Conditional Handoffs

​Handoff Chains and Workflows

​Handoff Routing Rules

​Static Routing

​Dynamic Routing

​Rule-Based Routing

​Tool Policy

​Handoff Security and Validation

​Performance Optimization

​Caching and Optimization

​Complete Handoff Configuration Example

​Environment Variables

​Best Practices

​See Also