name: swarms description: Build agents and multi-agent systems with the Swarms framework — the Agent class, tools, autonomous loops, memory, and the 15+ multi-agent architectures (SequentialWorkflow, ConcurrentWorkflow, GraphWorkflow, HierarchicalSwarm, SwarmRouter, and more). Use whenever writing, reviewing, or debugging code that imports 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,
    on_error="store",          # 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.", on_node_complete=on_done)

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

Overview

Swarms, The Enterprise-Grade Production-Ready Multi-Agent Orchestration Framework

Swarms is the most reliable, scalable, and adaptive multi-agent orchestration framework available today. We provide a comprehensive suite of production-ready, prebuilt multi-agent architectures, including sequential, concurrent, and hierarchical systems. Additionally, Swarms offers backward compatibility with leading agent frameworks and interoperability with protocols such as MCP, x402, skills, and much more.

Install

Using pip

$ pip3 install -U swarms

Using uv (Recommended)

uv is a fast Python package installer and resolver, written in Rust.

$ uv pip install swarms

Using poetry

$ poetry add swarms

From source

# Clone the repository
$ git clone https://github.com/kyegomez/swarms.git
$ cd swarms
$ pip install -r requirements.txt

Environment Configuration

Learn more about the environment configuration here

OPENAI_API_KEY=""
WORKSPACE_DIR="agent_workspace"
ANTHROPIC_API_KEY=""
GROQ_API_KEY=""

Your First Agent

An Agent is the fundamental building block of a swarm—an autonomous entity powered by an LLM + Tools + Memory. Learn more Here

from swarms import Agent

# Initialize a new agent
agent = Agent(
    model_name="gpt-5.4", # Specify the LLM
    max_loops="auto",              # Set the number of interactions
    interactive=True,         # Enable interactive mode for real-time feedback
    temperature=None,
)

# Run the agent with a task
agent.run("What are the key benefits of using a multi-agent system?")

Autonomous Agent with max_loops="auto"

Setting max_loops="auto" lets the agent decide for itself when the task is complete — it keeps reasoning and acting until it reaches a stopping condition, rather than halting after a fixed number of iterations. This is the recommended mode for open-ended, multi-step tasks where the number of steps isn't known in advance.

from swarms import Agent

agent = Agent(
    agent_name="Autonomous-Research-Agent",
    agent_description="An autonomous agent that conducts multi-step research independently.",
    system_prompt=(
        "You are an autonomous research agent. Break down complex tasks into steps, "
        "execute each step thoroughly, and signal completion only when the full task is done."
    ),
    model_name="gpt-5.4",
    max_loops="auto",       # Agent decides when it's done — no fixed iteration cap
    autosave=True,
    verbose=True,
)

# The agent will keep looping — planning, executing, and reflecting — until it
# determines the task is fully complete.
result = agent.run(
    "Research the current state of quantum computing, identify the top three "
    "hardware approaches, and summarize the key challenges each faces."
)
print(result)

When to use max_loops="auto":

• Open-ended research or analysis tasks
• Tasks that require iterative refinement (e.g., write → review → revise)
• Any workflow where the number of steps depends on intermediate results

When to use a fixed max_loops value:

• Latency-sensitive or cost-sensitive production pipelines
• Tasks with a well-defined, bounded number of steps

MCP Integration

The Model Context Protocol (MCP) lets agents easily access external tools and data by pointing to an MCP server URL, which automatically provides tools to the agent as needed. Agents become MCP-enabled by setting mcp_url or mcp_urls, and can use tools from one or many servers with no manual configuration. Free and public MCP servers like DeepWiki work out of the box, offering immediate access to useful agent tools.

from swarms import Agent

agent = Agent(
    agent_name="MCP-Agent",
    model_name="claude-sonnet-5",
    mcp_url="https://mcp.deepwiki.com/mcp",
    max_loops=1,
    temperature=None,
    max_tokens=16_000,
    reasoning_effort=None,
)

print(
    agent.run(
        "Use your tools to explain what the kyegomez/swarms repository does."
    )
)

Serve an Agent as an MCP Server

The reverse direction works too. MCPDeployer turns any agent, or any swarm, into an MCP server that other agents and MCP hosts can call, with an auth layer in front of it. Each target becomes one tool; pass a list or a dict to serve several from one server. See the MCPDeployer examples

from swarms import Agent, MCPDeployer

researcher = Agent(
    agent_name="Researcher",
    agent_description="Answers research questions with a short summary.",
    model_name="gpt-5.4",
    max_loops=1,
)

# Serves http://127.0.0.1:8000/mcp as the tool "researcher".
MCPDeployer(researcher, api_keys=["sk-local-dev"], port=8000).run()

Any other agent can then use it by pointing at the URL with the key:

from swarms import Agent
from swarms.schemas.mcp_schemas import MCPConnection

client = Agent(
    agent_name="Client",
    model_name="gpt-5.4",
    mcp_url=MCPConnection(url="http://127.0.0.1:8000/mcp", api_key="sk-local-dev"),
    max_loops=2,
)
client.run("Use the researcher tool to summarise the state of solid-state batteries.")

Auth can be static API keys, your own auth callable that reads the request headers, or an mcp TokenVerifier with required scopes. A server with no auth configured refuses to start unless you pass allow_anonymous=True. Transports: streamable HTTP (default), SSE, or stdio for desktop MCP hosts.

Your First Swarm: Multi-Agent Collaboration

A Swarm consists of multiple agents working together. This simple example creates a two-agent workflow for researching and writing a blog post. Learn More About SequentialWorkflow

from swarms import Agent, SequentialWorkflow

# Agent 1: The Researcher
researcher = Agent(
    agent_name="Researcher",
    system_prompt="Your job is to research the provided topic and provide a detailed summary.",
    model_name="gpt-5.4",
)

# Agent 2: The Writer
writer = Agent(
    agent_name="Writer",
    system_prompt="Your job is to take the research summary and write a beautiful, engaging blog post about it.",
    model_name="gpt-5.4",
)

# Create a sequential workflow where the researcher's output feeds into the writer's input
workflow =