Multi-Agent Execution Utilities

Overview

The multi_agent_exec module provides a comprehensive set of utility functions for running multiple agents using various execution strategies. It includes synchronous and asynchronous execution methods, concurrent batch processing, and utility functions for information retrieval. Every function on this page is exported from the top-level swarms package, so from swarms import run_agents_concurrently works. They also remain importable from swarms.structs.multi_agent_exec and swarms.structs.batch_agent_execution.

These helpers call agent.run() on the agents you pass and do not reset their conversations. Running the same Agent instance twice, or in two threads at once, carries one task’s history into the next. Pass distinct agent instances when tasks must not share context.

Installation

pip install -U swarms

Function Overview

Function	Category	Description
`run_single_agent`	Single Agent	Runs a single agent synchronously
`run_agent_async`	Single Agent	Runs a single agent asynchronously using asyncio
`run_agents_concurrently_async`	Concurrent	Runs multiple agents concurrently using asyncio
`run_agents_concurrently`	Concurrent	Optimized concurrent runner with image support and flexible output formats
`run_agents_concurrently_multiprocess`	Concurrent	Manages agents concurrently in batches with optimized performance
`batched_grid_agent_execution`	Batched & Grid	Runs `agents[i]` on `tasks[i]` concurrently
`batch_agent_execution`	Batched & Grid	Runs `agents[i]` on `tasks[i]` (and optional `imgs[i]`) concurrently (from `swarms.structs.batch_agent_execution`)
`run_agents_with_different_tasks`	Batched & Grid	Runs agents with different tasks concurrently in batches
`get_swarms_info`	Utility	Fetches and formats information about available swarms
`get_agents_info`	Utility	Fetches and formats information about available agents

Single Agent Functions

run_single_agent()

Runs a single agent synchronously.

def run_single_agent(agent: AgentType, task: str, *args, **kwargs) -> Any

Parameters:

agent (AgentType): Agent instance to run
task (str): Task string to execute
*args (Any): Additional positional arguments
**kwargs (Any): Additional keyword arguments

Returns: Agent execution result

from swarms import Agent, run_single_agent

agent = Agent(
    agent_name="Financial-Analyst",
    system_prompt="You are a financial analysis expert",
    model_name="claude-sonnet-4-6",
    max_loops=1
)

result = run_single_agent(agent, "Analyze the current stock market trends")
print(result)

run_agent_async()

Runs a single agent asynchronously using asyncio.

async def run_agent_async(agent: AgentType, task: str) -> Any

Parameters:

agent (AgentType): Agent instance to run
task (str): Task string to execute

Returns: Agent execution result. The blocking agent.run() call runs in a worker thread via asyncio.to_thread, so it does not block the event loop

import asyncio
from swarms import Agent, run_agent_async

async def main():
    agent = Agent(
        agent_name="Researcher",
        system_prompt="You are a research assistant",
        model_name="claude-sonnet-4-6",
        max_loops=1
    )

    result = await run_agent_async(agent, "Research AI advancements in 2024")
    print(result)

asyncio.run(main())

Concurrent Execution Functions

run_agents_concurrently_async()

Runs multiple agents concurrently using asyncio.

async def run_agents_concurrently_async(
    agents: List[AgentType],
    task: str
) -> List[Any]

Parameters:

agents (List[AgentType]): List of Agent instances to run concurrently
task (str): Task string to execute by all agents

Returns: List of outputs from each agent, in input order Raises: The first exception raised by any agent. Results from the other agents are not returned in that case

import asyncio
from swarms import Agent, run_agents_concurrently_async

async def main():
    agents = [
        Agent(
            agent_name=f"Analyst-{i}",
            system_prompt="You are a market analyst",
            model_name="claude-sonnet-4-6",
            max_loops=1
        )
        for i in range(3)
    ]

    task = "Analyze the impact of AI on job markets"
    results = await run_agents_concurrently_async(agents, task)

    for i, result in enumerate(results):
        print(f"Agent {i+1} result: {result}")

asyncio.run(main())

run_agents_concurrently()

Optimized concurrent agent runner using ThreadPoolExecutor with image support and flexible output formats.

def run_agents_concurrently(
    agents: List[AgentType],
    task: str,
    img: Optional[str] = None,
    max_workers: Optional[int] = None,
    return_agent_output_dict: bool = False,
) -> Union[List[Any], Dict[str, Any]]

Parameters:

agents (List[AgentType]): List of Agent instances to run concurrently
task (str): Task string to execute
img (Optional[str]): Optional image data to pass to agent run() if supported. Only passed when not None
max_workers (Optional[int]): Maximum number of threads in the executor. Defaults to 95% of CPU cores
return_agent_output_dict (bool): If True, returns a dict mapping agent names to outputs. Keys use agent_name, then name, then str(agent)

Returns:

If return_agent_output_dict=False: List of outputs from each agent in input order, so results[i] is agents[i]’s (exceptions included if agents fail)
If return_agent_output_dict=True: Dictionary mapping agent names to outputs, preserving agent input order

The executor carries the caller’s OpenTelemetry context into each worker thread, so agent spans nest under the span that called this function.

from swarms import Agent, run_agents_concurrently

# Create multiple agents
agents = [
    Agent(
        agent_name="Tech-Analyst",
        system_prompt="You are a technology analyst",
        model_name="claude-sonnet-4-6",
        max_loops=1
    ),
    Agent(
        agent_name="Finance-Analyst",
        system_prompt="You are a financial analyst",
        model_name="claude-sonnet-4-6",
        max_loops=1
    ),
    Agent(
        agent_name="Market-Strategist",
        system_prompt="You are a market strategist",
        model_name="claude-sonnet-4-6",
        max_loops=1
    )
]

task = "Analyze the future of electric vehicles in 2025"

# Basic concurrent execution
results = run_agents_concurrently(agents, task, max_workers=4)
for i, result in enumerate(results):
    print(f"Agent {i+1} ({agents[i].agent_name}): {result}")

# Return results as dictionary with agent names as keys
results_dict = run_agents_concurrently(
    agents, task, return_agent_output_dict=True
)
for agent_name, result in results_dict.items():
    print(f"{agent_name}: {result}")

run_agents_concurrently_multiprocess()

Runs multiple agents on the same task in batches. Despite the name, it does not start processes: each batch runs through run_agents_concurrently_async on a fresh event loop, and batches run one after another.

def run_agents_concurrently_multiprocess(
    agents: List[Agent],
    task: str,
    batch_size: int = os.cpu_count()
) -> List[Any]

Parameters:

agents (List[Agent]): List of Agent instances to run concurrently
task (str): Task string to execute by all agents
batch_size (int): Number of agents to run in parallel in each batch. Defaults to CPU count

Returns: List of outputs from each agent, in input order

It creates and runs its own event loop, so call it from synchronous code. Inside a running event loop, await run_agents_concurrently_async(...) instead.

import os
from swarms import Agent, run_agents_concurrently_multiprocess

agents = [
    Agent(
        agent_name=f"Research-Agent-{i}",
        system_prompt="You are a research specialist",
        model_name="claude-sonnet-4-6",
        max_loops=1
    )
    for i in range(5)
]

task = "Research the benefits of renewable energy"
batch_size = os.cpu_count()
results = run_agents_concurrently_multiprocess(agents, task, batch_size)

print(f"Completed {len(results)} agent executions")

Batched and Grid Execution

batched_grid_agent_execution()

Runs agents[i] on tasks[i] for every i, concurrently in a thread pool.

def batched_grid_agent_execution(
    agents: List[AgentType],
    tasks: List[str],
    max_workers: int = None,
) -> List[Any]

Parameters:

agents (List[AgentType]): List of agent instances
tasks (List[str]): List of tasks, one for each agent
max_workers (int): Maximum number of threads to use. Defaults to 95% of CPU cores

Returns: List of results in input order, so results[i] is agents[i]’s. A failed agent’s slot holds the exception it raised Raises: ValueError if the number of agents doesn’t match the number of tasks

from swarms import Agent, batched_grid_agent_execution

agents = [
    Agent(
        agent_name="Data-Scientist",
        system_prompt="You are a data science expert",
        model_name="claude-sonnet-4-6",
        max_loops=1
    ),
    Agent(
        agent_name="ML-Engineer",
        system_prompt="You are a machine learning engineer",
        model_name="claude-sonnet-4-6",
        max_loops=1
    ),
    Agent(
        agent_name="AI-Researcher",
        system_prompt="You are an AI researcher",
        model_name="claude-sonnet-4-6",
        max_loops=1
    )
]

tasks = [
    "Analyze machine learning algorithms performance",
    "Design a neural network architecture",
    "Research latest AI breakthroughs"
]

results = batched_grid_agent_execution(agents, tasks, max_workers=3)

for i, result in enumerate(results):
    print(f"Task {i+1}: {tasks[i]}")
    print(f"Result: {result}\n")

batch_agent_execution()

Runs a list of agents on a parallel list of tasks, with an optional parallel list of images. Each agents[i] runs tasks[i]. Unlike batched_grid_agent_execution, it accepts plain callables and returns None for a failed slot instead of the exception.

def batch_agent_execution(
    agents: List[Union[Agent, Callable]],
    tasks: Optional[List[str]] = None,
    imgs: Optional[List[str]] = None,
    max_workers: int = max(1, int(os.cpu_count() * 0.9)),
)

Parameters:

agents (List[Agent | Callable]): Agents to run. An object with a run method is called as agent.run(task, img); any other callable is called as agent(task, img).
tasks (List[str]): One task per agent. Must have the same length as agents. Required in practice: passing None fails.
imgs (List[str], optional): One image input per agent. Must be the same length as agents when provided. When None, every call receives img=None.
max_workers (int): Thread pool size. Defaults to ~90% of CPU cores.

Returns: List of results in agent order, so results[i] is agents[i]’s. A failed call is logged and leaves None in its slot. Raises: BatchAgentExecutionError wrapping any setup failure, including the ValueError for mismatched agents/tasks or agents/imgs lengths.

from swarms import Agent, batch_agent_execution

agents = [
    Agent(agent_name="Summarizer", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Translator", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Classifier", model_name="claude-sonnet-4-6", max_loops=1),
]

tasks = [
    "Summarize the article at https://example.com/post.",
    "Translate 'Bonjour le monde' to English.",
    "Classify this review as positive, negative, or neutral: 'Loved it!'",
]

results = batch_agent_execution(agents=agents, tasks=tasks)
for agent, result in zip(agents, results):
    print(f"{agent.agent_name}: {result}")

Use run_agents_concurrently when every agent runs the same task, and batch_agent_execution, batched_grid_agent_execution or run_agents_with_different_tasks when each agent gets its own task.

run_agents_with_different_tasks()

Runs multiple agents with different tasks concurrently, processing them in batches.

def run_agents_with_different_tasks(
    agent_task_pairs: List[tuple[AgentType, str]],
    batch_size: int = 10,
    max_workers: int = None,
) -> List[Any]

Parameters:

agent_task_pairs (List[tuple[AgentType, str]]): List of (agent, task) tuples
batch_size (int): Number of agents to run in parallel in each batch. Batches run one after another. Default: 10
max_workers (int): Maximum number of threads per batch. Defaults to 95% of CPU cores

Returns: List of outputs from each agent, in the same order as input pairs. A failed agent’s slot holds the exception it raised. An empty input returns []

from swarms import Agent, run_agents_with_different_tasks

# Create agents
agents = [
    Agent(
        agent_name="Content-Writer",
        system_prompt="You are a content writer",
        model_name="claude-sonnet-4-6",
        max_loops=1
    ),
    Agent(
        agent_name="Editor",
        system_prompt="You are an editor",
        model_name="claude-sonnet-4-6",
        max_loops=1
    ),
    Agent(
        agent_name="SEO-Specialist",
        system_prompt="You are an SEO specialist",
        model_name="claude-sonnet-4-6",
        max_loops=1
    )
]

# Create agent-task pairs
agent_task_pairs = [
    (agents[0], "Write a blog post about sustainable living"),
    (agents[1], "Edit and improve this article draft"),
    (agents[2], "Optimize this content for SEO")
]

results = run_agents_with_different_tasks(agent_task_pairs, batch_size=2)

for i, result in enumerate(results):
    agent, task = agent_task_pairs[i]
    print(f"{agent.agent_name} - {task}: {result}")

Utility Functions

get_swarms_info()

Fetches and formats information about all available swarms in the system.

def get_swarms_info(swarms: List[Callable]) -> str

Parameters:

swarms (List[Callable]): List of swarm objects to get information about

Returns: Formatted string with each swarm’s name, description, agent count and swarm_type. Every swarm must have those four attributes. Returns "No swarms currently available in the system." for an empty list

from swarms import get_swarms_info

swarms = [
    # Your swarm objects here
]

info = get_swarms_info(swarms)
print(info)
# Output:
# Available Swarms:
#
# [Swarm 1]
# Name: ResearchSwarm
# Description: A swarm for research tasks
# Length of Agents: 3
# Swarm Type: hierarchical

get_agents_info()

Fetches and formats information about all available agents in the system.

def get_agents_info(
    agents: List[Union[Agent, Callable]],
    team_name: str = None
) -> str

Parameters:

agents (List[Union[Agent, Callable]]): List of agent objects to get information about
team_name (str, optional): Optional team name to display

Returns: Formatted string with each agent’s agent_name, agent_description, role, model_name and max_loops. Returns "No agents currently available in the system." for an empty list

from swarms import Agent, get_agents_info

agents = [
    Agent(
        agent_name="Research-Agent",
        system_prompt="You are a research assistant",
        model_name="claude-sonnet-4-6",
        max_loops=2,
        role="Researcher"
    ),
    Agent(
        agent_name="Analysis-Agent",
        system_prompt="You are a data analyst",
        model_name="claude-sonnet-4-6",
        max_loops=1,
        role="Analyst"
    )
]

info = get_agents_info(agents, team_name="Data Team")
print(info)

Advanced Multi-Agent Workflow Example

from swarms import (
    Agent,
    batched_grid_agent_execution,
    get_agents_info,
    run_agents_concurrently,
    run_agents_with_different_tasks,
)

# Create specialized agents
agents = [
    Agent(
        agent_name="Market-Researcher",
        system_prompt="You are a market research expert specializing in consumer behavior",
        model_name="claude-sonnet-4-6",
        max_loops=1,
        role="Researcher"
    ),
    Agent(
        agent_name="Data-Analyst",
        system_prompt="You are a data analyst expert in statistical analysis",
        model_name="claude-sonnet-4-6",
        max_loops=1,
        role="Analyst"
    ),
    Agent(
        agent_name="Strategy-Consultant",
        system_prompt="You are a strategy consultant specializing in business development",
        model_name="claude-sonnet-4-6",
        max_loops=1,
        role="Consultant"
    ),
    Agent(
        agent_name="Financial-Advisor",
        system_prompt="You are a financial advisor specializing in investment strategies",
        model_name="claude-sonnet-4-6",
        max_loops=1,
        role="Advisor"
    )
]

# Display agent information
print("=== Agent Information ===")
print(get_agents_info(agents, "Business Intelligence Team"))

# Same task for all agents (concurrent execution)
task = "Analyze the impact of remote work trends on commercial real estate market"
results = run_agents_concurrently(agents, task, max_workers=4)

for i, result in enumerate(results):
    print(f"\n{agents[i].agent_name} Analysis:")
    print(f"Result: {result}")

# Dictionary output format
results_dict = run_agents_concurrently(
    agents, task, return_agent_output_dict=True, max_workers=4
)
for agent_name, result in results_dict.items():
    print(f"\n{agent_name} Analysis:")
    print(f"Result: {result}")

# Different tasks for different agents
agent_task_pairs = [
    (agents[0], "Research consumer preferences for electric vehicles"),
    (agents[1], "Analyze sales data for EV market penetration"),
    (agents[2], "Develop marketing strategy for EV adoption"),
    (agents[3], "Assess financial viability of EV charging infrastructure")
]

results = run_agents_with_different_tasks(agent_task_pairs, batch_size=2)

# Grid execution with matched agents and tasks
grid_agents = agents[:3]
grid_tasks = [
    "Forecast market trends for renewable energy",
    "Evaluate risk factors in green technology investments",
    "Compare traditional vs sustainable investment portfolios"
]

grid_results = batched_grid_agent_execution(grid_agents, grid_tasks, max_workers=3)

Error Handling and Best Practices

from swarms import Agent, run_agents_concurrently
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)

# Create agents with error handling
agents = [
    Agent(
        agent_name=f"Agent-{i}",
        system_prompt="You are a helpful assistant",
        model_name="claude-sonnet-4-6",
        max_loops=1
    )
    for i in range(5)
]

task = "Perform a complex analysis task"

try:
    results = run_agents_concurrently(agents, task, max_workers=4)

    # Handle results (some may be exceptions)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Agent {i+1} failed with error: {result}")
        else:
            print(f"Agent {i+1} succeeded: {result}")

except Exception as e:
    print(f"Execution failed: {e}")

Performance Considerations

Technique	Best Use Case / Description
ThreadPoolExecutor	Best for I/O-bound work such as LLM calls, supports image processing and flexible output formats
Batch Processing	Prevents system overload with large numbers of agents, maintains order with grid execution
Resource Monitoring	Adjust worker counts based on system capabilities (defaults to 95% of CPU cores)
Async/Await	Use async functions for better concurrency control and platform optimizations
Image Support	Pass image data to agents that support multimodal processing for enhanced capabilities
Dictionary Output	Use `return_agent_output_dict=True` for structured results with agent name mapping
Error Handling	The thread-pool functions return a failed agent’s exception (or `None` for `batch_agent_execution`) in its slot. `run_single_agent`, `run_agent_async`, `run_agents_concurrently_async` and `run_agents_concurrently_multiprocess` propagate the exception

Best Practices

Always handle exceptions in results, as some agents may fail
Use appropriate max_workers based on system resources
Monitor memory usage for large agent counts
Consider batch processing for very large numbers of agents
Use return_agent_output_dict=True for structured, named results
Pass image data to agents that support multimodal processing

Source Code

View the source code on GitHub

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Multi-Agent Execution Utilities

Overview

Installation

Function Overview

Single Agent Functions

run_single_agent()

run_agent_async()

Concurrent Execution Functions

run_agents_concurrently_async()

run_agents_concurrently()

run_agents_concurrently_multiprocess()

Batched and Grid Execution

batched_grid_agent_execution()

batch_agent_execution()

run_agents_with_different_tasks()

Utility Functions

get_swarms_info()

get_agents_info()

Advanced Multi-Agent Workflow Example

Error Handling and Best Practices

Performance Considerations

Best Practices

Source Code

​Overview

​Installation

​Function Overview

​Single Agent Functions

​run_single_agent()

​run_agent_async()

​Concurrent Execution Functions

​run_agents_concurrently_async()

​run_agents_concurrently()

​run_agents_concurrently_multiprocess()

​Batched and Grid Execution

​batched_grid_agent_execution()

​batch_agent_execution()

​run_agents_with_different_tasks()

​Utility Functions

​get_swarms_info()

​get_agents_info()

​Advanced Multi-Agent Workflow Example

​Error Handling and Best Practices

​Performance Considerations

​Best Practices

​Source Code

Overview

Installation

Function Overview

Single Agent Functions

run_single_agent()

run_agent_async()

Concurrent Execution Functions

run_agents_concurrently_async()

run_agents_concurrently()

run_agents_concurrently_multiprocess()

Batched and Grid Execution

batched_grid_agent_execution()

batch_agent_execution()

run_agents_with_different_tasks()

Utility Functions

get_swarms_info()

get_agents_info()

Advanced Multi-Agent Workflow Example

Error Handling and Best Practices

Performance Considerations

Best Practices

Source Code