strands.types.a2a

Additional A2A types.

A2AResponse = tuple[Task, TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None] | Message | Any module-attribute

A2AStreamEvent

Bases: TypedEvent

Event emitted for every update received from the remote A2A server.

This event wraps all A2A response types during streaming, including: - Partial task updates (TaskArtifactUpdateEvent) - Status updates (TaskStatusUpdateEvent) - Complete messages (Message) - Final task completions

The event is emitted for EVERY update from the server, regardless of whether it represents a complete or partial response. When streaming completes, an AgentResultEvent containing the final AgentResult is also emitted after all A2AStreamEvents.

Source code in strands/types/a2a.py

class A2AStreamEvent(TypedEvent):
    """Event emitted for every update received from the remote A2A server.

    This event wraps all A2A response types during streaming, including:
    - Partial task updates (TaskArtifactUpdateEvent)
    - Status updates (TaskStatusUpdateEvent)
    - Complete messages (Message)
    - Final task completions

    The event is emitted for EVERY update from the server, regardless of whether
    it represents a complete or partial response. When streaming completes, an
    AgentResultEvent containing the final AgentResult is also emitted after all
    A2AStreamEvents.
    """

    def __init__(self, a2a_event: A2AResponse) -> None:
        """Initialize with A2A event.

        Args:
            a2a_event: The original A2A event (Task tuple or Message)
        """
        super().__init__(
            {
                "type": "a2a_stream",
                "event": a2a_event,  # Nest A2A event to avoid field conflicts
            }
        )

__init__(a2a_event)

Initialize with A2A event.

Parameters:

Name	Type	Description	Default
a2a_event	A2AResponse	The original A2A event (Task tuple or Message)	required

Source code in strands/types/a2a.py

def __init__(self, a2a_event: A2AResponse) -> None:
    """Initialize with A2A event.

    Args:
        a2a_event: The original A2A event (Task tuple or Message)
    """
    super().__init__(
        {
            "type": "a2a_stream",
            "event": a2a_event,  # Nest A2A event to avoid field conflicts
        }
    )

TypedEvent

Bases: dict

Base class for all typed events in the agent system.

Source code in strands/types/_events.py

class TypedEvent(dict):
    """Base class for all typed events in the agent system."""

    def __init__(self, data: dict[str, Any] | None = None) -> None:
        """Initialize the typed event with optional data.

        Args:
            data: Optional dictionary of event data to initialize with
        """
        super().__init__(data or {})

    @property
    def is_callback_event(self) -> bool:
        """True if this event should trigger the callback_handler to fire."""
        return True

    def as_dict(self) -> dict:
        """Convert this event to a raw dictionary for emitting purposes."""
        return {**self}

    def prepare(self, invocation_state: dict) -> None:
        """Prepare the event for emission by adding invocation state.

        This allows a subset of events to merge with the invocation_state without needing to
        pass around the invocation_state throughout the system.
        """
        ...

is_callback_event property

True if this event should trigger the callback_handler to fire.

__init__(data=None)

Initialize the typed event with optional data.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | None	Optional dictionary of event data to initialize with	None

Source code in strands/types/_events.py

def __init__(self, data: dict[str, Any] | None = None) -> None:
    """Initialize the typed event with optional data.

    Args:
        data: Optional dictionary of event data to initialize with
    """
    super().__init__(data or {})

as_dict()

Convert this event to a raw dictionary for emitting purposes.

Source code in strands/types/_events.py

def as_dict(self) -> dict:
    """Convert this event to a raw dictionary for emitting purposes."""
    return {**self}

prepare(invocation_state)

Prepare the event for emission by adding invocation state.

This allows a subset of events to merge with the invocation_state without needing to pass around the invocation_state throughout the system.

Source code in strands/types/_events.py

def prepare(self, invocation_state: dict) -> None:
    """Prepare the event for emission by adding invocation state.

    This allows a subset of events to merge with the invocation_state without needing to
    pass around the invocation_state throughout the system.
    """
    ...