Create Your First Multi-Agent Swarm

This guide explains how to build a small multi-agent swarm with one handoff path and required variables. It does not cover advanced FastAPI transport or evaluation design.

This guide is for developers who already understand the single-agent flow and want to add routing. It assumes that you are comfortable with SwarmDefinition, SwarmNode, and provider-backed runtime execution.

After reading this guide, you should be able to:

• define a second agent node
• connect nodes with a SwarmEdge
• require variables before a handoff completes
• run a simple multi-agent routing flow

The first step from a single agent to a multi-agent swarm is adding a second node and connecting the two with a handoff edge.

In SwarmForge, handoffs are graph edges. Each edge can also declare required_variables, which the runtime tries to resolve before the transfer completes.

This page shows the shortest path in three parts:

• create a two-agent swarm
• add handoffs and required variables
• run a full example

Before running the full example, copy .env.example to .env, set MODEL_PROVIDER, set LLM_MODEL, and add the matching API key. For the routing example on this page, use a model that supports tool calling so it can invoke transfer_to_agent.

Quick Start Tabs

Create Swarm

from swarmforge.swarm import SwarmDefinition, SwarmEdge, SwarmNode, SwarmSession


swarm = SwarmDefinition(
    id="support",
    name="Support Swarm",
    nodes=[
        SwarmNode(
            id="triage",
            node_key="triage",
            name="Triage",
            system_prompt="You triage requests and transfer them when the right specialist is clear.",
            is_entry_node=True,
        ),
        SwarmNode(
            id="billing",
            node_key="billing",
            name="Billing",
            system_prompt="You handle billing problems after triage transfers the user to you.",
        ),
    ],
    edges=[],
)

session = SwarmSession(id="session-1", swarm=swarm)

Add Handoffs

from swarmforge.swarm import SwarmEdge


swarm.edges.append(
    SwarmEdge(
        id="triage->billing",
        source_node_id="triage",
        target_node_id="billing",
        handoff_description="Transfer only after confirming the request is billing-related.",
        required_variables=["account_id"],
    )
)

Example Usage

import asyncio
import json

from swarmforge.env import require_env_vars
from swarmforge.evaluation.provider import ModelConfig
from swarmforge.swarm import (
    InMemorySessionStore,
    SwarmDefinition,
    SwarmEdge,
    SwarmNode,
    SwarmSession,
    build_turn_runner,
    process_swarm_stream,
)


swarm = SwarmDefinition(
    id="support",
    name="Support Swarm",
    nodes=[
        SwarmNode(
            id="triage",
            node_key="triage",
            name="Triage",
            system_prompt=(
                "You triage support requests. "
                "Immediately call transfer_to_agent for billing issues once the route is clear."
            ),
            is_entry_node=True,
        ),
        SwarmNode(
            id="billing",
            node_key="billing",
            name="Billing",
            system_prompt="You handle billing issues once triage hands the user to you.",
        ),
    ],
    edges=[
        SwarmEdge(
            id="triage->billing",
            source_node_id="triage",
            target_node_id="billing",
            handoff_description="Transfer only after confirming the request is billing-related.",
            required_variables=["account_id"],
        )
    ],
)


async def extract_required_variables(user_input="", **_kwargs):
    if "ACME-991" in user_input:
        return {"account_id": "ACME-991"}
    return {}


async def main():
    require_env_vars("MODEL_PROVIDER", "LLM_MODEL")
    session = SwarmSession(id="session-1", swarm=swarm)
    store = InMemorySessionStore()
    turn_runner = build_turn_runner(ModelConfig())
    async for event in process_swarm_stream(
        session,
        "I need help with a charge on account ACME-991.",
        store=store,
        turn_runner=turn_runner,
        extract_required_variables=extract_required_variables,
    ):
        print(json.dumps(event, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

What Each Part Does

Create Swarm

A multi-agent swarm is still just a SwarmDefinition, but now it has multiple SwarmNode entries. One node must be the entry node, and that node is where each new session starts.

For a first multi-agent setup, keep the graph small:

• one entry node such as Triage
• one specialist such as Billing
• one edge between them

Add Handoffs

Handoffs are defined with SwarmEdge.

The important fields are:

• source_node_id
• target_node_id
• handoff_description
• required_variables

At runtime, SwarmForge injects a transfer_to_agent tool for valid outgoing routes. Your model layer can call that tool with the target node key, and the runtime will decide whether the handoff is allowed.

If required_variables are listed on the edge, the runtime tries to resolve them before completing the handoff. If it cannot resolve them, the turn emits handoff_blocked instead of handoff. SwarmForge also injects a structured tool result for that blocked transfer_to_agent call, including a concrete rejection_reason, so the next model turn can explain the block in user-facing language instead of falling back to a generic reply.

Required Variables

extract_required_variables(...) is the hook that fills edge requirements such as account_id, invoice_id, or region from the conversation.

When the extraction succeeds, the handoff event includes both:

• requiredVariables
• resolvedVariables

That makes the transfer explicit and inspectable in your event stream.

Because this page now uses a real provider-backed turn runner, the exact handoff timing and final wording vary by model. A tool-capable model should still emit a transfer_to_agent call for a clearly billing-related request.

Expected Event Flow

For the example above, the runtime emits:

open -> tool_use -> handoff -> chunk -> done

If the required variables cannot be resolved, the flow becomes:

open -> tool_use -> handoff_blocked -> chunk -> done

The visible event flow stays the same, but the follow-up chunk now comes from a second model turn that has the blocked transfer's structured tool result in its conversation history.

When To Use This Pattern

Use this page's pattern when you want:

• one triage agent and one specialist
• explicit routing behavior you can inspect in events
• handoffs that depend on conversation-derived variables before transfer

If you only need one node and no routing yet, use Create Your First Agent. If you want deeper runtime details, continue with Orchestration.