SwarmAgent: peer-to-peer agent handoffs

A peer-to-peer multi-agent pattern where agents hand off control to each other directly — no central coordinator.

Import path: tenxgraph.prebuilt.agent

Concept

In a supervisor pattern a single coordinator routes all work. In a swarm, any agent can decide to hand off to any other agent it knows about. This produces a flexible, decentralized flow with no bottleneck at the center.

Full graph — three-member example

flowchart TD
    START([START]) --> TRIAGE

    TRIAGE["TRIAGE\n(LLM)"]
    RESEARCHER["RESEARCHER\n(LLM + tools)"]
    RESEARCHER_TOOL["RESEARCHER_TOOL\n(ToolNode)"]
    WRITER["WRITER\n(LLM)"]
    END_NODE([END])

    TRIAGE -- "transfer_to_researcher" --> RESEARCHER
    TRIAGE -- "transfer_to_writer" --> WRITER
    TRIAGE -- "no handoff" --> END_NODE

    RESEARCHER -- "transfer_to_writer" --> WRITER
    RESEARCHER -- "regular tool call" --> RESEARCHER_TOOL
    RESEARCHER -- "no tool calls" --> END_NODE
    RESEARCHER_TOOL --> RESEARCHER

    WRITER -- "no handoff" --> END_NODE

Per-member routing logic

Each member node gets its own routing function. After every LLM call it inspects state.context[-1].tools_calls and picks a branch in priority order:

flowchart LR
    MEMBER["MEMBER\n(LLM)"]
    TOOL["MEMBER_TOOL\n(ToolNode)"]
    TARGET["TARGET MEMBER"]
    END_NODE([END])

    MEMBER -- "1. handoff tool call\n(transfer_to_X)" --> TARGET
    MEMBER -- "2. regular tool call" --> TOOL
    MEMBER -- "3. no tool calls" --> END_NODE
    TOOL --> MEMBER

Python

for tc in last.tools_calls:
    is_handoff, target = is_handoff_tool(tc["name"])   # "transfer_to_X" → target = "x"
    if is_handoff and target.upper() in allowed_set:
        return target.upper()                           # route to that member

if tool_node_name is not None:
    return tool_node_name                              # run regular tools

return END

Handoff tools are never executed

SwarmAgent auto-generates transfer_to_<name> functions and injects them into each member’s ToolNode. When the LLM calls one, the routing function intercepts it and navigates the graph — the tool body never runs. No spurious tool role messages appear in the conversation history.

Mini ReAct loop per member

A member with regular tools gets a dedicated <NAME>_TOOL node and a <NAME>_TOOL → NAME edge. This gives each member its own tool loop before it decides to hand off or stop.

`can_handoff_to` semantics

Value	Behaviour
`None`	Can hand off to all other members
`["A", "B"]`	Can hand off only to A and B
`[]`	Terminal — no handoffs; always routes to END

Per-member independence

Each member is a fully configured Agent instance. Members can use different models, tools, memory, skills, retry config, or multimodal settings. SwarmAgent only wires the graph and injects handoff tools; it does not constrain per-member configuration.

`SwarmMemberConfig` fields

Field	Type	Default	Description
`agent`	`BaseAgent`	required	Pre-built agent instance — do not add handoff tools manually
`can_handoff_to`	`list[str] \| None`	`None`	Allowed targets; `None` = all other members
`description`	`str`	`""`	Injected into other members’ handoff tool docstrings so the LLM knows when to route here

`SwarmAgent` Constructor Parameters

Parameter	Type	Description
`members`	`dict[str, SwarmMemberConfig]`	Mapping of node names to member configs (UPPER-CASE recommended)
`entry`	`str`	Name of the member that receives the first message
`state`	`AgentState \| None`	Optional custom state subclass
`context_manager`	`BaseContextManager \| None`	Optional custom context-trimming manager
`publisher`	`BasePublisher \| None`	Optional event publisher for streaming
`id_generator`	`BaseIDGenerator`	ID generation strategy
`container`	`InjectQ \| None`	Dependency injection container

`compile()` Parameters

Parameter	Type	Default	Description
`checkpointer`	`BaseCheckpointer`	`None`	Persist and restore conversation state
`store`	`BaseStore`	`None`	Long-term cross-thread storage
`interrupt_before`	`list[str]`	`None`	Pause before the named nodes
`interrupt_after`	`list[str]`	`None`	Pause after the named nodes
`callback_manager`	`CallbackManager`	default	Lifecycle hooks
`media_store`	`BaseMediaStore`	`None`	Binary/media file storage
`shutdown_timeout`	`float`	`30.0`	Seconds to wait for clean shutdown

Full Code

Three-member research swarm

Python

import asyncio
from dotenv import load_dotenv
from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.prebuilt.agent import SwarmAgent
from tenxgraph.prebuilt.agent.swarm import SwarmMemberConfig
from tenxgraph.prebuilt.tools import fetch_url, google_web_search
from tenxgraph.core.state import Message

load_dotenv()

def draft_report(topic: str, facts: str) -> str:
    """Draft a structured report from gathered facts."""
    return f"# Report: {topic}\n\n{facts}"

triage_agent = Agent(
    model="gpt-4o-mini",
    provider="openai",
    system_prompt=[{
        "role": "system",
        "content": (
            "You are a triage agent. Decide whether the task needs research "
            "or can go directly to the writer. Route accordingly."
        ),
    }],
)

researcher_agent = Agent(
    model="gpt-4o",
    provider="openai",
    tool_node=ToolNode([fetch_url, google_web_search]),
    system_prompt=[{
        "role": "system",
        "content": "You are a research specialist. Gather facts and hand off to the writer.",
    }],
)

writer_agent = Agent(
    model="gpt-4o-mini",
    provider="openai",
    tool_node=ToolNode([draft_report]),
    system_prompt=[{
        "role": "system",
        "content": "You are a writer. Produce the final document from the gathered information.",
    }],
)

swarm = SwarmAgent(
    members={
        "TRIAGE": SwarmMemberConfig(
            agent=triage_agent,
            can_handoff_to=["RESEARCHER", "WRITER"],
            description="Triages the request and routes to the right specialist.",
        ),
        "RESEARCHER": SwarmMemberConfig(
            agent=researcher_agent,
            can_handoff_to=["WRITER"],
            description="Gathers facts from the web. Use for research tasks.",
        ),
        "WRITER": SwarmMemberConfig(
            agent=writer_agent,
            can_handoff_to=[],  # terminal — no handoffs out
            description="Writes the final document.",
        ),
    },
    entry="TRIAGE",
)

app = swarm.compile()

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message(
            "Write a brief report on quantum computing progress in 2024."
        )]},
        config={"thread_id": "swarm-1"},
    )
    print(result["context"][-1].text())

asyncio.run(main())

Two-member swarm (no triage)

When every member can hand off to every other, omit can_handoff_to (defaults to None = all others):

Python

import asyncio
from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.prebuilt.agent import SwarmAgent
from tenxgraph.prebuilt.agent.swarm import SwarmMemberConfig
from tenxgraph.prebuilt.tools import google_web_search, safe_calculator
from tenxgraph.core.state import Message

researcher = Agent(
    model="gpt-4o-mini",
    provider="openai",
    tool_node=ToolNode([google_web_search]),
    system_prompt=[{"role": "system", "content": "Research topics and hand off to analyst when done."}],
)

analyst = Agent(
    model="gpt-4o-mini",
    provider="openai",
    tool_node=ToolNode([safe_calculator]),
    system_prompt=[{"role": "system", "content": "Analyse data and produce a final answer."}],
)

swarm = SwarmAgent(
    members={
        "RESEARCHER": SwarmMemberConfig(
            agent=researcher,
            description="Searches the web for facts.",
        ),
        "ANALYST": SwarmMemberConfig(
            agent=analyst,
            description="Runs calculations and produces the final answer.",
        ),
    },
    entry="RESEARCHER",
)

app = swarm.compile()

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message("What is the GDP of Germany in USD? Convert at today's rate.")]},
        config={"thread_id": "two-member-1"},
    )
    print(result["context"][-1].text())

asyncio.run(main())

With a checkpointer (persistent conversations)

Python

import asyncio
from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.prebuilt.agent import SwarmAgent
from tenxgraph.prebuilt.agent.swarm import SwarmMemberConfig
from tenxgraph.storage.checkpointer import PgCheckpointer
from tenxgraph.prebuilt.tools import google_web_search
from tenxgraph.core.state import Message

triage = Agent(model="gpt-4o-mini", provider="openai",
               system_prompt=[{"role": "system", "content": "Route requests."}])
researcher = Agent(model="gpt-4o-mini", provider="openai",
                   tool_node=ToolNode([google_web_search]),
                   system_prompt=[{"role": "system", "content": "Research and answer."}])

swarm = SwarmAgent(
    members={
        "TRIAGE": SwarmMemberConfig(triage, can_handoff_to=["RESEARCHER"],
                                    description="Routes requests."),
        "RESEARCHER": SwarmMemberConfig(researcher, description="Researches the topic."),
    },
    entry="TRIAGE",
)

checkpointer = PgCheckpointer(postgres_dsn="postgresql://user:pass@localhost/db")
app = swarm.compile(checkpointer=checkpointer)

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message("Who won the 2024 Nobel Prize in Physics?")]},
        config={"thread_id": "user-99-session-1"},
    )
    print(result["context"][-1].text())

asyncio.run(main())

Google Gemini members

Each member can use a different provider independently:

Python

from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.prebuilt.agent import SwarmAgent
from tenxgraph.prebuilt.agent.swarm import SwarmMemberConfig
from tenxgraph.prebuilt.tools import google_web_search

researcher = Agent(
    model="google/gemini-2.5-flash",
    provider="google",
    tool_node=ToolNode([google_web_search]),
    system_prompt=[{"role": "system", "content": "Research and hand off to writer."}],
    trim_context=True,
)

writer = Agent(
    model="gpt-4o-mini",
    provider="openai",
    system_prompt=[{"role": "system", "content": "Write the final answer."}],
)

swarm = SwarmAgent(
    members={
        "RESEARCHER": SwarmMemberConfig(researcher, can_handoff_to=["WRITER"],
                                        description="Researches the topic."),
        "WRITER": SwarmMemberConfig(writer, can_handoff_to=[],
                                    description="Writes the final document."),
    },
    entry="RESEARCHER",
)

app = swarm.compile()

Running with `agentflow play`

graph.py

Python

from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.prebuilt.agent import SwarmAgent
from tenxgraph.prebuilt.agent.swarm import SwarmMemberConfig
from tenxgraph.prebuilt.tools import google_web_search

triage = Agent(
    model="gpt-4o-mini",
    provider="openai",
    system_prompt=[{"role": "system", "content": "Route requests to researcher or writer."}],
)
researcher = Agent(
    model="gpt-4o-mini",
    provider="openai",
    tool_node=ToolNode([google_web_search]),
    system_prompt=[{"role": "system", "content": "Research the topic and hand off to writer."}],
)
writer = Agent(
    model="gpt-4o-mini",
    provider="openai",
    system_prompt=[{"role": "system", "content": "Write the final answer."}],
)

swarm = SwarmAgent(
    members={
        "TRIAGE": SwarmMemberConfig(triage, can_handoff_to=["RESEARCHER", "WRITER"],
                                    description="Routes the task."),
        "RESEARCHER": SwarmMemberConfig(researcher, can_handoff_to=["WRITER"],
                                        description="Researches the topic."),
        "WRITER": SwarmMemberConfig(writer, can_handoff_to=[],
                                    description="Writes the final answer."),
    },
    entry="TRIAGE",
)

app = swarm.compile()

10xgraph.json

JSON

{
  "agent": "graph:app",
  "env": ".env",
  "auth": null,
  "checkpointer": null,
  "injectq": null,
  "store": null,
  "redis": null,
  "thread_name_generator": null
}

Terminal

agentflow play

Concept

Full graph — three-member example

Per-member routing logic

Handoff tools are never executed

Mini ReAct loop per member

can_handoff_to semantics

Per-member independence

SwarmMemberConfig fields

SwarmAgent Constructor Parameters

compile() Parameters

Full Code

Three-member research swarm

Two-member swarm (no triage)

With a checkpointer (persistent conversations)

Google Gemini members

Running with agentflow play

`can_handoff_to` semantics

`SwarmMemberConfig` fields

`SwarmAgent` Constructor Parameters

`compile()` Parameters

Running with `agentflow play`