strands.experimental.hooks.multiagent.events

Multi-agent execution lifecycle events for hook system integration.

These events are fired by orchestrators (Graph/Swarm) at key points so hooks can persist, monitor, or debug execution. No intermediate state model is used—hooks read from the orchestrator directly.

AfterMultiAgentInvocationEvent dataclass

Bases: BaseHookEvent

Event triggered after orchestrator execution completes.

Attributes:

Name	Type	Description
source	MultiAgentBase	The multi-agent orchestrator instance
invocation_state	dict[str, Any] | None	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class AfterMultiAgentInvocationEvent(BaseHookEvent):
    """Event triggered after orchestrator execution completes.

    Attributes:
        source: The multi-agent orchestrator instance
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    invocation_state: dict[str, Any] | None = None

    @property
    def should_reverse_callbacks(self) -> bool:
        """True to invoke callbacks in reverse order."""
        return True

should_reverse_callbacks property

True to invoke callbacks in reverse order.

AfterNodeCallEvent dataclass

Bases: BaseHookEvent

Event triggered after individual node execution completes.

Attributes:

Name	Type	Description
source	MultiAgentBase	The multi-agent orchestrator instance
node_id	str	ID of the node that just completed execution
invocation_state	dict[str, Any] | None	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class AfterNodeCallEvent(BaseHookEvent):
    """Event triggered after individual node execution completes.

    Attributes:
        source: The multi-agent orchestrator instance
        node_id: ID of the node that just completed execution
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    node_id: str
    invocation_state: dict[str, Any] | None = None

    @property
    def should_reverse_callbacks(self) -> bool:
        """True to invoke callbacks in reverse order."""
        return True

should_reverse_callbacks property

True to invoke callbacks in reverse order.

BaseHookEvent dataclass

Base class for all hook events.

Source code in strands/hooks/registry.py

@dataclass
class BaseHookEvent:
    """Base class for all hook events."""

    @property
    def should_reverse_callbacks(self) -> bool:
        """Determine if callbacks for this event should be invoked in reverse order.

        Returns:
            False by default. Override to return True for events that should
            invoke callbacks in reverse order (e.g., cleanup/teardown events).
        """
        return False

    def _can_write(self, name: str) -> bool:
        """Check if the given property can be written to.

        Args:
            name: The name of the property to check.

        Returns:
            True if the property can be written to, False otherwise.
        """
        return False

    def __post_init__(self) -> None:
        """Disallow writes to non-approved properties."""
        # This is needed as otherwise the class can't be initialized at all, so we trigger
        # this after class initialization
        super().__setattr__("_disallow_writes", True)

    def __setattr__(self, name: str, value: Any) -> None:
        """Prevent setting attributes on hook events.

        Raises:
            AttributeError: Always raised to prevent setting attributes on hook events.
        """
        #  Allow setting attributes:
        #    - during init (when __dict__) doesn't exist
        #    - if the subclass specifically said the property is writable
        if not hasattr(self, "_disallow_writes") or self._can_write(name):
            return super().__setattr__(name, value)

        raise AttributeError(f"Property {name} is not writable")

should_reverse_callbacks property

Determine if callbacks for this event should be invoked in reverse order.

Returns:

Type	Description
bool	False by default. Override to return True for events that should
bool	invoke callbacks in reverse order (e.g., cleanup/teardown events).

__post_init__()

Disallow writes to non-approved properties.

Source code in strands/hooks/registry.py

def __post_init__(self) -> None:
    """Disallow writes to non-approved properties."""
    # This is needed as otherwise the class can't be initialized at all, so we trigger
    # this after class initialization
    super().__setattr__("_disallow_writes", True)

__setattr__(name, value)

Prevent setting attributes on hook events.

Raises:

Type	Description
AttributeError	Always raised to prevent setting attributes on hook events.

Source code in strands/hooks/registry.py

def __setattr__(self, name: str, value: Any) -> None:
    """Prevent setting attributes on hook events.

    Raises:
        AttributeError: Always raised to prevent setting attributes on hook events.
    """
    #  Allow setting attributes:
    #    - during init (when __dict__) doesn't exist
    #    - if the subclass specifically said the property is writable
    if not hasattr(self, "_disallow_writes") or self._can_write(name):
        return super().__setattr__(name, value)

    raise AttributeError(f"Property {name} is not writable")

BeforeMultiAgentInvocationEvent dataclass

Bases: BaseHookEvent

Event triggered before orchestrator execution starts.

Attributes:

Name	Type	Description
source	MultiAgentBase	The multi-agent orchestrator instance
invocation_state	dict[str, Any] | None	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class BeforeMultiAgentInvocationEvent(BaseHookEvent):
    """Event triggered before orchestrator execution starts.

    Attributes:
        source: The multi-agent orchestrator instance
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    invocation_state: dict[str, Any] | None = None

BeforeNodeCallEvent dataclass

Bases: BaseHookEvent, _Interruptible

Event triggered before individual node execution starts.

Attributes:

Name	Type	Description
source	MultiAgentBase	The multi-agent orchestrator instance
node_id	str	ID of the node about to execute
invocation_state	dict[str, Any] | None	Configuration that user passes in
cancel_node	bool | str	A user defined message that when set, will cancel the node execution with status FAILED. The message will be emitted under a MultiAgentNodeCancel event. If set to True, Strands will cancel the node using a default cancel message.

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class BeforeNodeCallEvent(BaseHookEvent, _Interruptible):
    """Event triggered before individual node execution starts.

    Attributes:
        source: The multi-agent orchestrator instance
        node_id: ID of the node about to execute
        invocation_state: Configuration that user passes in
        cancel_node: A user defined message that when set, will cancel the node execution with status FAILED.
            The message will be emitted under a MultiAgentNodeCancel event. If set to `True`, Strands will cancel the
            node using a default cancel message.
    """

    source: "MultiAgentBase"
    node_id: str
    invocation_state: dict[str, Any] | None = None
    cancel_node: bool | str = False

    def _can_write(self, name: str) -> bool:
        return name in ["cancel_node"]

    @override
    def _interrupt_id(self, name: str) -> str:
        """Unique id for the interrupt.

        Args:
            name: User defined name for the interrupt.

        Returns:
            Interrupt id.
        """
        node_id = uuid.uuid5(uuid.NAMESPACE_OID, self.node_id)
        call_id = uuid.uuid5(uuid.NAMESPACE_OID, name)
        return f"v1:before_node_call:{node_id}:{call_id}"

MultiAgentBase

Bases: ABC

Base class for multi-agent helpers.

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

Attributes:

Name	Type	Description
id	str	Unique MultiAgent id for session management,etc.

Source code in strands/multiagent/base.py

class MultiAgentBase(ABC):
    """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.
    """

    id: str

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

        Args:
            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.
        """
        raise NotImplementedError("invoke_async not implemented")

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """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.

        Args:
            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
        """
        # Default implementation for backward compatibility
        # Execute invoke_async and yield the result as a single event
        result = await self.invoke_async(task, invocation_state, **kwargs)
        yield {"result": result}

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

        Args:
            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.
        """
        if invocation_state is None:
            invocation_state = {}

        if kwargs:
            invocation_state.update(kwargs)
            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

        return run_async(lambda: self.invoke_async(task, invocation_state))

    def serialize_state(self) -> dict[str, Any]:
        """Return a JSON-serializable snapshot of the orchestrator state."""
        raise NotImplementedError

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore orchestrator state from a session dict."""
        raise NotImplementedError

    def _parse_trace_attributes(
        self, attributes: Mapping[str, AttributeValue] | None = None
    ) -> dict[str, AttributeValue]:
        trace_attributes: dict[str, AttributeValue] = {}
        if attributes:
            for k, v in attributes.items():
                if isinstance(v, (str, int, float, bool)) or (
                    isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
                ):
                    trace_attributes[k] = v
        return trace_attributes

__call__(task, invocation_state=None, **kwargs)

Invoke synchronously.

Parameters:

Name	Type	Description	Default
task	MultiAgentInput	The task to execute	required
invocation_state	dict[str, Any] | None	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	None
**kwargs	Any	Additional keyword arguments passed to underlying agents.	{}

Source code in strands/multiagent/base.py

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

    Args:
        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.
    """
    if invocation_state is None:
        invocation_state = {}

    if kwargs:
        invocation_state.update(kwargs)
        warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

    return run_async(lambda: self.invoke_async(task, invocation_state))

deserialize_state(payload)

Restore orchestrator state from a session dict.

Source code in strands/multiagent/base.py

def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore orchestrator state from a session dict."""
    raise NotImplementedError

invoke_async(task, invocation_state=None, **kwargs) abstractmethod async

Invoke asynchronously.

Parameters:

Name	Type	Description	Default
task	MultiAgentInput	The task to execute	required
invocation_state	dict[str, Any] | None	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	None
**kwargs	Any	Additional keyword arguments passed to underlying agents.	{}

Source code in strands/multiagent/base.py

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

    Args:
        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.
    """
    raise NotImplementedError("invoke_async not implemented")

serialize_state()

Return a JSON-serializable snapshot of the orchestrator state.

Source code in strands/multiagent/base.py

def serialize_state(self) -> dict[str, Any]:
    """Return a JSON-serializable snapshot of the orchestrator state."""
    raise NotImplementedError

stream_async(task, invocation_state=None, **kwargs) async

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.

Parameters:

Name	Type	Description	Default
task	MultiAgentInput	The task to execute	required
invocation_state	dict[str, Any] | None	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	None
**kwargs	Any	Additional keyword arguments passed to underlying agents.	{}

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	Dictionary events containing multi-agent execution information including:
AsyncIterator[dict[str, Any]]	Multi-agent coordination events (node start/complete, handoffs)
AsyncIterator[dict[str, Any]]	Forwarded single-agent events with node context
AsyncIterator[dict[str, Any]]	Final result event

Source code in strands/multiagent/base.py

async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """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.

    Args:
        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
    """
    # Default implementation for backward compatibility
    # Execute invoke_async and yield the result as a single event
    result = await self.invoke_async(task, invocation_state, **kwargs)
    yield {"result": result}

MultiAgentInitializedEvent dataclass

Bases: BaseHookEvent

Event triggered when multi-agent orchestrator initialized.

Attributes:

Name	Type	Description
source	MultiAgentBase	The multi-agent orchestrator instance
invocation_state	dict[str, Any] | None	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class MultiAgentInitializedEvent(BaseHookEvent):
    """Event triggered when multi-agent orchestrator initialized.

    Attributes:
        source: The multi-agent orchestrator instance
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    invocation_state: dict[str, Any] | None = None

_Interruptible

Bases: Protocol

Interface that adds interrupt support to hook events and tools.

Source code in strands/types/interrupt.py

class _Interruptible(Protocol):
    """Interface that adds interrupt support to hook events and tools."""

    def interrupt(self, name: str, reason: Any = None, response: Any = None) -> Any:
        """Trigger the interrupt with a reason.

        Args: name: User defined name for the interrupt.
                Must be unique across hook callbacks.
            reason: User provided reason for the interrupt.
            response: Preemptive response from user if available.

        Returns:
            The response from a human user when resuming from an interrupt state.

        Raises:
            InterruptException: If human input is required.
            RuntimeError: If agent instance attribute not set.
        """
        for attr_name in ["agent", "source"]:
            if hasattr(self, attr_name):
                agent = getattr(self, attr_name)
                break
        else:
            raise RuntimeError("agent instance attribute not set")

        id = self._interrupt_id(name)
        state = agent._interrupt_state

        interrupt_ = state.interrupts.setdefault(id, Interrupt(id, name, reason, response))
        if interrupt_.response is not None:
            return interrupt_.response

        raise InterruptException(interrupt_)

    def _interrupt_id(self, name: str) -> str:
        """Unique id for the interrupt.

        Args:
            name: User defined name for the interrupt.
            reason: User provided reason for the interrupt.

        Returns:
            Interrupt id.
        """
        ...

interrupt(name, reason=None, response=None)

Trigger the interrupt with a reason.

reason: User provided reason for the interrupt.
response: Preemptive response from user if available.

Returns:

Type	Description
Any	The response from a human user when resuming from an interrupt state.

Raises:

Type	Description
InterruptException	If human input is required.
RuntimeError	If agent instance attribute not set.

Source code in strands/types/interrupt.py

def interrupt(self, name: str, reason: Any = None, response: Any = None) -> Any:
    """Trigger the interrupt with a reason.

    Args: name: User defined name for the interrupt.
            Must be unique across hook callbacks.
        reason: User provided reason for the interrupt.
        response: Preemptive response from user if available.

    Returns:
        The response from a human user when resuming from an interrupt state.

    Raises:
        InterruptException: If human input is required.
        RuntimeError: If agent instance attribute not set.
    """
    for attr_name in ["agent", "source"]:
        if hasattr(self, attr_name):
            agent = getattr(self, attr_name)
            break
    else:
        raise RuntimeError("agent instance attribute not set")

    id = self._interrupt_id(name)
    state = agent._interrupt_state

    interrupt_ = state.interrupts.setdefault(id, Interrupt(id, name, reason, response))
    if interrupt_.response is not None:
        return interrupt_.response

    raise InterruptException(interrupt_)