Multi-Agent Blocks

Overview

swarms.structs.ma_blocks is a small set of helper functions for composing multi-agent workflows without instantiating a full swarm class.

Function	What it does
`aggregate`	Run a list of agents concurrently on the same task, then synthesize via an aggregator agent
`run_agent`	Run a single agent with argument validation and error wrapping
`find_agent_by_name`	Look up an agent by `.agent_name` (or `.name`) in a list
`find_agent_by_id`	Look up an agent by `.id` in a list
`find_multiple_agents_by_name`	Look up several agents by `.agent_name` at once
`return_all_agent_names`	Return every agent’s `.agent_name`

aggregate, run_agent, and find_agent_by_name are exported from the top-level swarms package. The rest must be imported from swarms.structs.ma_blocks. Reach for these when you want quick composability — a function call instead of ConcurrentWorkflow(...).run(...).

Installation

pip install -U swarms

aggregate()

Run every worker on the same task concurrently, then hand the combined transcript to an aggregator agent for synthesis.

def aggregate(
    workers: List[Callable],
    task: str = None,
    type: HistoryOutputType = "all",
    aggregator_model_name: str = "anthropic/claude-3-sonnet-20240229",
)

List[Callable]

required

Agents (or any callables matching the Agent interface) to run on the task.

str

required

Task passed to every worker.

HistoryOutputType

default:"\"all\""

Output format passed to history_output_formatter.

str

default:"\"anthropic/claude-3-sonnet-20240229\""

Model used by the synthesizing aggregator agent.

Raises: ValueError if task is None, workers is None, or workers is not a list of callables. Behavior:

All workers run concurrently via run_agents_concurrently.
Each worker’s result is added to a shared Conversation, keyed by worker.agent_name.
A new Aggregator agent runs with AGGREGATOR_SYSTEM_PROMPT and produces a ~3,000-word synthesis of the worker outputs.
The aggregator’s response is appended to the conversation.
The full conversation is returned formatted per type.

run_agent()

Run a single agent on a task with type-checking and error wrapping. Thin convenience over agent.run(task).

def run_agent(
    agent: Agent,
    task: str,
    type: HistoryOutputType = "all",
    *args,
    **kwargs,
)

Agent

required

Must be an instance of swarms.structs.agent.Agent.

str

required

Task passed to the agent.

HistoryOutputType

default:"\"all\""

Accepted but not currently consumed beyond the call — present for API parity with aggregate.

Raises:

Exception	Condition
`ValueError`	`agent` or `task` is `None`
`TypeError`	`agent` is not an `Agent` instance
`RuntimeError`	Any exception raised by `agent.run()` is wrapped and re-raised

Returns: Whatever agent.run(task) returns.

find_agent_by_name()

Look up an agent by name. Builds a name -> agent index the first time it is called for a given agents list — matching against both .agent_name and .name (if .name is set and differs from .agent_name) — and caches that index for subsequent lookups against the same list, turning repeated calls from O(n) into O(1).

def find_agent_by_name(
    agents: List[Union[Agent, Callable]],
    agent_name: str,
) -> Agent

List[Union[Agent, Callable]]

required

Non-empty list of agent-like objects.

str

required

Name to match. Checked against each agent’s .agent_name first, then .name if set and different.

Raises:

Exception	Condition
`ValueError`	`agents` is empty, `agent_name` is empty/whitespace, or no match found
`TypeError`	`agent_name` is not a string

find_agent_by_id()

Linear search for an agent by its .id attribute. Unlike find_agent_by_name, this does not raise on a miss.

def find_agent_by_id(
    agents: List[Union[Agent, Callable]],
    agent_id: str,
) -> Agent

List[Union[Agent, Callable]]

required

List of agent-like objects to search through.

str

required

The .id value to match.

Returns: The matching agent, or None if no agent has that .id. Import: not exported from top-level swarms — use from swarms.structs.ma_blocks import find_agent_by_id.

find_multiple_agents_by_name()

Look up several agents by .agent_name in one call.

def find_multiple_agents_by_name(
    agents: List[Union[Agent, Callable]],
    agent_names: List[str],
) -> List[Agent]

List[Union[Agent, Callable]]

required

List of agent-like objects to search through.

List[str]

required

Names to match against each agent’s .agent_name.

Returns: The subset of agents whose .agent_name is in agent_names. Names with no match are silently dropped — no exception is raised. Import: not exported from top-level swarms — use from swarms.structs.ma_blocks import find_multiple_agents_by_name.

return_all_agent_names()

Return every agent’s .agent_name.

def return_all_agent_names(
    agents: List[Union[Agent, Callable]],
) -> List[str]

List[Union[Agent, Callable]]

required

List of agent-like objects.

Returns: List[str] — agent.agent_name for every agent, in order. Import: not exported from top-level swarms — use from swarms.structs.ma_blocks import return_all_agent_names.

Usage Examples

Aggregate Concurrent Analyses

from swarms import Agent, aggregate

analysts = [
    Agent(agent_name="Bull-Case", model_name="claude-sonnet-4-6", max_loops=1,
          system_prompt="Argue the bull case using data."),
    Agent(agent_name="Bear-Case", model_name="claude-sonnet-4-6", max_loops=1,
          system_prompt="Argue the bear case using data."),
    Agent(agent_name="Macro-Lens", model_name="claude-sonnet-4-6", max_loops=1,
          system_prompt="Frame the question in macro context."),
]

result = aggregate(
    workers=analysts,
    task="Is now a good time to invest in industrial automation stocks?",
)

The returned value is the full conversation — three analyst responses plus the aggregator’s synthesis.

Safe Single-Agent Run

from swarms import Agent, run_agent

agent = Agent(
    agent_name="Drafter",
    model_name="claude-sonnet-4-6",
    max_loops=1,
)

# Raises TypeError if the first argument isn't an Agent
result = run_agent(agent, "Write a release-note bullet for the v12.1 streaming API")

Look Up an Agent by Name

from swarms import Agent, find_agent_by_name

agents = [
    Agent(agent_name="Researcher", model_name="claude-sonnet-4-6"),
    Agent(agent_name="Writer", model_name="claude-sonnet-4-6"),
    Agent(agent_name="Editor", model_name="claude-sonnet-4-6"),
]

# Matches against .agent_name by default
writer = find_agent_by_name(agents, "Writer")

find_agent_by_name matches against .agent_name first, and falls back to .name if an agent has a .name attribute set that differs from its .agent_name.

Other Lookup Helpers

from swarms import Agent
from swarms.structs.ma_blocks import (
    find_agent_by_id,
    find_multiple_agents_by_name,
    return_all_agent_names,
)

agents = [
    Agent(agent_name="Researcher", model_name="claude-sonnet-4-6"),
    Agent(agent_name="Writer", model_name="claude-sonnet-4-6"),
    Agent(agent_name="Editor", model_name="claude-sonnet-4-6"),
]

print(return_all_agent_names(agents))
# ['Researcher', 'Writer', 'Editor']

subset = find_multiple_agents_by_name(agents, ["Writer", "Editor"])

by_id = find_agent_by_id(agents, agents[0].id)

Source Code

View the source on GitHub.

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Multi-Agent Blocks

Overview

Installation

aggregate()

run_agent()

find_agent_by_name()

find_agent_by_id()

find_multiple_agents_by_name()

return_all_agent_names()

Usage Examples

Aggregate Concurrent Analyses

Safe Single-Agent Run

Look Up an Agent by Name

Other Lookup Helpers

Source Code

​Overview

​Installation

​aggregate()

​run_agent()

​find_agent_by_name()

​find_agent_by_id()

​find_multiple_agents_by_name()

​return_all_agent_names()

​Usage Examples

​Aggregate Concurrent Analyses

​Safe Single-Agent Run

​Look Up an Agent by Name

​Other Lookup Helpers

​Source Code

Overview

Installation

aggregate()

run_agent()

find_agent_by_name()

find_agent_by_id()

find_multiple_agents_by_name()

return_all_agent_names()

Usage Examples

Aggregate Concurrent Analyses

Safe Single-Agent Run

Look Up an Agent by Name

Other Lookup Helpers

Source Code