> ## Documentation Index
> Fetch the complete documentation index at: https://docs.swarms.world/llms.txt
> Use this file to discover all available pages before exploring further.

# Multi-Agent Blocks

> Small building blocks for multi-agent workflows: aggregate concurrent runs, run a single agent safely, and look up agents by name or id

## 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

```bash theme={null}
pip install -U swarms
```

## aggregate()

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

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

<ParamField path="workers" type="List[Callable]" required>
  Agents (or any callables matching the `Agent` interface) to run on the task.
</ParamField>

<ParamField path="task" type="str" required>
  Task passed to every worker.
</ParamField>

<ParamField path="type" type="HistoryOutputType" default="&#x22;all&#x22;">
  Output format passed to `history_output_formatter`.
</ParamField>

<ParamField path="aggregator_model_name" type="str" default="&#x22;anthropic/claude-3-sonnet-20240229&#x22;">
  Model used by the synthesizing aggregator agent.
</ParamField>

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

**Behavior:**

1. All workers run concurrently via `run_agents_concurrently`.
2. Each worker's result is added to a shared `Conversation`, keyed by `worker.agent_name`.
3. A new `Aggregator` agent runs with `AGGREGATOR_SYSTEM_PROMPT` and produces a \~3,000-word synthesis of the worker outputs.
4. The aggregator's response is appended to the conversation.
5. 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)`.

```python theme={null}
def run_agent(
    agent: Agent,
    task: str,
    type: HistoryOutputType = "all",
    *args,
    **kwargs,
)
```

<ParamField path="agent" type="Agent" required>
  Must be an instance of `swarms.structs.agent.Agent`.
</ParamField>

<ParamField path="task" type="str" required>
  Task passed to the agent.
</ParamField>

<ParamField path="type" type="HistoryOutputType" default="&#x22;all&#x22;">
  Accepted but not currently consumed beyond the call — present for API parity with `aggregate`.
</ParamField>

**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).

```python theme={null}
def find_agent_by_name(
    agents: List[Union[Agent, Callable]],
    agent_name: str,
) -> Agent
```

<ParamField path="agents" type="List[Union[Agent, Callable]]" required>
  Non-empty list of agent-like objects.
</ParamField>

<ParamField path="agent_name" type="str" required>
  Name to match. Checked against each agent's `.agent_name` first, then `.name` if set and different.
</ParamField>

**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.

```python theme={null}
def find_agent_by_id(
    agents: List[Union[Agent, Callable]],
    agent_id: str,
) -> Agent
```

<ParamField path="agents" type="List[Union[Agent, Callable]]" required>
  List of agent-like objects to search through.
</ParamField>

<ParamField path="agent_id" type="str" required>
  The `.id` value to match.
</ParamField>

**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.

```python theme={null}
def find_multiple_agents_by_name(
    agents: List[Union[Agent, Callable]],
    agent_names: List[str],
) -> List[Agent]
```

<ParamField path="agents" type="List[Union[Agent, Callable]]" required>
  List of agent-like objects to search through.
</ParamField>

<ParamField path="agent_names" type="List[str]" required>
  Names to match against each agent's `.agent_name`.
</ParamField>

**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`.

```python theme={null}
def return_all_agent_names(
    agents: List[Union[Agent, Callable]],
) -> List[str]
```

<ParamField path="agents" type="List[Union[Agent, Callable]]" required>
  List of agent-like objects.
</ParamField>

**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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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](https://github.com/kyegomez/swarms/blob/master/swarms/structs/ma_blocks.py).