Swarm

Swarm provides autonomous, collaborative agent orchestration. Agents work together as a team, handing off work dynamically based on the task — there is no fixed execution order declared upfront.

Swarm collaboration

1. Start — The entry agent (first node, or entry_point) receives the initial task.
2. Delegate — Agents hand off sub-tasks to specialists based on role and context.
3. Execute — Each agent runs its turn; results are shared across the swarm.
4. Repeat — Handoffs continue until the task completes or a limit is reached.
5. Return — The framework returns a SwarmResult with per-agent results and the handoff sequence.

Usage

from elsai import Agent
from elsai.agent import AgentConfig
from elsai.multiagent import Swarm

coordinator = Agent(
    system_prompt="You coordinate a team of specialists. "
                  "Break down tasks and delegate to the right expert.",
    config=AgentConfig(name="coordinator"),
)
data_analyst = Agent(
    system_prompt="You analyse numerical data and produce insights.",
    config=AgentConfig(name="data_analyst"),
)
writer = Agent(
    system_prompt="You write clear, engaging reports based on data insights.",
    config=AgentConfig(name="writer"),
)

swarm = Swarm(nodes=[coordinator, data_analyst, writer])
result = swarm("Analyse our Q4 sales data and write an executive summary")
print(result)

Custom entry point:

swarm = Swarm(
    nodes=[coordinator, researcher, coder],
    entry_point=coordinator,
)

Async execution:

import asyncio

async def main():
    result = await swarm.invoke_async("Build a REST API spec for a todo app")
    print(result)

asyncio.run(main())

Hooks:

from elsai.hooks import BeforeNodeCallEvent

def track_agents(event: BeforeNodeCallEvent) -> None:
    print(f"Node {event.node_id} is taking a turn")

swarm = Swarm(nodes=[...], hooks=[track_agents])

Configuration

Parameter	Default	Description
nodes	required	List of Agent instances in the swarm
entry_point	first node	Agent that receives the initial task
max_handoffs	20	Maximum handoffs between agents
max_iterations	20	Maximum node executions within the swarm
execution_timeout	900.0	Total swarm timeout in seconds
node_timeout	300.0	Per-node execution timeout in seconds
id	"default_swarm"	Swarm identifier for session management
hooks	None	Lifecycle hook callbacks
plugins	None	Plugins attached to the swarm
session_manager	None	Session backend for persistence
trace_attributes	None	OpenTelemetry attributes for tracing

SwarmResult

Extends GraphResult with the handoff sequence:

result = swarm("Do a complex task")

print(result.status)                       # "completed" | "failed" | "interrupted"
print(result.results)                      # dict[str, NodeResult] per agent
print(result.accumulated_usage["totalTokens"])
print(result.node_history)                 # handoff sequence

A2A not supported

A2A agents are not yet supported inside Swarm. Use Graph or Agent as Tool instead.