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

python

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:

yaml

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

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

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` sends everything, `LAST_N` sends the last `max_context_messages`, `SUMMARY` sends the first user message (the task goal) plus the last 3 messages, `NONE` sends nothing. `LAST_N` and `SUMMARY` drop any orphan `tool`/`tool_calls` message so providers do not reject the request.
`max_context_tokens`	`int`	`4000`	Token cap on shared context. Python declares it but does not read it — only the TypeScript SDK honours it.
`max_context_messages`	`int`	`10`	Message cap used by `LAST_N`. `0` means “no messages”, not “all messages”.
`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.

What `SUMMARY` actually does

SUMMARY is the default. It is not an LLM-generated summary — that stays opt-in via input_filter. It ships:

Any system messages (when preserve_system=True).
The first user message — usually the task goal.
The last 3 messages, trimmed so tool_calls/tool pairs stay intact.

If the first user message is already in that tail, it is not duplicated. LAST_N uses the same tool-pair safe trim, so the target agent never receives an orphan tool result.

python

from praisonaiagents import Agent
from praisonaiagents.agent.handoff import Handoff, HandoffConfig, ContextPolicy

src = Agent(name="src", instructions="x")
tgt = Agent(name="tgt", instructions="y")
src.chat_history = [
    {"role": "system", "content": "sys"},
    {"role": "user", "content": "GOAL: book flight to Paris"},
    {"role": "assistant", "content": "ok, searching"},
    {"role": "assistant", "content": None, "tool_calls": [
        {"id": "c1", "type": "function",
         "function": {"name": "search", "arguments": "{}"}},
    ]},
    {"role": "tool", "tool_call_id": "c1", "content": "results"},
    {"role": "assistant", "content": "found 3 options"},
]

for pol in (ContextPolicy.SUMMARY, ContextPolicy.LAST_N):
    cfg = HandoffConfig(context_policy=pol, max_context_messages=2)
    msgs = Handoff(agent=tgt, config=cfg)._prepare_context(src, {}).messages
    print(pol.value, [m["role"] for m in msgs])

# summary ['system', 'user', 'assistant', 'tool', 'assistant']  # goal kept, pair intact
# last_n  ['system', 'assistant']                               # orphan tool dropped

Handoff Filters

Filter Types and Configuration

python

# 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

python

# 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

python

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.

python

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.

Delegation Strategies

python

# 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

python

# 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

python

# 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

python

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

python

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

python

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.

python

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.

Error Handling

A failed handoff (cycle, depth, or exception) returns a dict from the handoff tool, so tool-failure handling sees a failed call.

python

{"error": "Maximum handoff depth exceeded - ...", "handoff_failed": True}

Check result.get("handoff_failed") in hooks. Before PraisonAI #5772, these returned an "Error: ..." string.

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.

python

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

python

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

python

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

bash

# 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