feat: Implement the core system of typed hooks & callbacks

Relates to #231

Add the HookRegistry and a small subset of events (AgentInitializedEvent, StartRequestEvent, EndRequestEvent) as a POC for how hooks will work.

Author: zastrowm

src/strands/agent/agent.py

@@ -22,6 +22,8 @@
 from ..event_loop.event_loop import event_loop_cycle
 from ..handlers.callback_handler import PrintingCallbackHandler, null_callback_handler
 from ..handlers.tool_handler import AgentToolHandler
+from ..hooks.events import AgentInitializedEvent, EndRequestEvent, StartRequestEvent
+from ..hooks.registry import HookRegistry
 from ..models.bedrock import BedrockModel
 from ..telemetry.metrics import EventLoopMetrics
 from ..telemetry.tracer import get_tracer
@@ -320,6 +322,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.
@@ -487,6 +493,8 @@ def _run_loop(
         self, callback_handler: Callable[..., Any], 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}}
@@ -501,6 +509,7 @@ def _run_loop(
 
         finally:
             self.conversation_manager.apply_management(self)
+            self._hooks.invoke_callbacks(EndRequestEvent(agent=self))
 
     def _execute_event_loop_cycle(
         self, callback_handler: Callable[..., Any], kwargs: dict[str, Any]

src/strands/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.
+"""
+# Exporting nothing yet until hook stabilize
+# from .events import AgentInitializedEvent, EndRequestEvent, StartRequestEvent
+# from .registry import HookCallback, HookEvent, HookProvider, HookRegistry
+#
+# __all__ = [
+#     "AgentInitializedEvent",
+#     "StartRequestEvent",
+#     "EndRequestEvent",
+#     "HookEvent",
+#     "HookProvider",
+#     "HookCallback",
+#     "HookRegistry",
+# ]

src/strands/hooks/events.py

@@ -0,0 +1,54 @@
+"""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.
+    """
+
+    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.
+    """
+
+    @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/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, Any, Callable, Generator, Protocol, Type, TypeVar
+
+if TYPE_CHECKING:
+    from strands 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)
+
+
+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):
+    """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: Any) -> 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):
+        """Initialize an empty hook registry."""
+        self._registered_callbacks = {}
+
+    def add_callback(self, event_type: Type, callback: HookCallback) -> 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, 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