strands.experimental.hooks.events

Experimental hook events emitted as part of invoking Agents and BidiAgents.

This module defines the events that are emitted as Agents and BidiAgents run through the lifecycle of a request.

_DEPRECATED_ALIASES = {'BeforeToolInvocationEvent': BeforeToolCallEvent, 'AfterToolInvocationEvent': AfterToolCallEvent, 'BeforeModelInvocationEvent': BeforeModelCallEvent, 'AfterModelInvocationEvent': AfterModelCallEvent} module-attribute

AfterModelCallEvent dataclass

Bases: HookEvent

Event triggered after the model invocation completes.

This event is fired after the agent has finished calling the model, regardless of whether the invocation was successful or resulted in an error. Hook providers can use this event for cleanup, logging, or post-processing.

Note: This event uses reverse callback ordering, meaning callbacks registered later will be invoked first during cleanup.

Note: This event is not fired for invocations to structured_output.

Model Retrying

When retry_model is set to True by a hook callback, the agent will discard the current model response and invoke the model again. This has important implications for streaming consumers:

• Streaming events from the discarded response will have already been emitted to callers before the retry occurs. Agent invokers consuming streamed events should be prepared to handle this scenario, potentially by tracking retry state or implementing idempotent event processing
• The original model message is thrown away internally and not added to the conversation history

Attributes:

Name	Type	Description
stop_response	Optional[ModelStopResponse]	The model response data if invocation was successful, None if failed.
exception	Optional[Exception]	Exception if the model invocation failed, None if successful.
retry	bool	Whether to retry the model invocation. Can be set by hook callbacks to trigger a retry. When True, the current response is discarded and the model is called again. Defaults to False.

Source code in strands/hooks/events.py

@dataclass
class AfterModelCallEvent(HookEvent):
    """Event triggered after the model invocation completes.

    This event is fired after the agent has finished calling the model,
    regardless of whether the invocation was successful or resulted in an error.
    Hook providers can use this event for cleanup, logging, or post-processing.

    Note: This event uses reverse callback ordering, meaning callbacks registered
    later will be invoked first during cleanup.

    Note: This event is not fired for invocations to structured_output.

    Model Retrying:
        When ``retry_model`` is set to True by a hook callback, the agent will discard
        the current model response and invoke the model again. This has important
        implications for streaming consumers:

        - Streaming events from the discarded response will have already been emitted
          to callers before the retry occurs. Agent invokers consuming streamed events
          should be prepared to handle this scenario, potentially by tracking retry state
          or implementing idempotent event processing
        - The original model message is thrown away internally and not added to the
          conversation history

    Attributes:
        stop_response: The model response data if invocation was successful, None if failed.
        exception: Exception if the model invocation failed, None if successful.
        retry: Whether to retry the model invocation. Can be set by hook callbacks
            to trigger a retry. When True, the current response is discarded and the
            model is called again. Defaults to False.
    """

    @dataclass
    class ModelStopResponse:
        """Model response data from successful invocation.

        Attributes:
            stop_reason: The reason the model stopped generating.
            message: The generated message from the model.
        """

        message: Message
        stop_reason: StopReason

    stop_response: Optional[ModelStopResponse] = None
    exception: Optional[Exception] = None
    retry: bool = False

    def _can_write(self, name: str) -> bool:
        return name == "retry"

    @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.

ModelStopResponse dataclass

Model response data from successful invocation.

Attributes:

Name	Type	Description
stop_reason	StopReason	The reason the model stopped generating.
message	Message	The generated message from the model.

Source code in strands/hooks/events.py

@dataclass
class ModelStopResponse:
    """Model response data from successful invocation.

    Attributes:
        stop_reason: The reason the model stopped generating.
        message: The generated message from the model.
    """

    message: Message
    stop_reason: StopReason

AfterToolCallEvent dataclass

Bases: HookEvent

Event triggered after a tool invocation completes.

This event is fired after the agent has finished executing a tool, regardless of whether the execution was successful or resulted in an error. Hook providers can use this event for cleanup, logging, or post-processing.

Note: This event uses reverse callback ordering, meaning callbacks registered later will be invoked first during cleanup.

Attributes:

Name	Type	Description
selected_tool	Optional[AgentTool]	The tool that was invoked. It may be None if tool lookup failed.
tool_use	ToolUse	The tool parameters that were passed to the tool invoked.
invocation_state	dict[str, Any]	Keyword arguments that were passed to the tool
result	ToolResult	The result of the tool invocation. Either a ToolResult on success or an Exception if the tool execution failed.
cancel_message	str | None	The cancellation message if the user cancelled the tool call.

Source code in strands/hooks/events.py

@dataclass
class AfterToolCallEvent(HookEvent):
    """Event triggered after a tool invocation completes.

    This event is fired after the agent has finished executing a tool,
    regardless of whether the execution was successful or resulted in an error.
    Hook providers can use this event for cleanup, logging, or post-processing.

    Note: This event uses reverse callback ordering, meaning callbacks registered
    later will be invoked first during cleanup.

    Attributes:
        selected_tool: The tool that was invoked. It may be None if tool lookup failed.
        tool_use: The tool parameters that were passed to the tool invoked.
        invocation_state: Keyword arguments that were passed to the tool
        result: The result of the tool invocation. Either a ToolResult on success
            or an Exception if the tool execution failed.
        cancel_message: The cancellation message if the user cancelled the tool call.
    """

    selected_tool: Optional[AgentTool]
    tool_use: ToolUse
    invocation_state: dict[str, Any]
    result: ToolResult
    exception: Optional[Exception] = None
    cancel_message: str | None = None

    def _can_write(self, name: str) -> bool:
        return name == "result"

    @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.

AgentTool

Bases: ABC

Abstract base class for all SDK tools.

This class defines the interface that all tool implementations must follow. Each tool must provide its name, specification, and implement a stream method that executes the tool's functionality.

Source code in strands/types/tools.py

class AgentTool(ABC):
    """Abstract base class for all SDK tools.

    This class defines the interface that all tool implementations must follow. Each tool must provide its name,
    specification, and implement a stream method that executes the tool's functionality.
    """

    _is_dynamic: bool

    def __init__(self) -> None:
        """Initialize the base agent tool with default dynamic state."""
        self._is_dynamic = False

    @property
    @abstractmethod
    # pragma: no cover
    def tool_name(self) -> str:
        """The unique name of the tool used for identification and invocation."""
        pass

    @property
    @abstractmethod
    # pragma: no cover
    def tool_spec(self) -> ToolSpec:
        """Tool specification that describes its functionality and parameters."""
        pass

    @property
    @abstractmethod
    # pragma: no cover
    def tool_type(self) -> str:
        """The type of the tool implementation (e.g., 'python', 'javascript', 'lambda').

        Used for categorization and appropriate handling.
        """
        pass

    @property
    def supports_hot_reload(self) -> bool:
        """Whether the tool supports automatic reloading when modified.

        Returns:
            False by default.
        """
        return False

    @abstractmethod
    # pragma: no cover
    def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
        """Stream tool events and return the final result.

        Args:
            tool_use: The tool use request containing tool ID and parameters.
            invocation_state: Caller-provided kwargs that were passed to the agent when it was invoked (agent(),
                              agent.invoke_async(), etc.).
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Tool events with the last being the tool result.
        """
        ...

    @property
    def is_dynamic(self) -> bool:
        """Whether the tool was dynamically loaded during runtime.

        Dynamic tools may have different lifecycle management.

        Returns:
            True if loaded dynamically, False otherwise.
        """
        return self._is_dynamic

    def mark_dynamic(self) -> None:
        """Mark this tool as dynamically loaded."""
        self._is_dynamic = True

    def get_display_properties(self) -> dict[str, str]:
        """Get properties to display in UI representations of this tool.

        Subclasses can extend this to include additional properties.

        Returns:
            Dictionary of property names and their string values.
        """
        return {
            "Name": self.tool_name,
            "Type": self.tool_type,
        }

is_dynamic property

Whether the tool was dynamically loaded during runtime.

Dynamic tools may have different lifecycle management.

Returns:

Type	Description
bool	True if loaded dynamically, False otherwise.

supports_hot_reload property

Whether the tool supports automatic reloading when modified.

Returns:

Type	Description
bool	False by default.

tool_name abstractmethod property

The unique name of the tool used for identification and invocation.

tool_spec abstractmethod property

Tool specification that describes its functionality and parameters.

tool_type abstractmethod property

The type of the tool implementation (e.g., 'python', 'javascript', 'lambda').

Used for categorization and appropriate handling.

__init__()

Initialize the base agent tool with default dynamic state.

Source code in strands/types/tools.py

def __init__(self) -> None:
    """Initialize the base agent tool with default dynamic state."""
    self._is_dynamic = False

get_display_properties()

Get properties to display in UI representations of this tool.

Subclasses can extend this to include additional properties.

Returns:

Type	Description
dict[str, str]	Dictionary of property names and their string values.

Source code in strands/types/tools.py

def get_display_properties(self) -> dict[str, str]:
    """Get properties to display in UI representations of this tool.

    Subclasses can extend this to include additional properties.

    Returns:
        Dictionary of property names and their string values.
    """
    return {
        "Name": self.tool_name,
        "Type": self.tool_type,
    }

mark_dynamic()

Mark this tool as dynamically loaded.

Source code in strands/types/tools.py

def mark_dynamic(self) -> None:
    """Mark this tool as dynamically loaded."""
    self._is_dynamic = True

stream(tool_use, invocation_state, **kwargs) abstractmethod

Stream tool events and return the final result.

Parameters:

Name	Type	Description	Default
tool_use	ToolUse	The tool use request containing tool ID and parameters.	required
invocation_state	dict[str, Any]	Caller-provided kwargs that were passed to the agent when it was invoked (agent(), agent.invoke_async(), etc.).	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
ToolGenerator	Tool events with the last being the tool result.

Source code in strands/types/tools.py

@abstractmethod
# pragma: no cover
def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
    """Stream tool events and return the final result.

    Args:
        tool_use: The tool use request containing tool ID and parameters.
        invocation_state: Caller-provided kwargs that were passed to the agent when it was invoked (agent(),
                          agent.invoke_async(), etc.).
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Tool events with the last being the tool result.
    """
    ...

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")

BeforeModelCallEvent dataclass

Bases: HookEvent

Event triggered before the model is invoked.

This event is fired just before the agent calls the model for inference, allowing hook providers to inspect or modify the messages and configuration that will be sent to the model.

Note: This event is not fired for invocations to structured_output.

Source code in strands/hooks/events.py

@dataclass
class BeforeModelCallEvent(HookEvent):
    """Event triggered before the model is invoked.

    This event is fired just before the agent calls the model for inference,
    allowing hook providers to inspect or modify the messages and configuration
    that will be sent to the model.

    Note: This event is not fired for invocations to structured_output.
    """

    pass

BeforeToolCallEvent dataclass

Bases: HookEvent, _Interruptible

Event triggered before a tool is invoked.

This event is fired just before the agent executes a tool, allowing hook providers to inspect, modify, or replace the tool that will be executed. The selected_tool can be modified by hook callbacks to change which tool gets executed.

Attributes:

Name	Type	Description
selected_tool	Optional[AgentTool]	The tool that will be invoked. Can be modified by hooks to change which tool gets executed. This may be None if tool lookup failed.
tool_use	ToolUse	The tool parameters that will be passed to selected_tool.
invocation_state	dict[str, Any]	Keyword arguments that will be passed to the tool.
cancel_tool	bool | str	A user defined message that when set, will cancel the tool call. The message will be placed into a tool result with an error status. If set to True, Strands will cancel the tool call and use a default cancel message.

Source code in strands/hooks/events.py

@dataclass
class BeforeToolCallEvent(HookEvent, _Interruptible):
    """Event triggered before a tool is invoked.

    This event is fired just before the agent executes a tool, allowing hook
    providers to inspect, modify, or replace the tool that will be executed.
    The selected_tool can be modified by hook callbacks to change which tool
    gets executed.

    Attributes:
        selected_tool: The tool that will be invoked. Can be modified by hooks
            to change which tool gets executed. This may be None if tool lookup failed.
        tool_use: The tool parameters that will be passed to selected_tool.
        invocation_state: Keyword arguments that will be passed to the tool.
        cancel_tool: A user defined message that when set, will cancel the tool call.
            The message will be placed into a tool result with an error status. If set to `True`, Strands will cancel
            the tool call and use a default cancel message.
    """

    selected_tool: Optional[AgentTool]
    tool_use: ToolUse
    invocation_state: dict[str, Any]
    cancel_tool: bool | str = False

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

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

        Args:
            name: User defined name for the interrupt.

        Returns:
            Interrupt id.
        """
        return f"v1:before_tool_call:{self.tool_use['toolUseId']}:{uuid.uuid5(uuid.NAMESPACE_OID, name)}"

BidiAfterConnectionRestartEvent dataclass

Bases: BidiHookEvent

Event emitted after agent attempts to restart model connection after timeout.

Attribtues

exception: Populated if exception was raised during connection restart. None value means the restart was successful.

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

@dataclass
class BidiAfterConnectionRestartEvent(BidiHookEvent):
    """Event emitted after agent attempts to restart model connection after timeout.

    Attribtues:
        exception: Populated if exception was raised during connection restart.
            None value means the restart was successful.
    """

    exception: Exception | None = None

BidiAfterInvocationEvent dataclass

Bases: BidiHookEvent

Event triggered when BidiAgent ends a streaming session.

This event is fired after the BidiAgent has completed a streaming session, regardless of whether it completed successfully or encountered an error. Hook providers can use this event for cleanup, logging, or state persistence.

Note: This event uses reverse callback ordering, meaning callbacks registered later will be invoked first during cleanup.

This event is triggered at the end of agent.stop().

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

@dataclass
class BidiAfterInvocationEvent(BidiHookEvent):
    """Event triggered when BidiAgent ends a streaming session.

    This event is fired after the BidiAgent has completed a streaming session,
    regardless of whether it completed successfully or encountered an error.
    Hook providers can use this event for cleanup, logging, or state persistence.

    Note: This event uses reverse callback ordering, meaning callbacks registered
    later will be invoked first during cleanup.

    This event is triggered at the end of agent.stop().
    """

    @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.

BidiAfterToolCallEvent dataclass

Bases: BidiHookEvent

Event triggered after BidiAgent executes a tool.

This event is fired after the BidiAgent has finished executing a tool during a streaming session, regardless of whether the execution was successful or resulted in an error. Hook providers can use this event for cleanup, logging, or post-processing.

Note: This event uses reverse callback ordering, meaning callbacks registered later will be invoked first during cleanup.

Attributes:

Name	Type	Description
selected_tool	AgentTool | None	The tool that was invoked. It may be None if tool lookup failed.
tool_use	ToolUse	The tool parameters that were passed to the tool invoked.
invocation_state	dict[str, Any]	Keyword arguments that were passed to the tool.
result	ToolResult	The result of the tool invocation. Either a ToolResult on success or an Exception if the tool execution failed.
exception	Exception | None	Exception if the tool execution failed, None if successful.
cancel_message	str | None	The cancellation message if the user cancelled the tool call.

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

@dataclass
class BidiAfterToolCallEvent(BidiHookEvent):
    """Event triggered after BidiAgent executes a tool.

    This event is fired after the BidiAgent has finished executing a tool during
    a streaming session, regardless of whether the execution was successful or
    resulted in an error. Hook providers can use this event for cleanup, logging,
    or post-processing.

    Note: This event uses reverse callback ordering, meaning callbacks registered
    later will be invoked first during cleanup.

    Attributes:
        selected_tool: The tool that was invoked. It may be None if tool lookup failed.
        tool_use: The tool parameters that were passed to the tool invoked.
        invocation_state: Keyword arguments that were passed to the tool.
        result: The result of the tool invocation. Either a ToolResult on success
            or an Exception if the tool execution failed.
        exception: Exception if the tool execution failed, None if successful.
        cancel_message: The cancellation message if the user cancelled the tool call.
    """

    selected_tool: AgentTool | None
    tool_use: ToolUse
    invocation_state: dict[str, Any]
    result: ToolResult
    exception: Exception | None = None
    cancel_message: str | None = None

    def _can_write(self, name: str) -> bool:
        return name == "result"

    @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.

BidiAgent

Agent for bidirectional streaming conversations.

Enables real-time audio and text interaction with AI models through persistent connections. Supports concurrent tool execution and interruption handling.

Source code in strands/experimental/bidi/agent/agent.py

class BidiAgent:
    """Agent for bidirectional streaming conversations.

    Enables real-time audio and text interaction with AI models through persistent
    connections. Supports concurrent tool execution and interruption handling.
    """

    def __init__(
        self,
        model: BidiModel | str | None = None,
        tools: list[str | AgentTool | ToolProvider] | None = None,
        system_prompt: str | None = None,
        messages: Messages | None = None,
        record_direct_tool_call: bool = True,
        load_tools_from_directory: bool = False,
        agent_id: str | None = None,
        name: str | None = None,
        description: str | None = None,
        hooks: list[HookProvider] | None = None,
        state: AgentState | dict | None = None,
        session_manager: "SessionManager | None" = None,
        tool_executor: ToolExecutor | None = None,
        **kwargs: Any,
    ):
        """Initialize bidirectional agent.

        Args:
            model: BidiModel instance, string model_id, or None for default detection.
            tools: Optional list of tools with flexible format support.
            system_prompt: Optional system prompt for conversations.
            messages: Optional conversation history to initialize with.
            record_direct_tool_call: Whether to record direct tool calls in message history.
            load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
            agent_id: Optional ID for the agent, useful for connection management and multi-agent scenarios.
            name: Name of the Agent.
            description: Description of what the Agent does.
            hooks: Optional list of hook providers to register for lifecycle events.
            state: Stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
            session_manager: Manager for handling agent sessions including conversation history and state.
                If provided, enables session-based persistence and state management.
            tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).
            **kwargs: Additional configuration for future extensibility.

        Raises:
            ValueError: If model configuration is invalid or state is invalid type.
            TypeError: If model type is unsupported.
        """
        self.model = (
            BidiNovaSonicModel()
            if not model
            else BidiNovaSonicModel(model_id=model)
            if isinstance(model, str)
            else model
        )
        self.system_prompt = system_prompt
        self.messages = messages or []

        # Agent identification
        self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
        self.name = name or _DEFAULT_AGENT_NAME
        self.description = description

        # Tool execution configuration
        self.record_direct_tool_call = record_direct_tool_call
        self.load_tools_from_directory = load_tools_from_directory

        # Initialize tool registry
        self.tool_registry = ToolRegistry()

        if tools is not None:
            self.tool_registry.process_tools(tools)

        self.tool_registry.initialize_tools(self.load_tools_from_directory)

        # Initialize tool watcher if directory loading is enabled
        if self.load_tools_from_directory:
            self.tool_watcher = ToolWatcher(tool_registry=self.tool_registry)

        # Initialize agent state management
        if state is not None:
            if isinstance(state, dict):
                self.state = AgentState(state)
            elif isinstance(state, AgentState):
                self.state = state
            else:
                raise ValueError("state must be an AgentState object or a dict")
        else:
            self.state = AgentState()

        # Initialize other components
        self._tool_caller = _ToolCaller(self)

        # Initialize tool executor
        self.tool_executor = tool_executor or ConcurrentToolExecutor()

        # Initialize hooks registry
        self.hooks = HookRegistry()
        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)

        # Initialize session management functionality
        self._session_manager = session_manager
        if self._session_manager:
            self.hooks.add_hook(self._session_manager)

        self._loop = _BidiAgentLoop(self)

        # Emit initialization event
        self.hooks.invoke_callbacks(BidiAgentInitializedEvent(agent=self))

        # TODO: Determine if full support is required
        self._interrupt_state = _InterruptState()

        # Lock to ensure that paired messages are added to history in sequence without interference
        self._message_lock = asyncio.Lock()

        self._started = False

    @property
    def tool(self) -> _ToolCaller:
        """Call tool as a function.

        Returns:
            ToolCaller for method-style tool execution.

        Example:
            ```
            agent = BidiAgent(model=model, tools=[calculator])
            agent.tool.calculator(expression="2+2")
            ```
        """
        return self._tool_caller

    @property
    def tool_names(self) -> list[str]:
        """Get a list of all registered tool names.

        Returns:
            Names of all tools available to this agent.
        """
        all_tools = self.tool_registry.get_all_tools_config()
        return list(all_tools.keys())

    async def start(self, invocation_state: dict[str, Any] | None = None) -> None:
        """Start a persistent bidirectional conversation connection.

        Initializes the streaming connection and starts background tasks for processing
        model events, tool execution, and connection management.

        Args:
            invocation_state: Optional context to pass to tools during execution.
                This allows passing custom data (user_id, session_id, database connections, etc.)
                that tools can access via their invocation_state parameter.

        Raises:
            RuntimeError:
                If agent already started.

        Example:
            ```python
            await agent.start(invocation_state={
                "user_id": "user_123",
                "session_id": "session_456",
                "database": db_connection,
            })
            ```
        """
        if self._started:
            raise RuntimeError("agent already started | call stop before starting again")

        logger.debug("agent starting")
        await self._loop.start(invocation_state)
        self._started = True

    async def send(self, input_data: BidiAgentInput | dict[str, Any]) -> None:
        """Send input to the model (text, audio, image, or event dict).

        Unified method for sending text, audio, and image input to the model during
        an active conversation session. Accepts TypedEvent instances or plain dicts
        (e.g., from WebSocket clients) which are automatically reconstructed.

        Args:
            input_data: Can be:

                - str: Text message from user
                - BidiInputEvent: TypedEvent
                - dict: Event dictionary (will be reconstructed to TypedEvent)

        Raises:
            RuntimeError: If start has not been called.
            ValueError: If invalid input type.

        Example:
            await agent.send("Hello")
            await agent.send(BidiAudioInputEvent(audio="base64...", format="pcm", ...))
            await agent.send({"type": "bidirectional_text_input", "text": "Hello", "role": "user"})
        """
        if not self._started:
            raise RuntimeError("agent not started | call start before sending")

        input_event: BidiInputEvent

        if isinstance(input_data, str):
            input_event = BidiTextInputEvent(text=input_data)

        elif isinstance(input_data, BidiInputEvent):
            input_event = input_data

        elif isinstance(input_data, dict) and "type" in input_data:
            input_type = input_data["type"]
            input_data = {key: value for key, value in input_data.items() if key != "type"}
            if input_type == "bidi_text_input":
                input_event = BidiTextInputEvent(**input_data)
            elif input_type == "bidi_audio_input":
                input_event = BidiAudioInputEvent(**input_data)
            elif input_type == "bidi_image_input":
                input_event = BidiImageInputEvent(**input_data)
            else:
                raise ValueError(f"input_type=<{input_type}> | input type not supported")

        else:
            raise ValueError("invalid input | must be str, BidiInputEvent, or event dict")

        await self._loop.send(input_event)

    async def receive(self) -> AsyncGenerator[BidiOutputEvent, None]:
        """Receive events from the model including audio, text, and tool calls.

        Yields:
            Model output events processed by background tasks including audio output,
            text responses, tool calls, and connection updates.

        Raises:
            RuntimeError: If start has not been called.
        """
        if not self._started:
            raise RuntimeError("agent not started | call start before receiving")

        async for event in self._loop.receive():
            yield event

    async def stop(self) -> None:
        """End the conversation connection and cleanup all resources.

        Terminates the streaming connection, cancels background tasks, and
        closes the connection to the model provider.
        """
        self._started = False
        await self._loop.stop()

    async def __aenter__(self, invocation_state: dict[str, Any] | None = None) -> "BidiAgent":
        """Async context manager entry point.

        Automatically starts the bidirectional connection when entering the context.

        Args:
            invocation_state: Optional context to pass to tools during execution.
                This allows passing custom data (user_id, session_id, database connections, etc.)
                that tools can access via their invocation_state parameter.

        Returns:
            Self for use in the context.
        """
        logger.debug("context_manager=<enter> | starting agent")
        await self.start(invocation_state)
        return self

    async def __aexit__(self, *_: Any) -> None:
        """Async context manager exit point.

        Automatically ends the connection and cleans up resources including
        when exiting the context, regardless of whether an exception occurred.
        """
        logger.debug("context_manager=<exit> | stopping agent")
        await self.stop()

    async def run(
        self, inputs: list[BidiInput], outputs: list[BidiOutput], invocation_state: dict[str, Any] | None = None
    ) -> None:
        """Run the agent using provided IO channels for bidirectional communication.

        Args:
            inputs: Input callables to read data from a source
            outputs: Output callables to receive events from the agent
            invocation_state: Optional context to pass to tools during execution.
                This allows passing custom data (user_id, session_id, database connections, etc.)
                that tools can access via their invocation_state parameter.

        Example:
            ```python
            # Using model defaults:
            model = BidiNovaSonicModel()
            audio_io = BidiAudioIO()
            text_io = BidiTextIO()
            agent = BidiAgent(model=model, tools=[calculator])
            await agent.run(
                inputs=[audio_io.input()],
                outputs=[audio_io.output(), text_io.output()],
                invocation_state={"user_id": "user_123"}
            )

            # Using custom audio config:
            model = BidiNovaSonicModel(
                provider_config={"audio": {"input_rate": 48000, "output_rate": 24000}}
            )
            audio_io = BidiAudioIO()
            agent = BidiAgent(model=model, tools=[calculator])
            await agent.run(
                inputs=[audio_io.input()],
                outputs=[audio_io.output()],
            )
            ```
        """

        async def run_inputs() -> None:
            async def task(input_: BidiInput) -> None:
                while True:
                    event = await input_()
                    await self.send(event)

            await asyncio.gather(*[task(input_) for input_ in inputs])

        async def run_outputs(inputs_task: asyncio.Task) -> None:
            async for event in self.receive():
                await asyncio.gather(*[output(event) for output in outputs])

            inputs_task.cancel()

        try:
            await self.start(invocation_state)

            input_starts = [input_.start for input_ in inputs if isinstance(input_, BidiInput)]
            output_starts = [output.start for output in outputs if isinstance(output, BidiOutput)]
            for start in [*input_starts, *output_starts]:
                await start(self)

            async with _TaskGroup() as task_group:
                inputs_task = task_group.create_task(run_inputs())
                task_group.create_task(run_outputs(inputs_task))

        finally:
            input_stops = [input_.stop for input_ in inputs if isinstance(input_, BidiInput)]
            output_stops = [output.stop for output in outputs if isinstance(output, BidiOutput)]

            await stop_all(*input_stops, *output_stops, self.stop)

    async def _append_messages(self, *messages: Message) -> None:
        """Append messages to history in sequence without interference.

        The message lock ensures that paired messages are added to history in sequence without interference. For
        example, tool use and tool result messages must be added adjacent to each other.

        Args:
            *messages: List of messages to add into history.
        """
        async with self._message_lock:
            for message in messages:
                self.messages.append(message)
                await self.hooks.invoke_callbacks_async(BidiMessageAddedEvent(agent=self, message=message))

tool property

Call tool as a function.

Returns:

Type	Description
_ToolCaller	ToolCaller for method-style tool execution.

Example

agent = BidiAgent(model=model, tools=[calculator])
agent.tool.calculator(expression="2+2")

tool_names property

Get a list of all registered tool names.

Returns:

Type	Description
list[str]	Names of all tools available to this agent.

__aenter__(invocation_state=None) async

Async context manager entry point.

Automatically starts the bidirectional connection when entering the context.

Parameters:

Name	Type	Description	Default
invocation_state	dict[str, Any] | None	Optional context to pass to tools during execution. This allows passing custom data (user_id, session_id, database connections, etc.) that tools can access via their invocation_state parameter.	None

Returns:

Type	Description
BidiAgent	Self for use in the context.

Source code in strands/experimental/bidi/agent/agent.py

async def __aenter__(self, invocation_state: dict[str, Any] | None = None) -> "BidiAgent":
    """Async context manager entry point.

    Automatically starts the bidirectional connection when entering the context.

    Args:
        invocation_state: Optional context to pass to tools during execution.
            This allows passing custom data (user_id, session_id, database connections, etc.)
            that tools can access via their invocation_state parameter.

    Returns:
        Self for use in the context.
    """
    logger.debug("context_manager=<enter> | starting agent")
    await self.start(invocation_state)
    return self

__aexit__(*_) async

Async context manager exit point.

Automatically ends the connection and cleans up resources including when exiting the context, regardless of whether an exception occurred.

Source code in strands/experimental/bidi/agent/agent.py

async def __aexit__(self, *_: Any) -> None:
    """Async context manager exit point.

    Automatically ends the connection and cleans up resources including
    when exiting the context, regardless of whether an exception occurred.
    """
    logger.debug("context_manager=<exit> | stopping agent")
    await self.stop()

__init__(model=None, tools=None, system_prompt=None, messages=None, record_direct_tool_call=True, load_tools_from_directory=False, agent_id=None, name=None, description=None, hooks=None, state=None, session_manager=None, tool_executor=None, **kwargs)

Initialize bidirectional agent.

Parameters:

Name	Type	Description	Default
model	BidiModel | str | None	BidiModel instance, string model_id, or None for default detection.	None
tools	list[str | AgentTool | ToolProvider] | None	Optional list of tools with flexible format support.	None
system_prompt	str | None	Optional system prompt for conversations.	None
messages	Messages | None	Optional conversation history to initialize with.	None
record_direct_tool_call	bool	Whether to record direct tool calls in message history.	True
load_tools_from_directory	bool	Whether to load and automatically reload tools in the ./tools/ directory.	False
agent_id	str | None	Optional ID for the agent, useful for connection management and multi-agent scenarios.	None
name	str | None	Name of the Agent.	None
description	str | None	Description of what the Agent does.	None
hooks	list[HookProvider] | None	Optional list of hook providers to register for lifecycle events.	None
state	AgentState | dict | None	Stateful information for the agent. Can be either an AgentState object, or a json serializable dict.	None
session_manager	SessionManager | None	Manager for handling agent sessions including conversation history and state. If provided, enables session-based persistence and state management.	None
tool_executor	ToolExecutor | None	Definition of tool execution strategy (e.g., sequential, concurrent, etc.).	None
**kwargs	Any	Additional configuration for future extensibility.	{}

Raises:

Type	Description
ValueError	If model configuration is invalid or state is invalid type.
TypeError	If model type is unsupported.

Source code in strands/experimental/bidi/agent/agent.py

def __init__(
    self,
    model: BidiModel | str | None = None,
    tools: list[str | AgentTool | ToolProvider] | None = None,
    system_prompt: str | None = None,
    messages: Messages | None = None,
    record_direct_tool_call: bool = True,
    load_tools_from_directory: bool = False,
    agent_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    hooks: list[HookProvider] | None = None,
    state: AgentState | dict | None = None,
    session_manager: "SessionManager | None" = None,
    tool_executor: ToolExecutor | None = None,
    **kwargs: Any,
):
    """Initialize bidirectional agent.

    Args:
        model: BidiModel instance, string model_id, or None for default detection.
        tools: Optional list of tools with flexible format support.
        system_prompt: Optional system prompt for conversations.
        messages: Optional conversation history to initialize with.
        record_direct_tool_call: Whether to record direct tool calls in message history.
        load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
        agent_id: Optional ID for the agent, useful for connection management and multi-agent scenarios.
        name: Name of the Agent.
        description: Description of what the Agent does.
        hooks: Optional list of hook providers to register for lifecycle events.
        state: Stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
        session_manager: Manager for handling agent sessions including conversation history and state.
            If provided, enables session-based persistence and state management.
        tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).
        **kwargs: Additional configuration for future extensibility.

    Raises:
        ValueError: If model configuration is invalid or state is invalid type.
        TypeError: If model type is unsupported.
    """
    self.model = (
        BidiNovaSonicModel()
        if not model
        else BidiNovaSonicModel(model_id=model)
        if isinstance(model, str)
        else model
    )
    self.system_prompt = system_prompt
    self.messages = messages or []

    # Agent identification
    self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
    self.name = name or _DEFAULT_AGENT_NAME
    self.description = description

    # Tool execution configuration
    self.record_direct_tool_call = record_direct_tool_call
    self.load_tools_from_directory = load_tools_from_directory

    # Initialize tool registry
    self.tool_registry = ToolRegistry()

    if tools is not None:
        self.tool_registry.process_tools(tools)

    self.tool_registry.initialize_tools(self.load_tools_from_directory)

    # Initialize tool watcher if directory loading is enabled
    if self.load_tools_from_directory:
        self.tool_watcher = ToolWatcher(tool_registry=self.tool_registry)

    # Initialize agent state management
    if state is not None:
        if isinstance(state, dict):
            self.state = AgentState(state)
        elif isinstance(state, AgentState):
            self.state = state
        else:
            raise ValueError("state must be an AgentState object or a dict")
    else:
        self.state = AgentState()

    # Initialize other components
    self._tool_caller = _ToolCaller(self)

    # Initialize tool executor
    self.tool_executor = tool_executor or ConcurrentToolExecutor()

    # Initialize hooks registry
    self.hooks = HookRegistry()
    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)

    # Initialize session management functionality
    self._session_manager = session_manager
    if self._session_manager:
        self.hooks.add_hook(self._session_manager)

    self._loop = _BidiAgentLoop(self)

    # Emit initialization event
    self.hooks.invoke_callbacks(BidiAgentInitializedEvent(agent=self))

    # TODO: Determine if full support is required
    self._interrupt_state = _InterruptState()

    # Lock to ensure that paired messages are added to history in sequence without interference
    self._message_lock = asyncio.Lock()

    self._started = False

receive() async

Receive events from the model including audio, text, and tool calls.

Yields:

Type	Description
AsyncGenerator[BidiOutputEvent, None]	Model output events processed by background tasks including audio output,
AsyncGenerator[BidiOutputEvent, None]	text responses, tool calls, and connection updates.

Raises:

Type	Description
RuntimeError	If start has not been called.

Source code in strands/experimental/bidi/agent/agent.py

async def receive(self) -> AsyncGenerator[BidiOutputEvent, None]:
    """Receive events from the model including audio, text, and tool calls.

    Yields:
        Model output events processed by background tasks including audio output,
        text responses, tool calls, and connection updates.

    Raises:
        RuntimeError: If start has not been called.
    """
    if not self._started:
        raise RuntimeError("agent not started | call start before receiving")

    async for event in self._loop.receive():
        yield event

run(inputs, outputs, invocation_state=None) async

Run the agent using provided IO channels for bidirectional communication.

Parameters:

Name	Type	Description	Default
inputs	list[BidiInput]	Input callables to read data from a source	required
outputs	list[BidiOutput]	Output callables to receive events from the agent	required
invocation_state	dict[str, Any] | None	Optional context to pass to tools during execution. This allows passing custom data (user_id, session_id, database connections, etc.) that tools can access via their invocation_state parameter.	None

Example

# Using model defaults:
model = BidiNovaSonicModel()
audio_io = BidiAudioIO()
text_io = BidiTextIO()
agent = BidiAgent(model=model, tools=[calculator])
await agent.run(
    inputs=[audio_io.input()],
    outputs=[audio_io.output(), text_io.output()],
    invocation_state={"user_id": "user_123"}
)

# Using custom audio config:
model = BidiNovaSonicModel(
    provider_config={"audio": {"input_rate": 48000, "output_rate": 24000}}
)
audio_io = BidiAudioIO()
agent = BidiAgent(model=model, tools=[calculator])
await agent.run(
    inputs=[audio_io.input()],
    outputs=[audio_io.output()],
)

Source code in strands/experimental/bidi/agent/agent.py

async def run(
    self, inputs: list[BidiInput], outputs: list[BidiOutput], invocation_state: dict[str, Any] | None = None
) -> None:
    """Run the agent using provided IO channels for bidirectional communication.

    Args:
        inputs: Input callables to read data from a source
        outputs: Output callables to receive events from the agent
        invocation_state: Optional context to pass to tools during execution.
            This allows passing custom data (user_id, session_id, database connections, etc.)
            that tools can access via their invocation_state parameter.

    Example:
        ```python
        # Using model defaults:
        model = BidiNovaSonicModel()
        audio_io = BidiAudioIO()
        text_io = BidiTextIO()
        agent = BidiAgent(model=model, tools=[calculator])
        await agent.run(
            inputs=[audio_io.input()],
            outputs=[audio_io.output(), text_io.output()],
            invocation_state={"user_id": "user_123"}
        )

        # Using custom audio config:
        model = BidiNovaSonicModel(
            provider_config={"audio": {"input_rate": 48000, "output_rate": 24000}}
        )
        audio_io = BidiAudioIO()
        agent = BidiAgent(model=model, tools=[calculator])
        await agent.run(
            inputs=[audio_io.input()],
            outputs=[audio_io.output()],
        )
        ```
    """

    async def run_inputs() -> None:
        async def task(input_: BidiInput) -> None:
            while True:
                event = await input_()
                await self.send(event)

        await asyncio.gather(*[task(input_) for input_ in inputs])

    async def run_outputs(inputs_task: asyncio.Task) -> None:
        async for event in self.receive():
            await asyncio.gather(*[output(event) for output in outputs])

        inputs_task.cancel()

    try:
        await self.start(invocation_state)

        input_starts = [input_.start for input_ in inputs if isinstance(input_, BidiInput)]
        output_starts = [output.start for output in outputs if isinstance(output, BidiOutput)]
        for start in [*input_starts, *output_starts]:
            await start(self)

        async with _TaskGroup() as task_group:
            inputs_task = task_group.create_task(run_inputs())
            task_group.create_task(run_outputs(inputs_task))

    finally:
        input_stops = [input_.stop for input_ in inputs if isinstance(input_, BidiInput)]
        output_stops = [output.stop for output in outputs if isinstance(output, BidiOutput)]

        await stop_all(*input_stops, *output_stops, self.stop)

send(input_data) async

Send input to the model (text, audio, image, or event dict).

Unified method for sending text, audio, and image input to the model during an active conversation session. Accepts TypedEvent instances or plain dicts (e.g., from WebSocket clients) which are automatically reconstructed.

Parameters:

Name	Type	Description	Default
input_data	BidiAgentInput | dict[str, Any]	Can be: str: Text message from user BidiInputEvent: TypedEvent dict: Event dictionary (will be reconstructed to TypedEvent)	required

Raises:

Type	Description
RuntimeError	If start has not been called.
ValueError	If invalid input type.

Example

await agent.send("Hello") await agent.send(BidiAudioInputEvent(audio="base64...", format="pcm", ...)) await agent.send({"type": "bidirectional_text_input", "text": "Hello", "role": "user"})

Source code in strands/experimental/bidi/agent/agent.py

async def send(self, input_data: BidiAgentInput | dict[str, Any]) -> None:
    """Send input to the model (text, audio, image, or event dict).

    Unified method for sending text, audio, and image input to the model during
    an active conversation session. Accepts TypedEvent instances or plain dicts
    (e.g., from WebSocket clients) which are automatically reconstructed.

    Args:
        input_data: Can be:

            - str: Text message from user
            - BidiInputEvent: TypedEvent
            - dict: Event dictionary (will be reconstructed to TypedEvent)

    Raises:
        RuntimeError: If start has not been called.
        ValueError: If invalid input type.

    Example:
        await agent.send("Hello")
        await agent.send(BidiAudioInputEvent(audio="base64...", format="pcm", ...))
        await agent.send({"type": "bidirectional_text_input", "text": "Hello", "role": "user"})
    """
    if not self._started:
        raise RuntimeError("agent not started | call start before sending")

    input_event: BidiInputEvent

    if isinstance(input_data, str):
        input_event = BidiTextInputEvent(text=input_data)

    elif isinstance(input_data, BidiInputEvent):
        input_event = input_data

    elif isinstance(input_data, dict) and "type" in input_data:
        input_type = input_data["type"]
        input_data = {key: value for key, value in input_data.items() if key != "type"}
        if input_type == "bidi_text_input":
            input_event = BidiTextInputEvent(**input_data)
        elif input_type == "bidi_audio_input":
            input_event = BidiAudioInputEvent(**input_data)
        elif input_type == "bidi_image_input":
            input_event = BidiImageInputEvent(**input_data)
        else:
            raise ValueError(f"input_type=<{input_type}> | input type not supported")

    else:
        raise ValueError("invalid input | must be str, BidiInputEvent, or event dict")

    await self._loop.send(input_event)

start(invocation_state=None) async

Start a persistent bidirectional conversation connection.

Initializes the streaming connection and starts background tasks for processing model events, tool execution, and connection management.

Parameters:

Name	Type	Description	Default
invocation_state	dict[str, Any] | None	Optional context to pass to tools during execution. This allows passing custom data (user_id, session_id, database connections, etc.) that tools can access via their invocation_state parameter.	None

Raises:

Type	Description
RuntimeError	If agent already started.

Example

await agent.start(invocation_state={
    "user_id": "user_123",
    "session_id": "session_456",
    "database": db_connection,
})

Source code in strands/experimental/bidi/agent/agent.py

async def start(self, invocation_state: dict[str, Any] | None = None) -> None:
    """Start a persistent bidirectional conversation connection.

    Initializes the streaming connection and starts background tasks for processing
    model events, tool execution, and connection management.

    Args:
        invocation_state: Optional context to pass to tools during execution.
            This allows passing custom data (user_id, session_id, database connections, etc.)
            that tools can access via their invocation_state parameter.

    Raises:
        RuntimeError:
            If agent already started.

    Example:
        ```python
        await agent.start(invocation_state={
            "user_id": "user_123",
            "session_id": "session_456",
            "database": db_connection,
        })
        ```
    """
    if self._started:
        raise RuntimeError("agent already started | call stop before starting again")

    logger.debug("agent starting")
    await self._loop.start(invocation_state)
    self._started = True

stop() async

End the conversation connection and cleanup all resources.

Terminates the streaming connection, cancels background tasks, and closes the connection to the model provider.

Source code in strands/experimental/bidi/agent/agent.py

async def stop(self) -> None:
    """End the conversation connection and cleanup all resources.

    Terminates the streaming connection, cancels background tasks, and
    closes the connection to the model provider.
    """
    self._started = False
    await self._loop.stop()

BidiAgentInitializedEvent dataclass

Bases: BidiHookEvent

Event triggered when a BidiAgent has finished initialization.

This event is fired after the BidiAgent has been fully constructed and all built-in components have been initialized. Hook providers can use this event to perform setup tasks that require a fully initialized agent.

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

@dataclass
class BidiAgentInitializedEvent(BidiHookEvent):
    """Event triggered when a BidiAgent has finished initialization.

    This event is fired after the BidiAgent has been fully constructed and all
    built-in components have been initialized. Hook providers can use this
    event to perform setup tasks that require a fully initialized agent.
    """

    pass

BidiBeforeConnectionRestartEvent dataclass

Bases: BidiHookEvent

Event emitted before agent attempts to restart model connection after timeout.

Attributes:

Name	Type	Description
timeout_error	BidiModelTimeoutError	Timeout error reported by the model.

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

@dataclass
class BidiBeforeConnectionRestartEvent(BidiHookEvent):
    """Event emitted before agent attempts to restart model connection after timeout.

    Attributes:
        timeout_error: Timeout error reported by the model.
    """

    timeout_error: "BidiModelTimeoutError"

BidiBeforeInvocationEvent dataclass

Bases: BidiHookEvent

Event triggered when BidiAgent starts a streaming session.

This event is fired before the BidiAgent begins a streaming session, before any model connection or audio processing occurs. Hook providers can use this event to perform session-level setup, logging, or validation.

This event is triggered at the beginning of agent.start().

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

@dataclass
class BidiBeforeInvocationEvent(BidiHookEvent):
    """Event triggered when BidiAgent starts a streaming session.

    This event is fired before the BidiAgent begins a streaming session,
    before any model connection or audio processing occurs. Hook providers can
    use this event to perform session-level setup, logging, or validation.

    This event is triggered at the beginning of agent.start().
    """

    pass

BidiBeforeToolCallEvent dataclass

Bases: BidiHookEvent

Event triggered before BidiAgent executes a tool.

This event is fired just before the BidiAgent executes a tool during a streaming session, allowing hook providers to inspect, modify, or replace the tool that will be executed. The selected_tool can be modified by hook callbacks to change which tool gets executed.

Attributes:

Name	Type	Description
selected_tool	AgentTool | None	The tool that will be invoked. Can be modified by hooks to change which tool gets executed. This may be None if tool lookup failed.
tool_use	ToolUse	The tool parameters that will be passed to selected_tool.
invocation_state	dict[str, Any]	Keyword arguments that will be passed to the tool.
cancel_tool	bool | str	A user defined message that when set, will cancel the tool call. The message will be placed into a tool result with an error status. If set to True, Strands will cancel the tool call and use a default cancel message.

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

@dataclass
class BidiBeforeToolCallEvent(BidiHookEvent):
    """Event triggered before BidiAgent executes a tool.

    This event is fired just before the BidiAgent executes a tool during a streaming
    session, allowing hook providers to inspect, modify, or replace the tool that
    will be executed. The selected_tool can be modified by hook callbacks to change
    which tool gets executed.

    Attributes:
        selected_tool: The tool that will be invoked. Can be modified by hooks
            to change which tool gets executed. This may be None if tool lookup failed.
        tool_use: The tool parameters that will be passed to selected_tool.
        invocation_state: Keyword arguments that will be passed to the tool.
        cancel_tool: A user defined message that when set, will cancel the tool call.
            The message will be placed into a tool result with an error status. If set to `True`, Strands will cancel
            the tool call and use a default cancel message.
    """

    selected_tool: AgentTool | None
    tool_use: ToolUse
    invocation_state: dict[str, Any]
    cancel_tool: bool | str = False

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

BidiHookEvent dataclass

Bases: BaseHookEvent

Base class for BidiAgent hook events.

Attributes:

Name	Type	Description
agent	BidiAgent	The BidiAgent instance that triggered this event.

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

@dataclass
class BidiHookEvent(BaseHookEvent):
    """Base class for BidiAgent hook events.

    Attributes:
        agent: The BidiAgent instance that triggered this event.
    """

    agent: "BidiAgent"

BidiInterruptionEvent dataclass

Bases: BidiHookEvent

Event triggered when model generation is interrupted.

This event is fired when the user interrupts the assistant (e.g., by speaking during the assistant's response) or when an error causes interruption. This is specific to bidirectional streaming and doesn't exist in standard agents.

Hook providers can use this event to log interruptions, implement custom interruption handling, or trigger cleanup logic.

Attributes:

Name	Type	Description
reason	Literal['user_speech', 'error']	The reason for the interruption ("user_speech" or "error").
interrupted_response_id	str | None	Optional ID of the response that was interrupted.

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

@dataclass
class BidiInterruptionEvent(BidiHookEvent):
    """Event triggered when model generation is interrupted.

    This event is fired when the user interrupts the assistant (e.g., by speaking
    during the assistant's response) or when an error causes interruption. This is
    specific to bidirectional streaming and doesn't exist in standard agents.

    Hook providers can use this event to log interruptions, implement custom
    interruption handling, or trigger cleanup logic.

    Attributes:
        reason: The reason for the interruption ("user_speech" or "error").
        interrupted_response_id: Optional ID of the response that was interrupted.
    """

    reason: Literal["user_speech", "error"]
    interrupted_response_id: str | None = None

BidiMessageAddedEvent dataclass

Bases: BidiHookEvent

Event triggered when BidiAgent adds a message to the conversation.

This event is fired whenever the BidiAgent adds a new message to its internal message history, including user messages (from transcripts), assistant responses, and tool results. Hook providers can use this event for logging, monitoring, or implementing custom message processing logic.

Note: This event is only triggered for messages added by the framework itself, not for messages manually added by tools or external code.

Attributes:

Name	Type	Description
message	Message	The message that was added to the conversation history.

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

@dataclass
class BidiMessageAddedEvent(BidiHookEvent):
    """Event triggered when BidiAgent adds a message to the conversation.

    This event is fired whenever the BidiAgent adds a new message to its internal
    message history, including user messages (from transcripts), assistant responses,
    and tool results. Hook providers can use this event for logging, monitoring, or
    implementing custom message processing logic.

    Note: This event is only triggered for messages added by the framework
    itself, not for messages manually added by tools or external code.

    Attributes:
        message: The message that was added to the conversation history.
    """

    message: Message

BidiModelTimeoutError

Bases: Exception

Model timeout error.

Bidirectional models are often configured with a connection time limit. Nova sonic for example keeps the connection open for 8 minutes max. Upon receiving a timeout, the agent loop is configured to restart the model connection so as to create a seamless, uninterrupted experience for the user.

Source code in strands/experimental/bidi/models/model.py

class BidiModelTimeoutError(Exception):
    """Model timeout error.

    Bidirectional models are often configured with a connection time limit. Nova sonic for example keeps the connection
    open for 8 minutes max. Upon receiving a timeout, the agent loop is configured to restart the model connection so as
    to create a seamless, uninterrupted experience for the user.
    """

    def __init__(self, message: str, **restart_config: Any) -> None:
        """Initialize error.

        Args:
            message: Timeout message from model.
            **restart_config: Configure restart specific behaviors in the call to model start.
        """
        super().__init__(self, message)

        self.restart_config = restart_config

__init__(message, **restart_config)

Initialize error.

Parameters:

Name	Type	Description	Default
message	str	Timeout message from model.	required
**restart_config	Any	Configure restart specific behaviors in the call to model start.	{}

Source code in strands/experimental/bidi/models/model.py

def __init__(self, message: str, **restart_config: Any) -> None:
    """Initialize error.

    Args:
        message: Timeout message from model.
        **restart_config: Configure restart specific behaviors in the call to model start.
    """
    super().__init__(self, message)

    self.restart_config = restart_config

Message

Bases: TypedDict

A message in a conversation with the agent.

Attributes:

Name	Type	Description
content	List[ContentBlock]	The message content.
role	Role	The role of the message sender.

Source code in strands/types/content.py

class Message(TypedDict):
    """A message in a conversation with the agent.

    Attributes:
        content: The message content.
        role: The role of the message sender.
    """

    content: List[ContentBlock]
    role: Role

ToolResult

Bases: TypedDict

Result of a tool execution.

Attributes:

Name	Type	Description
content	list[ToolResultContent]	List of result content returned by the tool.
status	ToolResultStatus	The status of the tool execution ("success" or "error").
toolUseId	str	The unique identifier of the tool use request that produced this result.

Source code in strands/types/tools.py

class ToolResult(TypedDict):
    """Result of a tool execution.

    Attributes:
        content: List of result content returned by the tool.
        status: The status of the tool execution ("success" or "error").
        toolUseId: The unique identifier of the tool use request that produced this result.
    """

    content: list[ToolResultContent]
    status: ToolResultStatus
    toolUseId: str

ToolUse

Bases: TypedDict

A request from the model to use a specific tool with the provided input.

Attributes:

Name	Type	Description
input	Any	The input parameters for the tool. Can be any JSON-serializable type.
name	str	The name of the tool to invoke.
toolUseId	str	A unique identifier for this specific tool use request.

Source code in strands/types/tools.py

class ToolUse(TypedDict):
    """A request from the model to use a specific tool with the provided input.

    Attributes:
        input: The input parameters for the tool.
            Can be any JSON-serializable type.
        name: The name of the tool to invoke.
        toolUseId: A unique identifier for this specific tool use request.
    """

    input: Any
    name: str
    toolUseId: str

__getattr__(name)

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

def __getattr__(name: str) -> Any:
    if name in _DEPRECATED_ALIASES:
        warnings.warn(
            f"{name} has been moved to production with an updated name. "
            f"Use {_DEPRECATED_ALIASES[name].__name__} from strands.hooks instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        return _DEPRECATED_ALIASES[name]
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")