strands.multiagent.base

Multi-Agent Base Class.

Provides minimal foundation for multi-agent patterns (Swarm, Graph).

Status

class Status(Enum)

Defined in: src/strands/multiagent/base.py:24

Execution status for both graphs and nodes.

Attributes:

• PENDING - Task has not started execution yet.
• EXECUTING - Task is currently running.
• COMPLETED - Task finished successfully.
• FAILED - Task encountered an error and could not complete.
• INTERRUPTED - Task was interrupted by user.

NodeResult

@dataclass
class NodeResult()

Defined in: src/strands/multiagent/base.py:43

Unified result from node execution - handles both Agent and nested MultiAgentBase results.

get_agent_results

def get_agent_results() -> list[AgentResult]

Defined in: src/strands/multiagent/base.py:59

Get all AgentResult objects from this node, flattened if nested.

to_dict

def to_dict() -> dict[str, Any]

Defined in: src/strands/multiagent/base.py:72

Convert NodeResult to JSON-serializable dict, ignoring state field.

from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "NodeResult"

Defined in: src/strands/multiagent/base.py:93

Rehydrate a NodeResult from persisted JSON.

MultiAgentResult

@dataclass
class MultiAgentResult()

Defined in: src/strands/multiagent/base.py:128

Result from multi-agent execution with accumulated metrics.

from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult"

Defined in: src/strands/multiagent/base.py:140

Rehydrate a MultiAgentResult from persisted JSON.

to_dict

def to_dict() -> dict[str, Any]

Defined in: src/strands/multiagent/base.py:164

Convert MultiAgentResult to JSON-serializable dict.

MultiAgentBase

class MultiAgentBase(ABC)

Defined in: src/strands/multiagent/base.py:178

Base class for multi-agent helpers.

This class integrates with existing Strands Agent instances and provides multi-agent orchestration capabilities.

Attributes:

• id - Unique MultiAgent id for session management,etc.

invoke_async

@abstractmethod
async def invoke_async(task: MultiAgentInput,
                       invocation_state: dict[str, Any] | None = None,
                       **kwargs: Any) -> MultiAgentResult

Defined in: src/strands/multiagent/base.py:191

Invoke asynchronously.

Arguments:

• task - The task to execute
• invocation_state - Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.
• **kwargs - Additional keyword arguments passed to underlying agents.

stream_async

async def stream_async(task: MultiAgentInput,
                       invocation_state: dict[str, Any] | None = None,
                       **kwargs: Any) -> AsyncIterator[dict[str, Any]]

Defined in: src/strands/multiagent/base.py:204

Stream events during multi-agent execution.

Default implementation executes invoke_async and yields the result as a single event. Subclasses can override this method to provide true streaming capabilities.

Arguments:

• task - The task to execute
• invocation_state - Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.
• **kwargs - Additional keyword arguments passed to underlying agents.

Yields:

Dictionary events containing multi-agent execution information including:

• Multi-agent coordination events (node start/complete, handoffs)
• Forwarded single-agent events with node context
• Final result event

__call__

def __call__(task: MultiAgentInput,
             invocation_state: dict[str, Any] | None = None,
             **kwargs: Any) -> MultiAgentResult

Defined in: src/strands/multiagent/base.py:229

Invoke synchronously.

Arguments:

• task - The task to execute
• invocation_state - Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.
• **kwargs - Additional keyword arguments passed to underlying agents.

serialize_state

def serialize_state() -> dict[str, Any]

Defined in: src/strands/multiagent/base.py:249

Return a JSON-serializable snapshot of the orchestrator state.

deserialize_state

def deserialize_state(payload: dict[str, Any]) -> None

Defined in: src/strands/multiagent/base.py:253

Restore orchestrator state from a session dict.