SubagentRegistry

Overview

SubagentRegistry runs agent tasks in the background on a ThreadPoolExecutor, tracking each as a SubagentTask with status, retries, depth, and parent linkage. Use it when one agent needs to fan out work to other agents — recursively if needed — and gather results later. The module exports three symbols:

Export	Kind	Purpose
`SubagentRegistry`	class	Spawns and tracks tasks, gathers results
`SubagentTask`	dataclass	Per-task record with status, result, retry info
`TaskStatus`	`str` enum	`PENDING`, `RUNNING`, `COMPLETED`, `FAILED`, `CANCELLED`

Installation

pip install -U swarms

TaskStatus

from swarms import TaskStatus

TaskStatus.PENDING     # created, not yet submitted
TaskStatus.RUNNING     # submitted to the pool, queued or executing
TaskStatus.COMPLETED   # finished successfully
TaskStatus.FAILED      # exhausted retries
TaskStatus.CANCELLED   # cancelled before completion

Backed by (str, Enum), so the values compare equal to plain strings (TaskStatus.PENDING == "pending"). spawn() sets a task to RUNNING as soon as it submits it, so a task waiting for a free worker thread also reports RUNNING.

SubagentTask

Dataclass describing a single in-flight or completed task. Populated by SubagentRegistry.spawn().

str

Unique task ID, e.g. task-a1b2c3d4.

Any

The agent instance assigned to this task.

str

The prompt/task handed to agent.run(...).

TaskStatus

default:"TaskStatus.PENDING"

Current execution status.

Any

default:"None"

Return value from the agent — set when status == COMPLETED.

Exception | None

default:"None"

Last exception raised — set when status == FAILED.

concurrent.futures.Future | None

default:"None"

Underlying Future returned by the thread pool.

str | None

default:"None"

ID of the parent task that spawned this one, if any.

int

default:"0"

Recursion depth — incremented when an agent spawns another subagent.

int

default:"0"

Number of retries used so far.

int

default:"0"

Retry budget for this task.

List[Type[Exception]] | None

default:"None"

Whitelist of exception classes that trigger retries. spawn() stores [] when you pass None; an empty list retries on any exception.

float

Unix timestamp when the task was spawned.

float | None

default:"None"

Unix timestamp when the task entered a terminal state.

SubagentRegistry

Constructor

from swarms import SubagentRegistry

registry = SubagentRegistry(max_depth=3, max_workers=None)

int

default:"3"

Maximum recursion depth. spawn() raises ValueError if depth > max_depth.

int | None

default:"None"

Thread-pool size. None defers to ThreadPoolExecutor’s default.

Methods

spawn()

Submit an agent task to the pool. Returns the new task ID synchronously; the agent runs in the background.

def spawn(
    self,
    agent: Any,
    task: str,
    parent_id: Optional[str] = None,
    depth: int = 0,
    max_retries: int = 0,
    retry_on: Optional[List[Type[Exception]]] = None,
    fail_fast: bool = True,
) -> str

Any

required

Agent instance with a .run(task) method.

str

required

The prompt to run.

str | None

default:"None"

Set when this task was spawned by another task. Used for tracking trees.

int

default:"0"

Recursion depth. Spawning from inside another task increments this.

int

default:"0"

Retry budget on failure.

List[Type[Exception]] | None

default:"None"

Only retry on these exception types. None retries on any exception.

bool

default:"True"

When True, the underlying thread re-raises on final failure (the exception surfaces when you call future.result() or gather()). When False, the failure is captured on the SubagentTask.error and the thread returns None. Either way the task’s status becomes FAILED.

Raises: ValueError if depth > max_depth.

get_task()

Look up a SubagentTask by ID.

def get_task(self, task_id: str) -> SubagentTask

Raises: KeyError if the ID is not in the registry.

get_results()

Return a Dict[task_id, Any] for every completed or failed task. Failed tasks map to their exception object. Running and cancelled tasks are left out. It does not wait, so call gather() first if you need every result.

def get_results(self) -> Dict[str, Any]

cancel()

Attempt to cancel a task that is still queued. Returns True and sets the status to CANCELLED if the underlying Future accepted the cancellation. A task that has started running cannot be cancelled.

def cancel(self, task_id: str) -> bool

Raises: KeyError if the ID is not in the registry.

gather()

Block until tasks complete and return a list of results.

def gather(
    self,
    strategy: str = "wait_all",
    timeout: Optional[float] = None,
) -> List[Any]

"wait_all" | "wait_first"

default:"\"wait_all\""

Wait policy. "wait_all" returns once every pending task settles; "wait_first" returns as soon as one does.

float | None

default:"None"

Max seconds to wait. None blocks indefinitely.

Returns: Mixed list. Tasks that had already finished when you called gather() come first, in spawn order: a completed task contributes its result and a failed task its exception. Tasks that finish during the wait follow in no guaranteed order, contributing their return value or the exception their future raised (a cancelled task contributes a CancelledError). Tasks still running when timeout expires, or when "wait_first" returns, are left out.

gather() results are not keyed by task. When you need to know which result belongs to which task, keep the IDs from spawn() and read get_task(task_id).result (or get_results()) after gather() returns. A task spawned with fail_fast=False that fails during the wait contributes None here; its exception is on get_task(task_id).error.

shutdown()

Shut down the thread pool without waiting. Tasks already spawned still run in the background; calling spawn() afterwards raises RuntimeError.

def shutdown(self) -> None

tasks

Read-only property — snapshot of all known tasks as Dict[task_id, SubagentTask].

@property
def tasks(self) -> Dict[str, SubagentTask]

Usage Examples

Fan Out, Gather, Map by Agent

from swarms import Agent, SubagentRegistry

researchers = [
    Agent(agent_name=f"Researcher-{i}", model_name="claude-sonnet-4-6", max_loops=1)
    for i in range(3)
]

registry = SubagentRegistry(max_workers=4)

topics = [
    "Recent advances in RAG retrieval",
    "Long-context LLM scaling laws",
    "Agent tool-use benchmarks",
]

task_ids = [
    registry.spawn(agent, topic)
    for agent, topic in zip(researchers, topics)
]

registry.gather(strategy="wait_all")
for tid in task_ids:
    task = registry.get_task(tid)
    print(task.status, str(task.result)[:80])

registry.shutdown()

Retries on Specific Exceptions

Retry only when the model call fails; surface anything else immediately. Agent.run() raises AgentLLMError once its own retries and any fallback_models are exhausted.

from swarms import Agent, AgentLLMError, SubagentRegistry

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

task_id = registry.spawn(
    agent=agent,
    task="Summarize the trade-offs between RAG and long-context models",
    max_retries=3,
    retry_on=[AgentLLMError],
    fail_fast=False,
)

registry.gather(timeout=60)
task = registry.get_task(task_id)
print(task.status, task.retries, task.result or task.error)

Take the First Successful Result

from swarms import Agent, SubagentRegistry

registry = SubagentRegistry()
candidates = [
    Agent(agent_name="GPT", model_name="gpt-5.4", max_loops=1),
    Agent(agent_name="Claude", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Gemini", model_name="gemini/gemini-2.5-pro", max_loops=1),
]

for agent in candidates:
    registry.spawn(agent, "Draft a one-sentence product tagline for a developer-tools startup")

first_results = registry.gather(strategy="wait_first", timeout=30)
print(first_results[0])

Depth-Limited Recursion

from swarms import Agent, SubagentRegistry

registry = SubagentRegistry(max_depth=2)
root = Agent(agent_name="Root", model_name="claude-sonnet-4-6", max_loops=1)

root_id = registry.spawn(root, "Plan a research project", depth=0)

# From inside a tool/handler the root agent can do:
# child_id = registry.spawn(child_agent, "Investigate sub-topic", parent_id=root_id, depth=1)
# Spawning at depth=3 would raise ValueError because max_depth=2.

How autonomous agents use the registry

An Agent running with max_loops="auto" delegates through the same registry. Its sub-agent tools work like this:

create_sub_agent builds each sub-agent with the parent’s model_name and tools. A sub-agent with tools gets max_loops=5 so it can call a tool and read the result; one without tools gets max_loops=1. The budget stays finite even when the parent runs with max_loops="auto".
assign_task spawns each assignment on a SubagentRegistry stored on the parent agent (created on first use with max_depth=3), with fail_fast=False and no retries. By default it waits for every task and returns one result or error per assignment.
check_sub_agent_status and cancel_sub_agent_tasks read and cancel tasks on that registry.

See Autonomous mode for the full tool set.

Source Code

View the source on GitHub.

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

SubagentRegistry

Overview

Installation

TaskStatus

SubagentTask

SubagentRegistry

Constructor

Methods

spawn()

get_task()

get_results()

cancel()

gather()

shutdown()

tasks

Usage Examples

Fan Out, Gather, Map by Agent

Retries on Specific Exceptions

Take the First Successful Result

Depth-Limited Recursion

How autonomous agents use the registry

Source Code

​Overview

​Installation

​TaskStatus

​SubagentTask

​SubagentRegistry

​Constructor

​Methods

​spawn()

​get_task()

​get_results()

​cancel()

​gather()

​shutdown()

​tasks

​Usage Examples

​Fan Out, Gather, Map by Agent

​Retries on Specific Exceptions

​Take the First Successful Result

​Depth-Limited Recursion

​How autonomous agents use the registry

​Source Code

Overview

Installation

TaskStatus

SubagentTask

SubagentRegistry

Constructor

Methods

spawn()

get_task()

get_results()

cancel()

gather()

shutdown()

tasks

Usage Examples

Fan Out, Gather, Map by Agent

Retries on Specific Exceptions

Take the First Successful Result

Depth-Limited Recursion

How autonomous agents use the registry

Source Code