Swarm

Swarm is a group of peer agents with no manager: whichever agent is active can pass the conversation to another with a handoff tool, and whoever answers without handing off gives the final answer. It works the way OpenAI’s Swarm, the OpenAI Agents SDK’s handoffs and LangGraph Swarm do. Use it for exploratory systems where the flow can’t be specified in advance.

How it runs

1. Setup. A set of peer Agents, one of them the entry agent that gets the first message.
2. Active turn. The active agent runs with handoff tools for the peers it may hand off to. It can use its own tools first.
3. Handoff. Calling a handoff tool ends the agent’s turn at once and the target takes over. Any peer can hand back to an earlier one.
4. Answer. The first agent that replies without handing off gives the final answer.

Two handoff styles

from pydantic import BaseModel
 
from loomflow import Agent, tool
from loomflow.architecture import Handoff
from loomflow.team import Team
 
@tool
async def lookup_invoice(invoice_id: str) -> str:
    """Look up an invoice."""
    return "invoice 42: charged twice"
 
triage = Agent("Triage incoming requests.", model="claude-opus-4-7")
billing = Agent("Handle billing.", model="claude-opus-4-7", tools=[lookup_invoice])
tech = Agent("Handle technical issues.", model="claude-opus-4-7")
 
# Style 1: one generic handoff(target, message) tool per agent
team = Team.swarm(
    agents={"triage": triage, "billing": billing, "tech": tech},
    entry_agent="triage",
    model="claude-opus-4-7",
)
 
# Style 2: typed handoffs. Wrap a peer in Handoff with an input_type
# and every agent gets a transfer_to_<name> tool with those arguments.
class BillingHandoff(BaseModel):
    invoice_id: str
    reason: str
 
team = Team.swarm(
    agents={
        "triage": triage,
        "billing": Handoff(billing, input_type=BillingHandoff),
        "tech": tech,
    },
    entry_agent="triage",
    model="claude-opus-4-7",
)

input_type= is a Pydantic model; the handoff tool’s schema comes from it, so the model sends structured arguments instead of free text. If they don’t validate, the call fails and the model can try again.

Who can hand off to whom

By default every agent can hand off to every other agent (never to itself). Pass handoffs= to draw the allowed paths:

team = Team.swarm(
    agents={"triage": triage, "billing": billing, "tech": tech},
    entry_agent="triage",
    model="claude-opus-4-7",
    handoffs={
        "triage": ["billing", "tech"],
        "billing": ["triage"],   # billing can only hand back
    },                           # tech isn't listed, so it must answer
)

What the next agent sees

The agent taking over sees the whole thread so far: the user’s request, the tools each earlier agent ran and what they returned, what it said, and a marker for each handoff:

I was charged twice for invoice 42
[triage called lookup_invoice({"invoice_id": "42"}) → invoice 42: charged twice]
Passing this to billing.
[Handoff: triage → billing] duplicate charge on invoice 42

With pass_full_history=False it gets only the request and the latest handoff (the previous agent’s tool calls, text and marker). A peer’s Handoff(input_filter=...) can rewrite what it sees instead.

Follow-up messages

Within one session_id, the user’s next message goes to the agent that answered last, so a conversation that moved to billing stays with billing. Set resume_last_agent=False to start every message at the entry agent. The last agent is remembered in the process, so after a restart a conversation starts at the entry agent again.

Loops and the handoff cap

Two agents bouncing a request back and forth three times (A→B→A→B→A→B) count as a cycle; handing back once or twice is normal and allowed. Longer loops through three or more agents are caught by max_handoffs (default 8).

When either limit stops the swarm, the agent that was just handed the conversation answers once more with no handoff tools, so the user still gets an answer. The run is marked as cut short:

result = await team.run("I was charged twice", session_id="conv-7")
if result.interrupted:
    print(result.interruption_reason)   # "swarm:cycle_detected" or "swarm:max_handoffs"

The stream also carries swarm.cycle_detected or swarm.max_handoffs events. Pass detect_cycles=False to rely on max_handoffs alone.

When (sparingly) to use Swarm

• Exploratory pipelines where the flow shape will change weekly.
• Research-mode systems where you’re testing what handoff patterns emerge.

For production multi-agent: use Supervisor or Router.