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` in a list
`find_agent_by_id`	Look up an agent by `.id` in a list
`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 their answers to an aggregator agent for synthesis.

def aggregate(
    workers: List[Agent],
    task: str = None,
    type: HistoryOutputType = "all",
    aggregator_model_name: str = "claude-sonnet-5",
)

List[Agent]

required

Agents to run on the task. Each worker’s agent_name labels its turn in the conversation, and its answer is read from its own conversation history.

str

required

Task passed to every worker.

HistoryOutputType

default:"\"all\""

Output format passed to history_output_formatter.

str

default:"\"claude-sonnet-5\""

Model used by the synthesizing aggregator agent. The aggregator sets no max_tokens, temperature or top_p, so it uses the model’s own output limit and the provider’s sampling defaults.

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 final answer, not its full transcript, is added to a new Conversation as one turn, with the worker’s agent_name as the role ("Worker" when the name is empty).
A new Aggregator agent runs with AGGREGATOR_SYSTEM_PROMPT and output_type="final". It receives the worker turns as separate chat messages, each labelled with the worker’s name, plus an instruction to write a ~3,000-word synthesis.
The aggregator’s response is appended to the conversation.
The conversation, one turn per worker plus the aggregator’s turn, 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. Performs a plain linear scan over agents, matching each agent’s .agent_name against agent_name and returning the first match — O(n) per call, with no caching and no fallback to .name.

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 against each agent’s .agent_name.

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.

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?",
    type="dict",
    aggregator_model_name="claude-sonnet-4-6",
)

for message in result:
    print(message["role"], len(str(message["content"])))

The returned value is the full conversation — three analyst responses plus the aggregator’s synthesis. With type="dict" it is a list of message dicts with role and content keys.

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 only against .agent_name — there is no fallback to .name.

Other Lookup Helpers

from swarms import Agent
from swarms.structs.ma_blocks import (
    find_agent_by_id,
    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']

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()

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()

​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()

return_all_agent_names()

Usage Examples

Aggregate Concurrent Analyses

Safe Single-Agent Run

Look Up an Agent by Name

Other Lookup Helpers

Source Code