Mastering Multi‑Agent Collaboration: Handoff Mode and Coordination

Recap of the previous lesson

Implemented multi‑turn dialogue with Session IDs, a task queue for batch execution, and full conversation‑history retention.

Why multi‑agent collaboration?

Complex real‑world tasks often require several specialized agents (search, coding, review). Multi‑agent collaboration isolates context per agent, enabling independent planning for long‑running tasks.

Simple tasks use Tools; complex tasks need Multi‑Agent.

Comparison: Tools vs. Multi‑Agent

Context : Tools share a single context; each agent in a Multi‑Agent system has its own isolated context.

Planning : Tools have no planning capability; each agent can perform independent planning.

Suitable scenarios : Tools for simple, few‑step tasks; Multi‑Agent for tasks requiring planning and specialization.

Complexity : Tools are simple; Multi‑Agent adds moderate complexity.

Common multi‑agent modes

Handoff : An agent transfers a task to another when it cannot finish it.

Router : Dispatches a task to the appropriate agent based on task type.

Supervisor : A single coordinating agent oversees others.

Parallel execution : Multiple agents work simultaneously and their results are merged.

Handoff mode

What is Handoff?

When an agent cannot complete a request, it transfers the whole task to a more specialized agent, preserving the full dialogue context.

# User: "Help me write a crawler"
# → Coding Agent writes code
# → Detects need to fetch webpages
# → Handoff to Execute Agent
# → Execution finishes, Handoff returns result to Coding Agent

Handoff vs. tool calls

Execution entity : Tools are called by an agent; Handoff transfers the entire task.

Context : Tools are stateless; Handoff preserves the full agent context.

Return value : Tools return raw output; Handoff returns the complete agent response.

Use cases : Tools for simple function calls; Handoff for complex tasks needing expertise.

Implementation skeleton

from dataclasses import dataclass
from typing import Optional

class Agent:
    """Base Agent class"""
    def __init__(self, name: str, specialty: str, system_prompt: str):
        self.name = name
        self.specialty = specialty
        self.system_prompt = system_prompt
    def can_handle(self, task: str) -> bool:
        """Simple rule: keyword match"""
        return self.specialty.lower() in task.lower()
    def run(self, task: str, context: dict = None) -> str:
        """Execute the task (calls LLM)"""
        pass

class HandoffAgent:
    """Agent that supports Handoff"""
    def __init__(self):
        self.agents: list[Agent] = []
        self.current_agent: Optional[Agent] = None
    def register(self, agent: Agent):
        self.agents.append(agent)
    def handoff(self, task: str, context: dict = None) -> str:
        target = self._select_agent(task)
        if not target:
            return "Sorry, no Agent can handle this task"
        if self.current_agent and self.current_agent != target:
            print(f"[Handoff] {self.current_agent.name} -> {target.name}")
        self.current_agent = target
        return target.run(task, context)
    def _select_agent(self, task: str) -> Optional[Agent]:
        for agent in self.agents:
            if agent.can_handle(task):
                return agent
        return None

Handoff with full context

@dataclass
class HandoffContext:
    """Handoff context"""
    original_task: str
    history: list[dict]
    shared_data: dict
    from_agent: str
    to_agent: str

def handoff_with_context(from_agent: Agent, to_agent: Agent, context: HandoffContext) -> str:
    handoff_prompt = f"You are taking over from {from_agent.name}.
"
    handoff_prompt += f"Original task: {context.original_task}
"
    handoff_prompt += f"Previous dialogue: {format_history(context.history)}
"
    handoff_prompt += f"Shared data: {format_data(context.shared_data)}
"
    handoff_prompt += "Please continue the task."
    return to_agent.run(handoff_prompt)

Router mode

What is Router?

Routes a task to a specific agent based on the task type.

class Router:
    """Task router"""
    def __init__(self):
        self.routes = {}
    def register(self, pattern: str, agent: Agent):
        self.routes[pattern] = agent
    def route(self, task: str) -> Agent:
        for pattern, agent in self.routes.items():
            if pattern in task.lower():
                return agent
        return None

router = Router()
router.register("write code", coding_agent)
router.register("search", search_agent)
router.register("analyze", analysis_agent)
agent = router.route("Help me search the latest AI news")
result = agent.run("Help me search the latest AI news")

Intelligent routing with an LLM

def intelligent_route(task: str, agents: list[Agent]) -> Agent:
    """Use an LLM to pick the best agent"""
    agent_descriptions = "
".join([f"- {a.name}: {a.specialty}" for a in agents])
    prompt = f"You are a task router. Choose the most suitable agent from the list:
{agent_descriptions}
Task: {task}
Return only the agent name."
    response = llm.chat([{"role": "user", "content": prompt}])
    for agent in agents:
        if agent.name in response:
            return agent
    return agents[0]

Flat‑coordinator architecture

The Coordinator registers agents, decides whether a task needs decomposition, dispatches tasks, performs handoffs, and merges results. All cross‑agent communication passes through the Coordinator, preventing direct agent‑to‑agent calls.

class Coordinator:
    def __init__(self):
        self.agents: dict[str, MiniManus] = {}
        self.cfg = load_config_from_env()
    def register(self, agent: MiniManus):
        self.agents[agent.name] = agent
    def dispatch(self, task: str) -> str:
        if self._need_decompose(task):
            return self._dispatch_with_decompose(task)
        return self._dispatch_direct(task)
    def handoff(self, from_agent: str, to_agent: str, task: str, context: list[dict]) -> str:
        target = self.agents.get(to_agent)
        if not target:
            return f"Error: Agent {to_agent} does not exist"
        ctx = [m for m in context if m.get("role") != "system"]
        return target.run(task, context=ctx)
    # _need_decompose, _decompose_task, _dispatch_direct, _dispatch_with_decompose,
    # _merge_results are LLM‑driven helpers.

MiniManus – individual agent implementation

class MiniManus:
    def __init__(self, spec: AgentSpec, cfg, tools_registry: dict, coordinator: Coordinator):
        self.spec = spec
        self.cfg = cfg
        self.tools_registry = tools_registry
        self.coordinator = coordinator
        self.max_steps = 10
    def run(self, task: str, context: list[dict] = None) -> str:
        messages = self._build_messages(task, context)
        tools = [tool.schema() for tool in self.tools_registry.values()]
        for step in range(1, self.max_steps + 1):
            resp = chat_completions(cfg=self.cfg, messages=messages, tools=tools, tool_choice="auto")
            msg = resp["choices"][0]["message"]
            tool_calls = msg.get("tool_calls", [])
            content = msg.get("content", "").strip()
            if tool_calls:
                for call in tool_calls:
                    name = call["function"]["name"]
                    args = json.loads(call["function"]["arguments"])
                    if name == "request_help":
                        result = self.coordinator.handoff(
                            from_agent=self.name,
                            to_agent=args.get("agent", ""),
                            task=args.get("task", ""),
                            context=messages,
                        )
                        messages.append({"role": "user", "content": f"[Coordinator reply]: {result}"})
                        continue
                    tool = self.tools_registry.get(name)
                    should_stop, output = tool.execute(**args) if tool else (False, "Unknown tool")
                    messages.append({"role": "user", "content": f"[{name}] {output}"})
                    if should_stop:
                        return output
            if content:
                return content
        return "Task timed out"

Usage example

# Create the central coordinator
coordinator = Coordinator()

# Define two specialized agents
coder = MiniManus(
    spec=AgentSpec(name="Coder", specialty="coding", description="..."),
    cfg=cfg,
    tools_registry={"search": SearchTool(), "terminate": TerminateTool()},
    coordinator=coordinator,
)
searcher = MiniManus(
    spec=AgentSpec(name="Searcher", specialty="information search", description="..."),
    cfg=cfg,
    tools_registry={"search": SearchTool(), "terminate": TerminateTool()},
    coordinator=coordinator,
)
# Register them
coordinator.register(coder)
coordinator.register(searcher)

# Dispatch a task – the coordinator will decompose and route automatically
result = coordinator.dispatch("Search the latest AI news and analyse the trend")

Running demonstrations

Simple task (direct dispatch)

$ uv run python 08_multi_agent/main.py --task "Search AI latest news"
[Coordinator] Register Agent: Coder (coding)
[Coordinator] Register Agent: Searcher (information search)
[Coordinator] Received main task: Search AI latest news
[Coordinator] Decision: simple task, direct dispatch
[Coordinator] Dispatch to: Searcher
[Searcher] Processing task: Search AI latest news...
[Searcher] Task completed

Complex task (automatic decomposition)

$ uv run python 08_multi_agent/main.py --task "Search AI latest news and analyse trend" --max-steps 5
[Coordinator] Register Agent: Coder (coding)
[Coordinator] Register Agent: Searcher (information search)
[Coordinator] Register Agent: Analyzer (analysis)
[Coordinator] Received main task: Search AI latest news and analyse trend
[Coordinator] Decision: task needs decomposition
[Coordinator] Decomposed into 2 subtasks
[Coordinator] Executing subtask 1/2: Search AI latest news → Searcher
[Searcher] Processing...
[Searcher] Subtask completed
[Coordinator] Subtask 1 completed
[Coordinator] Executing subtask 2/2: Analyse results → Analyzer
[Analyzer] Processing...
[Analyzer] Subtask completed
[Coordinator] Subtask 2 completed
[Coordinator] Merging results …

Agent‑to‑Agent help request

$ uv run python 08_multi_agent/main.py --task "Write a crawler and execute"
[Coordinator] Received main task: Write a crawler and execute
[Coordinator] Dispatch to: Coder
[Coder] Starting task...
[Coder] Using request_help to hand off to Executor
[Coordinator] Handoff: Coder → Executor
[Executor] Executing crawler …
[Executor] Task completed

Advanced topics

Communication protocols

Shared context : Pass full dialogue history between agents.

Structured data : Pass JSON/dict payloads for clear contracts.

Result reference : Forward only key results to keep agents decoupled.

Error handling with retry

def handoff_with_retry(manager, task: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return manager.handoff(task)
        except Exception as e:
            logger.warning(f"[Retry] {attempt + 1}/{max_retries}: {e}")
            if attempt == max_retries - 1:
                return f"Task failed: {e}"

Monitoring and logging

def log_agent_activity(agent_name: str, activity: str, details: dict):
    logger.info(f"[Agent:{agent_name}] {activity}")
    logger.debug(f"[Agent:{agent_name}] Details: {json.dumps(details)}")

Design principles recap

Every agent is a MiniManus with its own tool set.

The Coordinator is the single hub; all cross‑agent communication passes through it.

Direct communication between sub‑agents is prohibited to avoid complexity.

The coordinator automatically decides whether a task needs decomposition.

Open‑source repository

https://github.com/HUANGLIWEN/mini-manus