Swarm

from loomflow.architecture import Swarm, Handoff

Swarm is a group of peer agents with no coordinator: whichever agent is active either answers or hands the conversation to another peer with a handoff tool. Use it for exploratory systems where you can’t map out the flow in advance. It behaves like OpenAI’s Swarm, the OpenAI Agents SDK’s handoffs and LangGraph Swarm: a handoff ends the current agent’s turn at once, the next agent sees the shared thread, and a follow-up message goes to whoever answered last.

For the conceptual page see Swarm.

Class signature

class Swarm:
    name: str = "swarm"
 
    def __init__(
        self,
        *,
        agents: dict[str, Agent | Handoff],
        entry_agent: str,
        max_handoffs: int = 8,
        detect_cycles: bool = True,
        pass_full_history: bool = True,
        handoff_tool_name: str = "handoff",
        handoffs: Mapping[str, Sequence[str]] | None = None,
        resume_last_agent: bool = True,
        worker_registry: dict[str, Any] | None = None,
        role_to_worker_id: dict[str, str] | None = None,
    ) -> None: ...

Constructor parameters

agents

Peer agents keyed by name. Plain Agent values get a default Handoff config (untyped, free-form messages); wrap a peer with Handoff(...) to declare a typed handoff (a per-target tool with a structured argument schema). Must be non-empty.

entry_agent

Name of the peer that receives the first user message. Must be a key in agents; raises ValueError otherwise.

max_handoffs

Maximum handoffs in one run. When the cap is reached the swarm emits swarm.max_handoffs, the peer just handed the conversation answers once more with no handoff tools, and the run ends with interrupted=True, interruption_reason="swarm:max_handoffs". Must be >= 0.

detect_cycles

When True (default), two peers bouncing the conversation back and forth three times (the last 6 handoffs alternating A→B, B→A) trip cycle detection: the swarm emits swarm.cycle_detected, the peer just handed the conversation answers once more with no handoff tools, and the run ends with interrupted=True, interruption_reason="swarm:cycle_detected". Handing back once or twice is allowed. Loops through three or more peers are left to max_handoffs.

pass_full_history

When True (default), the receiving agent’s prompt is the whole thread: the request, then for each handoff the previous peer’s tool calls with their results, its text, and a [Handoff: A → B] marker. When False, it gets the request plus only the latest handoff segment. A peer’s Handoff(input_filter=...) replaces either.

handoff_tool_name

Name of the single generic handoff tool used in untyped mode (when no peer declares input_type). In typed mode (any Handoff has input_type=), per-target tools transfer_to_<key> are generated instead, using the peer’s key in the agents dict.

handoffs

Who may hand off to whom, e.g. {"triage": ["billing", "tech"], "billing": ["triage"]}. With None every peer may hand off to every other peer. A peer left out of the map gets no handoff tool and has to answer. Unknown names or a peer listing itself raise ValueError.

resume_last_agent

When True, the next run with the same user_id and session_id starts with the peer that answered last instead of entry_agent. The mapping is kept in process (bounded, least recently used dropped), so after a restart a conversation starts at entry_agent again.

worker_registry / role_to_worker_id

Internal wiring set by Team.swarm when persistent_subagents=True (the default), so each peer keeps one conversation per user across turns. You normally don’t pass these yourself.

Methods

declared_workers

def declared_workers(self) -> dict[str, Agent]:
    return dict(self._agents)

Returns the peer mapping (the Handoff-wrapped agents are unwrapped to the underlying Agent).

run

1. Start with the peer that answered last in this session (resume_last_agent), or entry_agent.
2. Active turn. The active peer runs with handoff tools for the peers it may hand off to. They are ends_turn tools: a successful call ends the peer’s turn without another model call. A call naming a peer it can’t hand off to fails, and the model sees the error.
3. Handoff. Append the peer’s tool calls, text and the handoff marker to the thread and switch to the target.
4. Cycle / cap check. If detect_cycles sees three round trips, or max_handoffs is reached, the new peer answers once with no handoff tools and the run is marked interrupted.
5. Otherwise loop until a peer answers without handing off.

Events: swarm.started (with entry_agent and resumed), then swarm.active and swarm.handoff per turn, and finally swarm.completed, swarm.cycle_detected or swarm.max_handoffs. If a peer’s own run is interrupted (budget, max_turns), the swarm stops with swarm.peer_interrupted and interruption_reason = "swarm:<peer>:<reason>".

Related types

Handoff

@dataclass(frozen=True)
class Handoff:
    agent: Agent
    input_type: type[BaseModel] | None = None
    input_filter: Callable[[list[str], dict[str, Any]], str] | None = None
    description: str | None = None
    tool_name: str | None = None

Per-peer handoff configuration.

Two handoff modes

Untyped mode

When no peer has input_type=, the swarm exposes a single generic handoff(target, message) tool. The model picks the target name and writes a free-form message.

team = Team.swarm(
    agents={"triage": triage, "billing": billing, "tech": tech},
    entry_agent="triage",
    model="claude-opus-4-7",
)

Typed mode

When ANY peer declares input_type=, the swarm switches to per-target tools transfer_to_<key>(...) with the peer’s input schema. You can mix typed and untyped peers: a peer without input_type gets a transfer_to_<key> tool that takes an optional free-form message.

class BillingHandoff(BaseModel):
    customer_id: str
    issue_summary: str
    priority: Literal["low", "medium", "high"]
 
team = Team.swarm(
    agents={
        "triage": triage,
        "billing": Handoff(billing, input_type=BillingHandoff),
        "tech": Handoff(tech, input_type=TechHandoff),
    },
    entry_agent="triage",
    model="claude-opus-4-7",
)

Source

loomflow/architecture/swarm.py