Handoff Configuration

Complete reference for handoff filters and advanced delegation settings

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

Context sharing (Python API)

HandoffConfig controls what context the target sees. The filtered messages are seeded onto the target agent’s chat_history for the duration of the handoff, then restored on exit.

The context_policy, max_context_messages, and preserve_system knobs live on HandoffConfig; input_filter is passed to handoff() (or Handoff) directly.

Python

from praisonaiagents import Agent, handoff, HandoffConfig, ContextPolicy, handoff_filters

specialist = Agent(name="Specialist", instructions="Handle the escalated task.")

config = HandoffConfig(
    context_policy=ContextPolicy.LAST_N,   # FULL | LAST_N | SUMMARY | NONE
    max_context_messages=10,               # used by LAST_N
    preserve_system=True,                  # keep system messages in the shared context
)

coordinator = Agent(
    name="Coordinator",
    instructions="Escalate hard cases.",
    handoffs=[
        handoff(
            specialist,
            config=config,
            input_filter=handoff_filters.remove_all_tools,  # callable or list of callables
        )
    ],
)

TypeScript

import { Agent, handoff } from 'praisonai';

const specialist = new Agent({ name: 'Specialist', instructions: 'Handle the escalated task.' });

const coordinator = new Agent({
  name: 'Coordinator',
  instructions: 'Escalate hard cases.',
  handoffs: [
    handoff({
      agent: specialist,
      contextPolicy: 'last_n',   // 'full' | 'last_n' | 'summary' | 'none'
      maxContextMessages: 10,    // used by 'last_n'
      preserveSystem: true,      // keep system messages in the shared context
    }),
  ],
});

TypeScript also honours maxContextTokens — unlike Python, which declares the field but never reads it. See TypeScript Handoffs.

HandoffConfig fields:

Option	Type	Default	Description
context_policy	ContextPolicy	ContextPolicy.SUMMARY	How much history to share: FULL, LAST_N, SUMMARY, or NONE
max_context_tokens	int	4000	Token cap on shared context
max_context_messages	int	10	Message cap used by LAST_N
preserve_system	bool	True	Keep system messages when building the shared context

input_filter is a handoff() / Handoff argument — a callable (or list of callables) applied to the shared messages before they reach the target.

A handoff’s seeded history is invocation-scoped — it does not leak into subsequent chats or later handoffs that reuse the same target, and sequential handoffs do not accumulate context. See Handoff Filters.

On releases before PR #4213, context_policy, max_context_messages, preserve_system, and input_filter had no runtime effect — every handoff behaved as if there were no context sharing. If you “tuned” these on an older release and saw no change, they now take effect.

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

Per-Instance Concurrency (max_concurrent)

max_concurrent caps how many copies of a single handoff run in parallel. Each Handoff enforces its own limit — different handoffs on the same coordinator can carry different limits, each enforced independently.

from praisonaiagents import Agent, handoff, HandoffConfig

researcher = Agent(name="Researcher", instructions="Gather sources.")
writer     = Agent(name="Writer",     instructions="Draft the report.")

research_handoff = handoff(researcher, config=HandoffConfig(max_concurrent=10))
writer_handoff   = handoff(writer,     config=HandoffConfig(max_concurrent=1))

coordinator = Agent(
    name="Coordinator",
    handoffs=[research_handoff, writer_handoff],
)
# Up to 10 research handoffs may run in parallel; only 1 writer handoff at a time.

As of PR #3515, max_concurrent is enforced per Handoff. Previously the semaphore was class-level, so only the first handoff’s limit took effect and every subsequent handoff silently shared it regardless of its own max_concurrent.

Handoffs and asyncio: the concurrency semaphore is now created per instance and rebound when the running event loop changes. Repeated asyncio.run() calls in the same process — common in test suites and notebooks — no longer raise RuntimeError: … bound to a different event loop.

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

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

See Also

• Handoff Concepts - Understanding handoff patterns
• Agent Configuration - Agent handoff settings
• Best Practices - Configuration guidelines