Delegation & Handoff

Patterns for transferring work between agents: task decomposition, context transfer, and seamless handoffs.

Delegation vs Handoff

Delegation Patterns

1. Explicit Delegation (Manager-Initiated)

class DelegationManager:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents

    async def delegate(self, task: Task, assignee: str, 
                      context: DelegationContext) -> Result:
        agent = self.agents[assignee]

        # Prepare delegation package
        delegation = DelegationPackage(
            task=task,
            context=context,
            delegator=self.manager_id,
            assignee=assignee,
            deadline=context.deadline,
            priority=context.priority,
            expected_output=context.expected_output,
            constraints=context.constraints
        )

        # Execute
        result = await agent.execute_delegation(delegation)

        # Validate
        if not self.validate_result(delegation, result):
            raise DelegationFailed("Result validation failed")

        return result

2. Capability-Based Delegation

class CapabilityBasedDelegation:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
        self.capability_index = self._build_capability_index()

    def _build_capability_index(self) -> dict[str, list[str]]:
        """Map capability → list of agent IDs"""
        index = defaultdict(list)
        for agent_id, agent in self.agents.items():
            for cap in agent.capabilities:
                index[cap].append(agent_id)
        return index

    def find_best_agent(self, required_capabilities: list[str]) -> str | None:
        candidates = set()
        for cap in required_capabilities:
            agents = set(self.capability_index.get(cap, []))
            if candidates:
                candidates &= agents
            else:
                candidates = agents

        if not candidates:
            return None

        # Select best (least loaded, highest skill)
        return min(candidates, key=lambda a: self.agents[a].current_load)

3. Recursive Delegation

class RecursiveDelegation:
    def __init__(self, agents: dict[str, Agent], max_depth: int = 5):
        self.agents = agents
        self.max_depth = max_depth

    async def delegate_recursive(self, task: Task, agent_id: str, 
                                 depth: int = 0) -> Result:
        if depth >= self.max_depth:
            raise MaxDelegationDepthExceeded()

        agent = self.agents[agent_id]

        # Agent decides: execute or sub-delegate
        decision = await agent.decide_delegation(task)

        if decision.action == "execute":
            return await agent.execute(task)

        elif decision.action == "delegate":
            # Sub-delegate to specialist
            subtask = decision.subtask
            specialist = decision.specialist
            return await self.delegate_recursive(subtask, specialist, depth + 1)

        elif decision.action == "decompose":
            # Break into parallel subtasks
            subtasks = decision.subtasks
            results = await asyncio.gather(*[
                self.delegate_recursive(st.task, st.assignee, depth + 1)
                for st in subtasks
            ])
            return self.merge_results(results)

Handoff Patterns

1. Context-Preserving Handoff

class HandoffPackage:
    def __init__(self, 
                 conversation_history: list[Message],
                 current_state: AgentState,
                 active_tools: list[Tool],
                 active_goal: Goal,
                 pending_actions: list[Action],
                 metadata: dict):
        self.conversation_history = conversation_history
        self.current_state = current_state
        self.active_tools = active_tools
        self.active_goal = active_goal
        self.pending_actions = pending_actions
        self.metadata = metadata
        self.handoff_id = str(uuid4())
        self.timestamp = datetime.utcnow()
        self.from_agent = None
        self.to_agent = None

class HandoffManager:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents

    async def handoff(self, from_agent: str, to_agent: str, 
                      reason: HandoffReason) -> HandoffResult:
        from_agent_obj = self.agents[from_agent]
        to_agent_obj = self.agents[to_agent]

        # 1. Prepare handoff package
        package = await from_agent_obj.prepare_handoff(reason)
        package.from_agent = from_agent
        package.to_agent = to_agent

        # 2. Validate package completeness
        if not self.validate_package(package):
            raise IncompleteHandoffPackage()

        # 3. Execute handoff
        await to_agent_obj.receive_handoff(package)

        # 3. Confirm
        confirmation = await to_agent_obj.confirm_handoff(package.handoff_id)

        return HandoffResult(
            handoff_id=package.handoff_id,
            success=True,
            new_agent=to_agent,
            context_preserved=True
        )

2. Streaming Handoff (Live Transfer)

class StreamingHandoff:
    def __init__(self, from_agent: Agent, to_agent: Agent):
        self.from_agent = from_agent
        self.to_agent = to_agent
        self.stream_queue = asyncio.Queue()

    async def start_streaming_handoff(self):
        # 1. Start streaming state
        asyncio.create_task(self._stream_state())

        # 2. Transfer control
        await self.to_agent.prepare_for_handoff()

        # 3. Stream events
        await self.stream_queue.put(HandoffEvent.START)

        while not self.handoff_complete:
            event = await self.from_agent.get_next_event()
            await self.stream_queue.put(event)

            if event.type == "tool_result" and event.is_final:
                self.handoff_complete = True

        await self.stream_queue.put(HandoffEvent.COMPLETE)

    async def _stream_state(self):
        # Stream memory, context, tools incrementally
        for chunk in self.from_agent.stream_state(chunk_size=1000):
            await self.to_agent.receive_state_chunk(chunk)

3. Human-Initiated Handoff

class HumanHandoffManager:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents

    async def request_human_handoff(self, agent_id: str, 
                                    reason: str, 
                                    urgency: Urgency) -> HandoffResult:
        agent = self.agents[agent_id]

        # 1. Prepare summary for human
        summary = await agent.prepare_human_summary()

        # 2. Notify human (Slack, email, dashboard)
        notification = HandoffNotification(
            agent_id=agent_id,
            reason=reason,
            urgency=urgency,
            summary=summary,
            conversation_link=f"/conversations/{agent.conversation_id}"
        )
        await self.notify_human(notification)

        # 3. Wait for human response
        response = await self.wait_for_human_response(
            timeout=300 if urgency == Urgency.NORMAL else 60
        )

        if response.action == "take_over":
            # Human takes over directly
            return await self.human_takes_over(agent, response)
        elif response.action == "reassign":
            # Reassign to another agent
            return await self.reassign_agent(agent, response.target_agent)
        else:
            # Continue with agent
            return HandoffResult(continued=True)

OpenAI Agents SDK Handoffs

Built-in Handoffs

from agents import Agent, Runner, handoff

# Define specialist agents
billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing inquiries, refunds, subscriptions.",
    tools=[refund_tool, subscription_tool]
)

technical_agent = Agent(
    name="Technical Support",
    instructions="Handle technical issues, bugs, integrations.",
    tools=[diagnose_tool, logs_tool]
)

# Triage agent with handoffs
triage_agent = Agent(
    name="Triage",
    instructions="Route to appropriate specialist.",
    handoffs=[billing_agent, technical_agent]
)

# Automatic handoff based on LLM decision
result = await Runner.run(triage_agent, "I was charged twice!")
# LLM decides: "This is a billing issue" → handoffs to billing_agent

Custom Handoff Configuration

from agents import handoff

# Custom handoff with input transformation
billing_handoff = handoff(
    agent=billing_agent,
    tool_name_override="transfer_to_billing",
    tool_description_override="Transfer to billing for refunds, charges, subscriptions",
    on_handoff=lambda ctx: log_handoff("billing", ctx)
)

# Input filter for handoff
def filter_billing_context(ctx):
    return {
        "issue": ctx.context.get("issue"),
        "account_id": ctx.context.get("account_id"),
        "amount": ctx.context.get("amount")
    }

billing_handoff = handoff(
    agent=billing_agent,
    input_filter=filter_billing_context
)

Handoff Patterns in SDK

# 1. Explicit handoff tool
result = await Runner.run(triage_agent, "Help with my bill")
# Agent calls: transfer_to_billing(issue="double charge", account_id="123")

# 2. Automatic handoff via instructions
# "If billing question, handoff to billing_agent"

# 3. Chained handoffs
# triage → billing → refund_specialist

# 4. Conditional handoff
# "If amount > $1000, handoff to senior_billing"

LangGraph Handoffs

Subgraph Handoffs

from langgraph.graph import StateGraph, START, END
from langgraph.graph import send

# Subgraph as specialist
research_subgraph = StateGraph(ResearchState)
research_subgraph.add_node("researcher", researcher_node)
research_subgraph.add_node("analyst", analyst_node)
research_subgraph.add_edge(START, "researcher")
research_subgraph.add_edge("researcher", "analyst")
research_subgraph.add_edge("analyst", END)
compiled_research = research_subgraph.compile()

# Main graph with handoff
main_graph = StateGraph(State)
main_graph.add_node("orchestrator", orchestrator_node)
main_graph.add_node("research_team", compiled_research)  # Handoff to subgraph
main_graph.add_node("writer", writer_node)

main_graph.add_edge(START, "orchestrator")
main_graph.add_conditional_edges("orchestrator", route_to_team)
main_graph.add_edge("research_team", "writer")
main_graph.add_edge("writer", END)

Explicit Handoff with send()

from langgraph.graph import send

def orchestrator_node(state):
    # Decide which team handles this
    if state["task_type"] == "research":
        # Handoff to research subgraph
        return send("research_team", {"topic": state["topic"], "depth": "deep"})
    elif state["task_type"] == "coding":
        return send("coding_team", {"spec": state["spec"]})
    else:
        return {"next": "writer"}

# Inter-agent communication via send()
def researcher_node(state):
    # Can send to another agent within graph
    if need_fact_check:
        return send("fact_checker", {"claim": state["claim"]})
    return {"findings": findings}

Handoff Best Practices

1. Context Preservation Checklist

• [ ] Full conversation history (or compressed summary)
• [ ] Current agent state (variables, memory, tools)
• [ ] Active goal and sub-goals
• [ ] Pending actions and tool calls
• [ ] Tool configurations and auth state
• [ ] User preferences and context
• [ ] Error history and recovery state

2. Handoff Validation

def validate_handoff(package: HandoffPackage) -> ValidationResult:
    errors = []

    # Required fields
    if not package.conversation_history:
        errors.append("Missing conversation history")
    if not package.active_goal:
        errors.append("Missing active goal")
    if not package.current_state:
        errors.append("Missing agent state")

    # Context continuity
    if package.conversation_history:
        last_msg = package.conversation_history[-1]
        if last_msg.role == "assistant" and not last_msg.tool_calls:
            errors.append("Last message appears incomplete")

    # Tool continuity
    if package.active_tools:
        for tool in package.active_tools:
            if tool.requires_auth and not tool.auth_valid:
                errors.append(f"Tool {tool.name} auth expired")

    return ValidationResult(valid=len(errors) == 0, errors=errors)

3. Graceful Degradation

async def robust_handoff(from_agent: Agent, to_agent: Agent, 
                         max_retries: int = 3) -> HandoffResult:
    for attempt in range(max_retries):
        try:
            return await handoff_manager.handoff(from_agent, to_agent)
        except HandoffError as e:
            if attempt == max_retries - 1:
                raise

            # Degraded handoff: minimal context
            minimal_package = from_agent.prepare_minimal_handoff()
            await to_agent.receive_handoff(minimal_package)
            logger.warning(f"Degraded handoff on attempt {attempt + 1}")

    raise HandoffFailed("All handoff attempts failed")

Handoff Patterns in Frameworks

Handoff Anti-Patterns

Related Notes

• Communication Protocols — Messages for handoff
• Coordination Patterns — Delegation in coordination
• Negotiation Consensus — Handoff as negotiation
• Safety Coordination — Safe handoffs
• OpenAI Agents SDK — SDK handoffs
• LangGraph — LangGraph send() and subgraphs
• AutoGen — AutoGen nested chats
• CrewAI — CrewAI delegation
• OpenAI Agents SDK — SDK handoffs
• Control Flow — Handoff as control flow
• Human in the Loop — Human handoffs
• Reliability — Handoff reliability
• Agent Computer Interface — Desktop handoff

Sources: OpenAI Agents SDK docs, LangGraph send() docs, AutoGen GroupChat, CrewAI processes, FIPA ACL handoff, delegation patterns, handoff patterns in microservices, human-agent interaction research.