agent-handoff-protocols

Agent-to-Agent Handoff Protocols

Overview

In a multi-agent system, agents need to hand off tasks — and context — to each other seamlessly. A broken handoff means lost context, frustrated users, and failed workflows. This skill covers structured protocols for passing control between agents, handling escalations, and maintaining continuity across agent boundaries.

---

Core Concepts

When Handoffs Happen

Scenario	From	To	Why
Escalation	Tier-1 agent	Tier-2 specialist	Task exceeds capability
Specialization	Router agent	Domain expert	Task matches expertise
Supervision	Sub-agent	Supervisor	Needs approval or guidance
Recovery	Failed agent	Fallback agent	Primary agent broken
Load shedding	Overloaded agent	Idle agent	Balance workload

Handoff Types

Type	Description	Latency	Risk
Warm Handoff	Full context + current state passed explicitly	Medium	Low — all state transferred
Cold Handoff	Only task description passed, receiving agent starts fresh	Low	High — context loss
Supervised Handoff	Supervisor mediates, validates, then transfers	High	Very Low — human/LLM checks
Broadcast Handoff	All agents notified, first capable claims	Medium	Medium — race conditions
Delegation Handoff	Sender waits for result	High	Low — synchronous, traceable

---

Step-by-Step Implementation

Step 1: Define the Handoff Contract

from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
import json
import time

class HandoffReason(Enum):
    ESCALATION = "escalation"
    SPECIALIZATION = "specialization"
    RECOVERY = "recovery"
    LOAD_SHEDDING = "load_shedding"
    SUPERVISION = "supervision"

@dataclass
class HandoffContext:
    """Complete context transferred between agents."""
    
    # Identity
    source_agent: str
    target_agent: str
    handoff_id: str
    
    # The task
    task_id: str
    original_task: str
    current_state: str  # What has been done so far
    
    # Conversation history (condensed)
    conversation_summary: str
    key_facts: list[str] = field(default_factory=list)
    decisions_made: list[str] = field(default_factory=list)
    
    # State
    collected_data: dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0  # How confident source was in resolution
    reason: HandoffReason = HandoffReason.SPECIALIZATION
    
    # Metadata
    created_at: float = None
    expires_at: Optional[float] = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = time.time()
    
    def serialize(self) -> str:
        """Serialize to JSON for transport."""
        return json.dumps({
            "source_agent": self.source_agent,
            "target_agent": self.target_agent,
            "handoff_id": self.handoff_id,
            "task_id": self.task_id,
            "original_task": self.original_task,
            "current_state": self.current_state,
            "conversation_summary": self.conversation_summary,
            "key_facts": self.key_facts,
            "decisions_made": self.decisions_made,
            "collected_data": self.collected_data,
            "confidence": self.confidence,
            "reason": self.reason.value,
            "created_at": self.created_at,
        })
    
    @classmethod
    def deserialize(cls, data: str) -> "HandoffContext":
        """Deserialize from JSON."""
        obj = json.loads(data)
        obj["reason"] = HandoffReason(obj["reason"])
        return cls(**obj)