multi-agent-orchestration

Multi-Agent Orchestration

Coordinate multiple specialized agents for complex tasks.

When Claude Code is the Orchestrator

If Claude Code itself is fanning out (not your user's application code), emit multiple Agent tool calls in a single response message — do not serialize. Opus 4.7 is conservative about parallel delegation and will run agents sequentially unless explicitly instructed otherwise. The code patterns below (asyncio.gather, etc.) describe application-level fan-out; when Claude is the orchestrator, the equivalent is: call Agent N times in one tool-use block, then synthesize after all return.

Canonical example: plugins/engineering-toolkit/skills/brainstorming/references/deep-mode-phases.md ("Launch ALL 8 agents in ONE message").

Fan-Out/Fan-In Pattern (Application Code)

async def multi_agent_analysis(content: str) -> dict:
    """Fan-out to specialists, fan-in to synthesize."""
    agents = [
        ("security", security_agent),
        ("performance", performance_agent),
        ("code_quality", quality_agent),
        ("architecture", architecture_agent),
    ]

    # Fan-out: Run all agents in parallel
    tasks = [agent(content) for _, agent in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter successful results
    findings = [
        {"agent": name, "result": result}
        for (name, _), result in zip(agents, results)
        if not isinstance(result, Exception)
    ]

    # Fan-in: Synthesize findings
    return await synthesize_findings(findings)

Supervisor Pattern

class Supervisor:
    """Central coordinator that routes to specialists."""

    def __init__(self, agents: dict):
        self.agents = agents  # {"security": agent, "performance": agent}
        self.completed = []

    async def run(self, task: str) -> dict:
        """Route task through appropriate agents."""
        # 1. Determine which agents to use
        plan = await self.plan_routing(task)

        # 2. Execute in dependency order
        results = {}
        for agent_name in plan.execution_order:
            if plan.can_parallelize(agent_name):
                # Run parallel batch
                batch = plan.get_parallel_batch(agent_name)
                batch_results = await asyncio.gather(*[
                    self.agents[name](task, context=results)
                    for name in batch
                ])
                results.update(dict(zip(batch, batch_results)))
            else:
                # Run sequential
                results[agent_name] = await self.agents[agent_name](
                    task, context=results
                )

        return results

    async def plan_routing(self, task: str) -> RoutingPlan:
        """Use LLM to determine agent routing."""
        response = await llm.chat([{
            "role": "user",
            "content": f"""Task: {task}

Available agents: {list(self.agents.keys())}

Which agents should handle this task?
What order? Can any run in parallel?"""
        }])
        return parse_routing_plan(response.content)