Swarms

Swarms

Swarms is a multi-agent orchestration framework. Everything is built from one primitive — Agent — which multi-agent structures compose. This document is verified against swarms v14.0.0.

Golden rules

1. Import from the top level: from swarms import Agent, never from swarms.structs.agent import Agent. The one common exception is PlannerWorkerSwarm (see below).
2. Every agent needs a unique agent_name — memory files and swarm routing key on it.
3. Default to max_loops=1. Use a specific integer for production. Use "auto" only for genuinely open-ended work.
4. Pass tools=None, not tools=[]. An empty list breaks schema generation.
5. Check examples/ — 586 runnable examples live there. One is probably close to what you need.
6. Never set streaming_on=True and streaming_callback together. Pick one.

Setup

pip install -U swarms

Set the key for whichever provider you use — any LiteLLM model string works:

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GROQ_API_KEY="..."
export WORKSPACE_DIR="agent_workspace"   # where agent state and memory land

Part 1 — The Agent

from swarms import Agent

agent = Agent(
    agent_name="Analyst",
    agent_description="Analyzes market data and produces summaries.",
    system_prompt="You are a precise financial analyst.",
    model_name="gpt-5.4",
    max_loops=1,
)

result = agent.run("Summarize the state of the semiconductor market.")

Agent.__init__ accepts 90+ parameters. These are the ones that matter:

output_type options: "str", "list", "dict", "json", "yaml", "xml", "final", "last", "all", "basemodel", "str-all-except-first", "dict-all-except-first", "dict-final", "list-final".

Running

agent.run(task="...")                          # standard
agent.run(task="...", img="chart.png")         # one image
agent.run(task="...", imgs=["a.png", "b.png"]) # several images
agent.run(task="...", n=3)                     # 3 independent samples
await agent.arun("...")                        # async

Agent.run signature: run(task=None, img=None, imgs=None, correct_answer=None, streaming_callback=None, n=1).

Streaming

# To stdout
agent = Agent(agent_name="Writer", model_name="gpt-5.4", streaming_on=True)
agent.run("Write a haiku about distributed systems.")

# To a callback (do NOT combine with streaming_on)
def on_token(token: str) -> None:
    print(token, end="", flush=True)

agent = Agent(agent_name="Writer", model_name="gpt-5.4", streaming_callback=on_token)
agent.run("Write a haiku.")

# Async streaming
async for token in agent.arun_stream("Explain async/await."):
    print(token, end="", flush=True)

Part 2 — Tools

Any Python function with type hints and a docstring becomes a tool. The framework generates the OpenAI function schema automatically — the docstring is the tool description the model reads, so write it for the model.

from swarms import Agent

def get_stock_price(ticker: str) -> str:
    """Fetch the current stock price for a ticker symbol.

    Args:
        ticker: Stock ticker symbol, e.g. 'AAPL'.

    Returns:
        The current price as a formatted string.
    """
    import yfinance as yf
    return f"{ticker}: ${yf.Ticker(ticker).fast_info['last_price']:.2f}"

agent = Agent(
    agent_name="StockAnalyst",
    model_name="gpt-5.4",
    tools=[get_stock_price],
    max_loops=3,          # needs > 1 so it can act on the tool result
)
agent.run("What are Apple and Microsoft trading at?")

max_loops must exceed 1 for tool use — loop 1 calls the tool, loop 2 uses the result.

Related knobs: tool_call_summary=True (summarize tool output), show_tool_execution_output=True (print raw returns), tool_retry_attempts (retries on tool failure).

MCP servers

agent = Agent(
    agent_name="MCPAgent",
    model_name="gpt-5.4",
    mcp_url="http://localhost:8000/sse",
    # or: mcp_urls=["http://localhost:8000/sse", "http://localhost:8001/sse"]
    max_loops=3,
)

Inspect what a server exposes before wiring it up:

from swarms.tools.mcp_manager import MCPManager

mgr = MCPManager(mcp_url="http://localhost:8000/sse")
print(mgr.list_tool_names())
schemas = mgr.get_tools()          # aget_tools() for the async form

Handoffs

Give an agent a roster it can delegate to. It receives a handoff_task tool automatically.

triage = Agent(
    agent_name="Triage",
    model_name="gpt-5.4",
    handoffs=[billing_agent, technical_agent, refunds_agent],
    max_loops=3,
)
triage.run("My invoice is wrong and the app won't load.")

Part 3 — Autonomous mode (max_loops="auto")

The agent runs plan → execute → reflect until it decides it is finished, with 16 built-in tools available:

agent = Agent(
    agent_name="Researcher",
    model_name="gpt-5.4",
    max_loops="auto",
    tools=[search_web],           # your tools stack on top of the built-ins
    persistent_memory=True,
    context_compression=True,
    context_length=32000,
)
agent.run("Research the top 5 vector databases and write compare.md")

Restrict the built-in set with selected_tools (default "all"):

agent = Agent(
    agent_name="ReadOnly",
    max_loops="auto",
    selected_tools=["create_plan", "think", "read_file", "grep", "complete_task"],
)

Inspect the full list at runtime with agent.get_all_selected_tools().

⚠️ run_bash and delete_file are real. In autonomous mode the agent can modify and delete files and execute shell commands. Scope selected_tools and set WORKSPACE_DIR deliberately.

Part 4 — Memory and conversation

Persistent memory

persistent_memory=True reads {WORKSPACE_DIR}/agents/{agent_name}/MEMORY.md on startup and appends to it each response. It is off by default — set it in every process that should share the memory.

agent = Agent(agent_name="ProjectAssistant", model_name="gpt-5.4", persistent_memory=True)
agent.run("My project is called Helios. Remember that.")

# Later process, same agent_name and the flag set again → it remembers.

Context compression

context_compression=True (default) fires at 90% of context_length, summarizing history in place so long sessions never hit the wall. Leave it on for anything long-running.

Conversation

from swarms import Conversation

conv = Conversation(
    name="my-conversation",     # note: `name`, not `agent_name`
    system_prompt="You are helpful.",
    time_enabled=True,
    token_count=True,
)
conv.add("user", "What is 2+2?")
conv.add("assistant", "4.")

conv.return_history_as_string()
conv.search("2+2")
conv.compact(summary="User asked arithmetic. Answer: 4.")   # archives, then collapses
conv.save_as_json("conv.json")

Part 5 — Multi-agent architectures

Choosing one

SequentialWorkflow

Each agent's output becomes the next agent's context.

from swarms import Agent, SequentialWorkflow

pipeline = SequentialWorkflow(
    agents=[researcher, analyst, writer],
    max_loops=1,
    output_type="dict",
)
pipeline.run("Analyze how rate hikes affect tech stocks.")

Options: team_awareness=True (agents see the roster), multi_agent_collab_prompt=True, drift_detection=True.

ConcurrentWorkflow

All agents run the same task in parallel.

from swarms import Agent, ConcurrentWorkflow

workflow = ConcurrentWorkflow(
    agents=agents,
    max_workers=5,
    show_dashboard=True,
             # or "raise"
)
workflow.run("List 10 use cases for multi-agent AI.")

AgentRearrange — flow DSL

from swarms import Agent, AgentRearrange

pipeline = AgentRearrange(
    agents=[planner, coder, reviewer, tester],
    flow="Planner -> Coder -> Reviewer, Tester",
    max_loops=1,
)
pipeline.run("Build an email validator.")

• A -> B — sequential, B receives A's output
• A, B — concurrent, same input
• A -> B, C -> D — A, then B and C in parallel, then D on their combined output

Every name in flow must match an agent_name in agents, or it fails at run time. There is no human-in-the-loop step — split into separate .run() calls and insert your own input() between them.

GraphWorkflow — DAG

Pass agents directly to add_node/add_edge; there is no need to wrap them in Node objects.

from swarms import Agent, GraphWorkflow

wf = GraphWorkflow(name="research-dag", max_loops=1)

for a in (ingestion, branch_a, branch_b, merger):
    wf.add_node(a)

wf.add_edge(ingestion, branch_a)      # fan out
wf.add_edge(ingestion, branch_b)
wf.add_edge(branch_a, merger)         # fan in
wf.add_edge(branch_b, merger)

wf.set_entry_points(["Ingestion"])
wf.set_end_points(["Merger"])

def on_done(node: str, result) -> None:
    print(f"[{node}] {len(str(result))} chars")

results = wf.run(task="Analyze this dataset two ways and merge.",

add_node also accepts a nested GraphWorkflow. Other options: backend="networkx"|"rustworkx", max_parallel_nodes, checkpoint_dir, streaming_callback.

SwarmRouter — one entry point

Swap architectures without rewriting orchestration.

from swarms import Agent, SwarmRouter

router = SwarmRouter(agents=agents, swarm_type="SequentialWorkflow", max_loops=1)
router.run("Write a post about transformers.")

Valid swarm_type values — exactly these 16:

"AgentRearrange", "MixtureOfAgents", "SequentialWorkflow", "ConcurrentWorkflow", "GroupChat", "MultiAgentRouter", "HierarchicalSwarm", "MajorityVoting", "CouncilAsAJudge", "HeavySwarm", "BatchedGridWorkflow", "LLMCouncil", "DebateWithJudge", "RoundRobin", "PlannerWorkerSwarm", "auto".

"AutoSwarmBuilder" and "SpreadSheetSwarm" are not router types — use those classes directly. With swarm_type="AgentRearrange" you must also pass rearrange_flow.

MixtureOfAgents

Workers answer independently; an aggregator synthesizes. Best with diverse providers.

from swarms import Agent, MixtureOfAgents

moa = MixtureOfAgents(
    agents=[worker_gpt, worker_claude, worker_llama],
    aggregator_agent=aggregator,       # optional; falls back to aggregator_model_name
    layers=3,
    max_loops=1,
)
moa.run("Best practices for securing a Kubernetes cluster?")

HierarchicalSwarm

A director decomposes the task, delegates, and synthesizes results.

from swarms import Agent, HierarchicalSwarm

swarm = HierarchicalSwarm(
    agents=[data_worker, writing_worker, review_worker],
    director=director,            # optional; else built from director_model_name
    max_loops=2,
    planning_enabled=True,
    parallel_execution=True,
    director_feedback_on=True,
)
swarm.run("Produce a competitive analysis of the AI chip market.")

Also: agent_as_judge=True, max_agent_retries, max_reassignment_attempts, interactive=True.

GroupChat

Asynchronous and self-selecting — no rounds, no speaker-selection function. Every agent scores how much it wants to speak (0–1); replies above threshold are broadcast. Ends at max_loops messages or after idle_timeout seconds of silence.

from swarms import Agent, GroupChat

chat = GroupChat(
    agents=[optimist, pessimist, realist],   # at least 2 required
    max_loops=10,
    threshold=0.5,           # raise for a more selective room
    recency_penalty=0.3,     # discourages one agent dominating
    idle_timeout=8.0,
)
chat.run("Should we adopt AI for medical diagnosis?")

auto_equip=True (default) injects the required RESPOND_TOOL into every agent — you do not need to pass it yourself. Set auto_equip=False only if you attach RESPOND_TOOL manually via tools_list_dictionary.

MajorityVoting

Agents answer independently; a consensus agent picks the winner.

from swarms import Agent, MajorityVoting

mv = MajorityVoting(
    agents=voters,
    consensus_agent_model_name="gpt-5.4",
    max_loops=1,
)
mv.run("Python or Rust for a high-performance web server?")

CouncilAsAJudge

Evaluates a response across dimensions. It builds its own council from model names — it does not take an agents list or a judge agent.

from swarms import CouncilAsAJudge

council = CouncilAsAJudge(
    model_name="gpt-5.4",
    aggregation_model_name="gpt-5.4",
    random_model_name=True,
    max_loops=1,
)
council.run("Should we store biometric data on-device only?")

DebateWithJudge

from swarms import Agent, DebateWithJudge

debate = DebateWithJudge(
    pro_agent=pro,
    con_agent=con,
    judge_agent=judge,
    max_loops=3,          # rounds
)
debate.run("Motion: open-source LLMs will surpass closed-source by 2027.")

preset_agents=True generates pro/con/judge for you from model_name. The kwargs are pro_agent/con_agent/judge_agent — not agents=[...] plus judge=.

HeavySwarm

Deep multi-stage analysis. Configured by model names, not by an agents list.

from swarms import HeavySwarm

swarm = HeavySwarm(
    question_agent_model_name="gpt-5.4",
    worker_model_name="gpt-5.4",
    max_loops=1,
    timeout=900,
    show_dashboard=True,
    worker_tools=[search_web],
)
swarm.run("Analyze the implications of AGI on global labour markets.")

PlannerWorkerSwarm

A planner decomposes the task and workers execute; a judge checks completion each cycle. Not exported at the top level:

from swarms.structs.planner_worker_swarm import PlannerWorkerSwarm

swarm = PlannerWorkerSwarm(
    agents=workers,                    # workers only — the planner is built internally
    planner_model_name="gpt-5.4",
    judge_model_name="gpt-5.4",
    max_planner_depth=1,
    max_loops=1,
)
swarm.run("Build a go-to-market strategy for a B2B SaaS product.")

Others

from swarms import (
    MultiAgentRouter,      # routes each task to the best-fit agent
    RoundRobinSwarm,       # fixed rotation
    LLMCouncil,            # members answer, rank peers anonymously, chairman synthesizes
    BatchedGridWorkflow,   # agent i runs task i
    AutoSwarmBuilder,      # generates the agents and architecture from a description
    SpreadSheetSwarm,      # structured tabular processing
    AdvisorSwarm, SelfMoASeq, HybridHierarchicalClusterSwarm,
)

builder = AutoSwarmBuilder(name="MarketResearch", description="...", max_loops=1)
builder.run("Research the EV market and find growth opportunities.")

Part 6 — Execution helpers

from swarms import (
    run_agents_concurrently,
    run_agents_with_different_tasks,
    run_agents_concurrently_async,
    batch_agent_execution,
    run_single_agent,
    aggregate,
)

run_agents_concurrently(agents=agents, task="Summarize today's news.", max_workers=8)
run_agents_with_different_tasks([(agent_a, "task A"), (agent_b, "task B")])  # list of tuples
batch_agent_execution(agents=agents, tasks=tasks, max_workers=10)
aggregate(workers=agents, task="...", aggregator_model_name="gpt-5.4")

Note run_agents_with_different_tasks takes a list of (agent, task) tuples, not a dict.

Scheduling

from swarms import CronJob

job = CronJob(agent=agent, interval="10minutes", job_id="market-check")
job.run(task="Check for unusual market activity.")

interval is "<number><unit>", and the unit must be one of second, seconds, minute, minutes, hour, hours. Abbreviations like "30s" raise CronJobConfigError, as does a zero interval.

Loading agents from files

from swarms import AgentLoader

loader = AgentLoader(concurrent=True)
agents = loader.load_agents_from_markdown("agents/")   # also: _from_yaml, _from_csv
agent = loader.load_agent_from_markdown("agents/researcher.md")

Part 7 — Pitfalls

Production configuration

agent = Agent(
    agent_name="ProductionAgent",
    agent_description="...",
    model_name="gpt-5.4",
    max_loops=3,
    context_length=32000,
    context_compression=True,
    persistent_memory=True,
    autosave=True,
    retry_attempts=3,
    fallback_models=["claude-sonnet-4-6"],
    verbose=False,
)

Debugging

• verbose=True — full internal logging
• show_tool_execution_output=True — raw tool returns
• output_type="all" — the complete conversation instead of just the final message
• agent.get_all_selected_tools() — the autonomous tool roster
• agent.short_memory.return_history_as_string() — dump the conversation

Reference

• Docs: docs.swarms.world · Agent API
• Examples: examples/ — single_agent/, multi_agent/, tools/, guides/
• Source: swarms/structs/ (agents + swarms), swarms/agents/ (loops, judges, routers), swarms/tools/
• Contributing: CONTRIBUTING.md