feat: Implement the core system of typed hooks & callbacks

src/strands/agent/agent.py

@@ -20,6 +20,7 @@
 from pydantic import BaseModel
 
 from ..event_loop.event_loop import event_loop_cycle
+from ..experimental.hooks import AgentInitializedEvent, EndRequestEvent, HookRegistry, StartRequestEvent
 from ..handlers.callback_handler import PrintingCallbackHandler, null_callback_handler
 from ..handlers.tool_handler import AgentToolHandler
 from ..models.bedrock import BedrockModel
@@ -308,6 +309,10 @@ def __init__(
         self.name = name
         self.description = description
 
+        self._hooks = HookRegistry()
+        # Register built-in hook providers (like ConversationManager) here
+        self._hooks.invoke_callbacks(AgentInitializedEvent(agent=self))
+
     @property
     def tool(self) -> ToolCaller:
         """Call tool as a function.
@@ -405,21 +410,26 @@ def structured_output(self, output_model: Type[T], prompt: Optional[str] = None)
                 that the agent will use when responding.
             prompt: The prompt to use for the agent.
         """
-        messages = self.messages
-        if not messages and not prompt:
-            raise ValueError("No conversation history or prompt provided")
+        self._hooks.invoke_callbacks(StartRequestEvent(agent=self))
 
-        # add the prompt as the last message
-        if prompt:
-            messages.append({"role": "user", "content": [{"text": prompt}]})
+        try:
+            messages = self.messages
+            if not messages and not prompt:
+                raise ValueError("No conversation history or prompt provided")
 
-        # get the structured output from the model
-        events = self.model.structured_output(output_model, messages)
-        for event in events:
-            if "callback" in event:
-                self.callback_handler(**cast(dict, event["callback"]))
+            # add the prompt as the last message
+            if prompt:
+                messages.append({"role": "user", "content": [{"text": prompt}]})
 
-        return event["output"]
+            # get the structured output from the model
+            events = self.model.structured_output(output_model, messages)
+            for event in events:
+                if "callback" in event:
+                    self.callback_handler(**cast(dict, event["callback"]))
+
+            return event["output"]
+        finally:
+            self._hooks.invoke_callbacks(EndRequestEvent(agent=self))
 
     async def stream_async(self, prompt: str, **kwargs: Any) -> AsyncIterator[Any]:
         """Process a natural language prompt and yield events as an async iterator.
@@ -473,6 +483,8 @@ async def stream_async(self, prompt: str, **kwargs: Any) -> AsyncIterator[Any]:
 
     def _run_loop(self, prompt: str, kwargs: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
         """Execute the agent's event loop with the given prompt and parameters."""
+        self._hooks.invoke_callbacks(StartRequestEvent(agent=self))
+
         try:
             # Extract key parameters
             yield {"callback": {"init_event_loop": True, **kwargs}}
@@ -487,6 +499,7 @@ def _run_loop(self, prompt: str, kwargs: dict[str, Any]) -> Generator[dict[str,
 
         finally:
             self.conversation_manager.apply_management(self)
+            self._hooks.invoke_callbacks(EndRequestEvent(agent=self))
 
     def _execute_event_loop_cycle(self, kwargs: dict[str, Any]) -> Generator[dict[str, Any], None, None]:
         """Execute the event loop cycle with retry logic for context window limits.

src/strands/experimental/__init__.py

@@ -0,0 +1,4 @@
+"""Experimental features.
+
+This module implements experimental features that are subject to change in future revisions without notice.
+"""

src/strands/experimental/hooks/__init__.py

@@ -0,0 +1,43 @@
+"""Typed hook system for extending agent functionality.
+
+This module provides a composable mechanism for building objects that can hook
+into specific events during the agent lifecycle. The hook system enables both
+built-in SDK components and user code to react to or modify agent behavior
+through strongly-typed event callbacks.
+
+Example Usage:
+    ```python
+    from strands.hooks import HookProvider, HookRegistry
+    from strands.hooks.events import StartRequestEvent, EndRequestEvent
+
+    class LoggingHooks(HookProvider):
+        def register_hooks(self, registry: HookRegistry) -> None:
+            registry.add_callback(StartRequestEvent, self.log_start)
+            registry.add_callback(EndRequestEvent, self.log_end)
+
+        def log_start(self, event: StartRequestEvent) -> None:
+            print(f"Request started for {event.agent.name}")
+
+        def log_end(self, event: EndRequestEvent) -> None:
+            print(f"Request completed for {event.agent.name}")
+
+    # Use with agent
+    agent = Agent(hooks=[LoggingHooks()])
+    ```
+
+This replaces the older callback_handler approach with a more composable,
+type-safe system that supports multiple subscribers per event type.
+"""
+
+from .events import AgentInitializedEvent, EndRequestEvent, StartRequestEvent
+from .registry import HookCallback, HookEvent, HookProvider, HookRegistry
+
+__all__ = [
+    "AgentInitializedEvent",
+    "StartRequestEvent",
+    "EndRequestEvent",
+    "HookEvent",
+    "HookProvider",
+    "HookCallback",
+    "HookRegistry",
+]

src/strands/experimental/hooks/events.py

@@ -0,0 +1,64 @@
+"""Hook events emitted as part of invoking Agents.
+
+This module defines the events that are emitted as Agents run through the lifecycle of a request.
+"""
+
+from dataclasses import dataclass
+
+from .registry import HookEvent
+
+
+@dataclass
+class AgentInitializedEvent(HookEvent):
+    """Event triggered when an agent has finished initialization.
+
+    This event is fired after the agent 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
+
+
+@dataclass
+class StartRequestEvent(HookEvent):
+    """Event triggered at the beginning of a new agent request.
+
+    This event is fired when the agent begins processing a new user request,
+    before any model inference or tool execution occurs. Hook providers can
+    use this event to perform request-level setup, logging, or validation.
+
+    This event is triggered at the beginning of the following api calls:
+      - Agent.__call__
+      - Agent.stream_async
+      - Agent.structured_output
+    """
+
+    pass
+
+
+@dataclass
+class EndRequestEvent(HookEvent):
+    """Event triggered at the end of an agent request.
+
+    This event is fired after the agent has completed processing a request,
+    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 the following api calls:
+      - Agent.__call__
+      - Agent.stream_async
+      - Agent.structured_output
+    """
+
+    @property
+    def should_reverse_callbacks(self) -> bool:
+        """Return True to invoke callbacks in reverse order for proper cleanup.
+
+        Returns:
+            True, indicating callbacks should be invoked in reverse order.
+        """
+        return True

src/strands/experimental/hooks/registry.py

@@ -0,0 +1,195 @@
+"""Hook registry system for managing event callbacks in the Strands Agent SDK.
+
+This module provides the core infrastructure for the typed hook system, enabling
+composable extension of agent functionality through strongly-typed event callbacks.
+The registry manages the mapping between event types and their associated callback
+functions, supporting both individual callback registration and bulk registration
+via hook provider objects.
+"""
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Callable, Generator, Generic, Protocol, Type, TypeVar
+
+if TYPE_CHECKING:
+    from ...agent import Agent
+
+
+@dataclass
+class HookEvent:
+    """Base class for all hook events.
+
+    Attributes:
+        agent: The agent instance that triggered this event.
+    """
+
+    agent: "Agent"
+
+    @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
+
+
+T = TypeVar("T", bound=Callable)
+TEvent = TypeVar("TEvent", bound=HookEvent, contravariant=True)
+
+
+class HookProvider(Protocol):
+    """Protocol for objects that provide hook callbacks to an agent.
+
+    Hook providers offer a composable way to extend agent functionality by
+    subscribing to various events in the agent lifecycle. This protocol enables
+    building reusable components that can hook into agent events.
+
+    Example:
+        ```python
+        class MyHookProvider(HookProvider):
+            def register_hooks(self, registry: HookRegistry) -> None:
+                hooks.add_callback(StartRequestEvent, self.on_request_start)
+                hooks.add_callback(EndRequestEvent, self.on_request_end)
+
+        agent = Agent(hooks=[MyHookProvider()])
+        ```
+    """
+
+    def register_hooks(self, registry: "HookRegistry") -> None:
+        """Register callback functions for specific event types.
+
+        Args:
+            registry: The hook registry to register callbacks with.
+        """
+        ...
+
+
+class HookCallback(Protocol, Generic[TEvent]):
+    """Protocol for callback functions that handle hook events.
+
+    Hook callbacks are functions that receive a single strongly-typed event
+    argument and perform some action in response. They should not return
+    values and any exceptions they raise will propagate to the caller.
+
+    Example:
+        ```python
+        def my_callback(event: StartRequestEvent) -> None:
+            print(f"Request started for agent: {event.agent.name}")
+        ```
+    """
+
+    def __call__(self, event: TEvent) -> None:
+        """Handle a hook event.
+
+        Args:
+            event: The strongly-typed event to handle.
+        """
+        ...
+
+
+class HookRegistry:
+    """Registry for managing hook callbacks associated with event types.
+
+    The HookRegistry maintains a mapping of event types to callback functions
+    and provides methods for registering callbacks and invoking them when
+    events occur.
+
+    The registry handles callback ordering, including reverse ordering for
+    cleanup events, and provides type-safe event dispatching.
+    """
+
+    def __init__(self) -> None:
+        """Initialize an empty hook registry."""
+        self._registered_callbacks: dict[Type, list[HookCallback]] = {}
+
+    def add_callback(self, event_type: Type[TEvent], callback: HookCallback[TEvent]) -> None:
+        """Register a callback function for a specific event type.
+
+        Args:
+            event_type: The class type of events this callback should handle.
+            callback: The callback function to invoke when events of this type occur.
+
+        Example:
+            ```python
+            def my_handler(event: StartRequestEvent):
+                print("Request started")
+
+            registry.add_callback(StartRequestEvent, my_handler)
+            ```
+        """
+        callbacks = self._registered_callbacks.setdefault(event_type, [])
+        callbacks.append(callback)
+
+    def add_hook(self, hook: HookProvider) -> None:
+        """Register all callbacks from a hook provider.
+
+        This method allows bulk registration of callbacks by delegating to
+        the hook provider's register_hooks method. This is the preferred
+        way to register multiple related callbacks.
+
+        Args:
+            hook: The hook provider containing callbacks to register.
+
+        Example:
+            ```python
+            class MyHooks(HookProvider):
+                def register_hooks(self, registry: HookRegistry):
+                    registry.add_callback(StartRequestEvent, self.on_start)
+                    registry.add_callback(EndRequestEvent, self.on_end)
+
+            registry.add_hook(MyHooks())
+            ```
+        """
+        hook.register_hooks(self)
+
+    def invoke_callbacks(self, event: TEvent) -> None:
+        """Invoke all registered callbacks for the given event.
+
+        This method finds all callbacks registered for the event's type and
+        invokes them in the appropriate order. For events with is_after_callback=True,
+        callbacks are invoked in reverse registration order.
+
+        Args:
+            event: The event to dispatch to registered callbacks.
+
+        Raises:
+            Any exceptions raised by callback functions will propagate to the caller.
+
+        Example:
+            ```python
+            event = StartRequestEvent(agent=my_agent)
+            registry.invoke_callbacks(event)
+            ```
+        """
+        for callback in self.get_callbacks_for(event):
+            callback(event)
+
+    def get_callbacks_for(self, event: TEvent) -> Generator[HookCallback[TEvent], None, None]:
+        """Get callbacks registered for the given event in the appropriate order.
+
+        This method returns callbacks in registration order for normal events,
+        or reverse registration order for events that have is_after_callback=True.
+        This enables proper cleanup ordering for teardown events.
+
+        Args:
+            event: The event to get callbacks for.
+
+        Yields:
+            Callback functions registered for this event type, in the appropriate order.
+
+        Example:
+            ```python
+            event = EndRequestEvent(agent=my_agent)
+            for callback in registry.get_callbacks_for(event):
+                callback(event)
+            ```
+        """
+        event_type = type(event)
+
+        callbacks = self._registered_callbacks.get(event_type, [])
+        if event.should_reverse_callbacks:
+            yield from reversed(callbacks)
+        else:
+            yield from callbacks