Skip to main content

Agent Handoff Patterns: Smooth Workflow Transitions

An agent handoff is the moment when work transitions from one agent to another. Agent A finishes its task, hands off results and context to agent B, and exits. The handoff can be smooth (agent B immediately understands context and continues) or problematic (context is lost, assumptions differ, state is inconsistent). Handoff patterns ensure context transfer, validate state alignment, and provide recovery mechanisms when handoffs fail.

Handoffs are fundamental to multi-agent systems: pipelines are chains of handoffs; hierarchical systems delegate and handoff results; swarms handoff discoveries via the scratchpad. Mastering handoffs is essential for building reliable orchestrations. A poorly designed handoff can cascade failures through the entire system.

Core Elements of a Handoff​

1. Context transfer​

All information agent B needs to continue must be explicitly passed:

{
  "previous_agent": "research_agent",
  "current_agent": "validation_agent",
  "context": {
    "original_query": "What is the capital of France?",
    "research_findings": {
      "claim": "Paris",
      "confidence": 0.95,
      "sources": ["wikipedia", "britannica"]
    },
    "constraints": ["provide citations", "avoid opinion"],
    "deadline": "2026-06-02T11:00:00Z"
  },
  "handoff_metadata": {
    "timestamp": "2026-06-02T10:15:30Z",
    "handoff_id": "hof_001"
  }
}

2. State validation​

Agent B must verify that the state it receives is valid before proceeding:

def validate_handoff_state(context: dict) -> tuple[bool, str]:
    """Verify context is valid for validation agent."""
    if "research_findings" not in context:
        return False, "Missing research findings"
    if not context["research_findings"].get("claim"):
        return False, "Missing claim"
    if context["research_findings"].get("confidence", 0) < 0.3:
        return False, "Confidence too low"
    return True, "Valid"

3. Graceful error handling​

If validation fails, the handoff should not silently fail. Log, escalate, or trigger recovery:

def handle_handoff(previous_agent: str, current_agent: str, context: dict):
    """Execute a handoff with error handling."""
    is_valid, reason = validate_handoff_state(context)
    if not is_valid:
        log_handoff_failure(previous_agent, current_agent, reason)
        # Option 1: Escalate to orchestrator
        # Option 2: Route to a fallback agent
        # Option 3: Return error to user
        raise HandoffError(reason)
    
    # Proceed with handoff
    return current_agent.process(context)

Handoff Pattern Types​

Sequential pipeline handoff​

Agent A → Agent B → Agent C in order. Results flow forward; context flows with them.

[Research] → [Validate] → [Synthesize] → [Output]

Example: Research agent → Validation agent → Response synthesizer.

Conditional handoff​

Based on the result of agent A, route to different agents.

[Router] → if result == "approved" → [Deliver]
        → if result == "review" → [ManualReview]
        → if result == "error" → [ErrorHandler]

Example: Classification agent → domain-specific handlers.

Bidirectional handoff​

Agent A hands off to agent B, which may hand back to A for revision.

[Write] → [Critique] → if issues → [Revise] → [Critique]
                    → if approved → [Finalize]

Example: Content generation with iterative refinement.

Building a Robust Handoff System​

Here is a complete example with error handling and state validation:

import anthropic
import json
from typing import Optional
from datetime import datetime
from enum import Enum

client = anthropic.Anthropic()

class HandoffStatus(Enum):
    SUCCESS = "success"
    FAILED = "failed"
    RETRY = "retry"

class HandoffState:
    """Represents state passed between agents."""
    
    def __init__(self, original_query: str):
        self.original_query = original_query
        self.research_findings = None
        self.validation_result = None
        self.synthesis = None
        self.history = []
    
    def to_dict(self) -> dict:
        return {
            "original_query": self.original_query,
            "research_findings": self.research_findings,
            "validation_result": self.validation_result,
            "synthesis": self.synthesis,
            "history": self.history
        }
    
    def record_handoff(self, from_agent: str, to_agent: str, status: str):
        """Log the handoff for auditing."""
        self.history.append({
            "from": from_agent,
            "to": to_agent,
            "status": status,
            "timestamp": datetime.utcnow().isoformat()
        })

class HandoffAgent:
    """Base class for agents in a handoff pipeline."""
    
    def __init__(self, name: str):
        self.name = name
        self.model = "claude-3-5-sonnet-20241022"
    
    def validate_input(self, state: HandoffState) -> tuple[bool, str]:
        """Override to validate incoming state."""
        return True, "Valid"
    
    def process(self, state: HandoffState) -> HandoffState:
        """Override to implement agent logic."""
        raise NotImplementedError
    
    def handoff_to(self, next_agent: "HandoffAgent", state: HandoffState) -> tuple[HandoffStatus, HandoffState]:
        """Execute handoff to next agent."""
        # Validate incoming state
        is_valid, reason = next_agent.validate_input(state)
        if not is_valid:
            print(f"Handoff from {self.name} to {next_agent.name} FAILED: {reason}")
            state.record_handoff(self.name, next_agent.name, "failed")
            return HandoffStatus.FAILED, state
        
        print(f"Handoff from {self.name} to {next_agent.name} OK")
        state.record_handoff(self.name, next_agent.name, "success")
        
        # Process in next agent
        try:
            result = next_agent.process(state)
            return HandoffStatus.SUCCESS, result
        except Exception as e:
            print(f"Processing error in {next_agent.name}: {e}")
            state.record_handoff(next_agent.name, "error_handler", "failed")
            return HandoffStatus.FAILED, state

class ResearchAgent(HandoffAgent):
    """Produces research findings."""
    
    def process(self, state: HandoffState) -> HandoffState:
        response = client.messages.create(
            model=self.model,
            max_tokens=600,
            system="You are a research agent. Provide a concise finding with sources. Output JSON: {\"claim\": \"...\", \"sources\": [...], \"confidence\": 0.0-1.0}",
            messages=[{"role": "user", "content": state.original_query}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            state.research_findings = result
        except:
            state.research_findings = {"claim": response.content[0].text, "sources": [], "confidence": 0.5}
        
        return state

class ValidationAgent(HandoffAgent):
    """Validates research findings."""
    
    def validate_input(self, state: HandoffState) -> tuple[bool, str]:
        """Check that research findings are present."""
        if state.research_findings is None:
            return False, "No research findings to validate"
        if not state.research_findings.get("claim"):
            return False, "Missing claim in research findings"
        return True, "Valid"
    
    def process(self, state: HandoffState) -> HandoffState:
        claim = state.research_findings.get("claim", "")
        sources = state.research_findings.get("sources", [])
        
        response = client.messages.create(
            model=self.model,
            max_tokens=400,
            system="You are a validation agent. Assess the claim's validity. Output JSON: {\"valid\": true/false, \"reasoning\": \"...\", \"confidence\": 0.0-1.0}",
            messages=[{"role": "user", "content": f"Claim: {claim}\nSources: {sources}"}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            state.validation_result = result
        except:
            state.validation_result = {"valid": True, "reasoning": "Could not validate", "confidence": 0.5}
        
        return state

class SynthesisAgent(HandoffAgent):
    """Synthesizes validated findings into final response."""
    
    def validate_input(self, state: HandoffState) -> tuple[bool, str]:
        """Check that both research and validation have occurred."""
        if state.research_findings is None:
            return False, "Missing research findings"
        if state.validation_result is None:
            return False, "Missing validation result"
        return True, "Valid"
    
    def process(self, state: HandoffState) -> HandoffState:
        claim = state.research_findings.get("claim")
        valid = state.validation_result.get("valid", False)
        reasoning = state.validation_result.get("reasoning", "")
        
        prompt = f"""Synthesize this into a final answer:
        Claim: {claim}
        Valid: {valid}
        Validation reasoning: {reasoning}
        
        Output JSON: {{"final_answer": "...", "confidence": 0.0-1.0}}"""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=400,
            system="You are a synthesis agent. Produce a clear final answer.",
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            state.synthesis = result
        except:
            state.synthesis = {"final_answer": response.content[0].text, "confidence": 0.5}
        
        return state

class HandoffPipeline:
    """Orchestrates a pipeline of agents with robust handoffs."""
    
    def __init__(self):
        self.research = ResearchAgent("ResearchAgent")
        self.validation = ValidationAgent("ValidationAgent")
        self.synthesis = SynthesisAgent("SynthesisAgent")
    
    def execute(self, query: str) -> HandoffState:
        """Run the full pipeline."""
        print(f"Query: {query}\n")
        
        state = HandoffState(query)
        
        # Step 1: Research
        print("=== Step 1: Research ===")
        state = self.research.process(state)
        print(f"Research result: {state.research_findings}\n")
        
        # Step 2: Validation with handoff
        print("=== Step 2: Validation ===")
        status, state = self.research.handoff_to(self.validation, state)
        if status == HandoffStatus.FAILED:
            return state
        print(f"Validation result: {state.validation_result}\n")
        
        # Step 3: Synthesis with handoff
        print("=== Step 3: Synthesis ===")
        status, state = self.validation.handoff_to(self.synthesis, state)
        if status == HandoffStatus.FAILED:
            return state
        print(f"Synthesis result: {state.synthesis}\n")
        
        return state

# Example
if __name__ == "__main__":
    pipeline = HandoffPipeline()
    final_state = pipeline.execute("What is the capital of France?")
    
    print("=== Final State and Handoff History ===")
    print(json.dumps(final_state.to_dict(), indent=2))

Comparison of Handoff Patterns​

PatternLatencyFlexibilityDebuggabilityBest For
Sequential pipelineLow (linear)Low (fixed order)Excellent (clear flow)Transformative workflows
Conditional routingMedium (branching)High (dynamic paths)Good (trace each branch)Categorization, classification
BidirectionalHigh (iteration)High (feedback loops)Medium (loops hard to debug)Iterative refinement
Swarm handoff via scratchpadLow (async reads)High (any agent can read)Poor (distributed state)Exploration, parallel work

Handoff Best Practices​

  1. Always validate state on receipt: Never assume the sending agent did everything correctly.
  2. Include metadata: Timestamp, sender ID, handoff ID for tracking.
  3. Keep context minimal: Only pass what the next agent needs. Bloated context slows things down.
  4. Document assumptions: If agent B assumes certain fields are present, document it in agent A's output schema.
  5. Implement retry logic: If validation fails, retry with exponential backoff before escalating.
  6. Log handoffs: Every handoff should be logged for auditing and debugging.

Key Takeaways​

  • Handoffs are critical moments where work transitions between agents; they require explicit context transfer and validation.
  • Always validate incoming state before processing; never assume the previous agent succeeded.
  • Use handoff status codes (success, failed, retry) to implement graceful error handling.
  • Sequential pipelines are simple but rigid; conditional routing and bidirectional handoffs add flexibility at the cost of complexity.
  • Log and audit all handoffs for debugging and compliance.

Frequently Asked Questions​

What if context is very large and expensive to transfer?​

Use references instead: store large payloads in a database or object store, and pass only an ID or URI in the handoff. The next agent retrieves the full context if needed. This reduces handoff latency and cost.

How do I handle a handoff timeout?​

If agent B takes too long to respond after a handoff, define a timeout and escalate. Options: retry with a different worker, return a partial answer, or notify the user. Always log timeouts.

Can I pipeline heterogeneous agents (different models, architectures)?​

Yes. As long as each agent accepts the handoff state format and produces output in the agreed format, they can use different underlying architectures. This enables, for example, piping Claude output into a regex validator into a specialized tool.

What if agent B rejects the handoff state (validation fails)?​

Implement a fallback: route back to agent A with a specific error message, or escalate to a supervisor agent to decide next steps. Avoid silently dropping the request.

How do I test handoff patterns?​

Create unit tests for each agent's validate_input and process methods. Integration tests should verify that handoff chains work end-to-end with realistic inputs. Test failure cases: what happens if validation fails? What if an agent times out?

Further Reading​