feat(strands-py): add optional hook order (#2559)

Author: lizradway

site/src/content/docs/user-guide/concepts/agents/hooks.mdx

@@ -392,10 +392,33 @@ Most event properties are read-only to prevent unintended modifications. However
 <Tabs>
 <Tab label="Python">
 
-After event callbacks run in reverse registration order for cleanup symmetry:
+By default, After event callbacks run in reverse registration order for cleanup symmetry. You can override this with explicit priority using the `order` option — lower values run first.
+
+The SDK exports convenience presets that mark where the SDK's own hooks run, so you can position yours relative to them:
+
+- `HookOrder.SDK_FIRST` (-100) — where the SDK's earliest hooks run
+- `HookOrder.DEFAULT` (0) — implicit when no order is specified
+- `HookOrder.SDK_LAST` (100) — where the SDK's latest hooks run
+
+These are not enforced bounds — any numeric value works. Use values beyond them (e.g. `SDK_FIRST - 1`) to run before or after the SDK's hooks, or `float('-inf')`/`float('inf')` for guaranteed absolute ordering.
+
+```python
+from strands import Agent
+from strands.hooks import BeforeModelCallEvent, HookOrder
+
+agent = Agent()
+
+def early_hook(event: BeforeModelCallEvent) -> None:
+    print("I run first")
 
-- **Before**: A, B, C (registration order)
-- **After**: C, B, A (reverse registration order)
+def late_hook(event: BeforeModelCallEvent) -> None:
+    print("I run last")
+
+agent.add_hook(early_hook, order=HookOrder.SDK_FIRST)
+agent.add_hook(late_hook, order=HookOrder.SDK_LAST)
+```
+
+Within the same order group, Before events preserve registration order and After events reverse it.
 
 </Tab>
 <Tab label="TypeScript">

strands-py/src/strands/agent/agent.py

@@ -46,6 +46,7 @@
     AgentInitializedEvent,
     BeforeInvocationEvent,
     HookCallback,
+    HookOrder,
     HookProvider,
     HookRegistry,
     MessageAddedEvent,
@@ -725,7 +726,11 @@ def cleanup(self) -> None:
         self.tool_registry.cleanup()
 
     def add_hook(
-        self, callback: HookCallback[TEvent], event_type: type[TEvent] | list[type[TEvent]] | None = None
+        self,
+        callback: HookCallback[TEvent],
+        event_type: type[TEvent] | list[type[TEvent]] | None = None,
+        *,
+        order: float = HookOrder.DEFAULT,
     ) -> None:
         """Register a callback function for a specific event type.
 
@@ -745,6 +750,8 @@ def add_hook(
                 Can be a single type, a list of types, or None to infer from
                 the callback's first parameter type hint. If a list is provided,
                 the callback is registered for each type in the list.
+            order: Execution priority. Lower values execute first.
+                Use HookOrder.SDK_FIRST (-100), HookOrder.DEFAULT (0), or HookOrder.SDK_LAST (100).
 
         Raises:
             ValueError: If event_type is not provided and cannot be inferred from
@@ -776,7 +783,7 @@ def multi_handler(event) -> None:
         Docs:
             https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/hooks/
         """
-        self.hooks.add_callback(event_type, callback)
+        self.hooks.add_callback(event_type, callback, order=order)
 
     def __del__(self) -> None:
         """Clean up resources when agent is garbage collected."""

strands-py/src/strands/hooks/__init__.py

@@ -45,7 +45,7 @@ def log_end(self, event: AfterInvocationEvent) -> None:
     MessageAddedEvent,
     MultiAgentInitializedEvent,
 )
-from .registry import BaseHookEvent, HookCallback, HookEvent, HookProvider, HookRegistry
+from .registry import BaseHookEvent, HookCallback, HookEvent, HookOrder, HookProvider, HookRegistry
 
 __all__ = [
     "AgentInitializedEvent",
@@ -57,6 +57,7 @@ def log_end(self, event: AfterInvocationEvent) -> None:
     "AfterInvocationEvent",
     "MessageAddedEvent",
     "HookEvent",
+    "HookOrder",
     "HookProvider",
     "HookCallback",
     "HookRegistry",

strands-py/src/strands/hooks/registry.py

@@ -7,10 +7,12 @@
 via hook provider objects.
 """
 
+import bisect
 import inspect
 import logging
 from collections.abc import Awaitable, Generator
 from dataclasses import dataclass
+from itertools import groupby
 from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, runtime_checkable
 
 from ..interrupt import Interrupt, InterruptException
@@ -22,6 +24,25 @@
 logger = logging.getLogger(__name__)
 
 
+class HookOrder:
+    """Named constants for hook execution priority.
+
+    Lower values execute first. Hooks with the same order preserve registration order.
+    """
+
+    SDK_FIRST: int = -100
+    DEFAULT: int = 0
+    SDK_LAST: int = 100
+
+
+@dataclass
+class _CallbackEntry:
+    """Internal entry pairing a callback with its execution order."""
+
+    callback: "HookCallback"
+    order: float
+
+
 @dataclass
 class BaseHookEvent:
     """Base class for all hook events."""
@@ -156,12 +177,14 @@ class HookRegistry:
 
     def __init__(self) -> None:
         """Initialize an empty hook registry."""
-        self._registered_callbacks: dict[type, list[HookCallback]] = {}
+        self._registered_callbacks: dict[type, list[_CallbackEntry]] = {}
 
     def add_callback(
         self,
         event_type: type[TEvent] | list[type[TEvent]] | None,
         callback: HookCallback[TEvent],
+        *,
+        order: float = HookOrder.DEFAULT,
     ) -> None:
         """Register a callback function for a specific event type.
 
@@ -176,6 +199,7 @@ def add_callback(
             event_type: The lifecycle event type(s) this callback should handle.
                 Can be a single type, a list of types, or None to infer from type hints.
             callback: The callback function to invoke when events of this type occur.
+            order: Execution priority. Lower values execute first.
 
         Raises:
             ValueError: If event_type is not provided and cannot be inferred from
@@ -227,8 +251,9 @@ def multi_handler(event):
             if resolved_event_type.__name__ == "AgentInitializedEvent" and inspect.iscoroutinefunction(callback):
                 raise ValueError("AgentInitializedEvent can only be registered with a synchronous callback")
 
-            callbacks = self._registered_callbacks.setdefault(resolved_event_type, [])
-            callbacks.append(callback)
+            entries = self._registered_callbacks.setdefault(resolved_event_type, [])
+            entry = _CallbackEntry(callback=callback, order=order)
+            bisect.insort(entries, entry, key=lambda e: e.order)
 
     def _validate_event_type_list(self, event_types: list[type[TEvent]]) -> list[type[TEvent]]:
         """Validate that all types in a list are valid BaseHookEvent subclasses.
@@ -381,9 +406,12 @@ def has_callbacks(self) -> bool:
     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 should_reverse_callbacks=True.
-        This enables proper cleanup ordering for teardown events.
+        For normal events, callbacks are returned in order priority (lower first),
+        with registration order preserved within the same priority.
+
+        For reversed events (should_reverse_callbacks=True), order priority still
+        applies (lower first), but within the same priority group, registration
+        order is reversed.
 
         Args:
             event: The event to get callbacks for.
@@ -400,8 +428,11 @@ def get_callbacks_for(self, event: TEvent) -> Generator[HookCallback[TEvent], No
         """
         event_type = type(event)
 
-        callbacks = self._registered_callbacks.get(event_type, [])
+        entries = self._registered_callbacks.get(event_type, [])
         if event.should_reverse_callbacks:
-            yield from reversed(callbacks)
+            for _order, group in groupby(entries, key=lambda e: e.order):
+                for entry in reversed(list(group)):
+                    yield entry.callback
         else:
-            yield from callbacks
+            for entry in entries:
+                yield entry.callback

strands-py/src/strands/multiagent/base.py

@@ -13,7 +13,7 @@
 
 from .._async import run_async
 from ..agent import AgentResult
-from ..hooks.registry import HookCallback
+from ..hooks.registry import HookCallback, HookOrder
 from ..interrupt import Interrupt
 from ..types.event_loop import Metrics, Usage
 from ..types.multiagent import MultiAgentInput
@@ -255,7 +255,9 @@ def deserialize_state(self, payload: dict[str, Any]) -> None:
         """Restore orchestrator state from a session dict."""
         raise NotImplementedError
 
-    def add_hook(self, callback: HookCallback, event_type: type | list[type] | None = None) -> None:
+    def add_hook(
+        self, callback: HookCallback, event_type: type | list[type] | None = None, *, order: float = HookOrder.DEFAULT
+    ) -> None:
         """Register a hook callback with the orchestrator.
 
         Subclasses that support hooks should override this method to register
@@ -266,6 +268,7 @@ def add_hook(self, callback: HookCallback, event_type: type | list[type] | None
             event_type: The class type(s) of events this callback should handle.
                 Can be a single type, a list of types, or None to infer from
                 the callback's first parameter type hint.
+            order: Execution priority. Lower values execute first.
         """
         raise NotImplementedError(f"{type(self).__name__} must implement add_hook() to support plugins")
 

strands-py/src/strands/multiagent/graph.py

@@ -35,7 +35,7 @@
     BeforeNodeCallEvent,
     MultiAgentInitializedEvent,
 )
-from ..hooks.registry import HookCallback, HookProvider, HookRegistry
+from ..hooks.registry import HookCallback, HookOrder, HookProvider, HookRegistry
 from ..interrupt import Interrupt, _InterruptState
 from ..plugins.multiagent_plugin import MultiAgentPlugin
 from ..plugins.multiagent_registry import _MultiAgentPluginRegistry
@@ -495,16 +495,19 @@ def __init__(
 
         run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))
 
-    def add_hook(self, callback: HookCallback, event_type: type | list[type] | None = None) -> None:
+    def add_hook(
+        self, callback: HookCallback, event_type: type | list[type] | None = None, *, order: float = HookOrder.DEFAULT
+    ) -> None:
         """Register a hook callback with the graph.
 
         Args:
             callback: The callback function to invoke when events of this type occur.
             event_type: The class type(s) of events this callback should handle.
                 Can be a single type, a list of types, or None to infer from
                 the callback's first parameter type hint.
+            order: Execution priority. Lower values execute first.
         """
-        self.hooks.add_callback(event_type, callback)
+        self.hooks.add_callback(event_type, callback, order=order)
 
     def __call__(
         self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any

strands-py/src/strands/multiagent/swarm.py

@@ -35,7 +35,7 @@
     BeforeNodeCallEvent,
     MultiAgentInitializedEvent,
 )
-from ..hooks.registry import HookCallback, HookProvider, HookRegistry
+from ..hooks.registry import HookCallback, HookOrder, HookProvider, HookRegistry
 from ..interrupt import Interrupt, _InterruptState
 from ..plugins.multiagent_plugin import MultiAgentPlugin
 from ..plugins.multiagent_registry import _MultiAgentPluginRegistry
@@ -315,16 +315,19 @@ def __init__(
         self._inject_swarm_tools()
         run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))
 
-    def add_hook(self, callback: HookCallback, event_type: type | list[type] | None = None) -> None:
+    def add_hook(
+        self, callback: HookCallback, event_type: type | list[type] | None = None, *, order: float = HookOrder.DEFAULT
+    ) -> None:
         """Register a hook callback with the swarm.
 
         Args:
             callback: The callback function to invoke when events of this type occur.
             event_type: The class type(s) of events this callback should handle.
                 Can be a single type, a list of types, or None to infer from
                 the callback's first parameter type hint.
+            order: Execution priority. Lower values execute first.
         """
-        self.hooks.add_callback(event_type, callback)
+        self.hooks.add_callback(event_type, callback, order=order)
 
     def __call__(
         self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any

strands-py/tests/strands/agent/hooks/test_hook_registry.py

@@ -58,7 +58,7 @@ def test_add_callback(hook_registry, normal_event):
     hook_registry.add_callback(NormalTestEvent, callback)
 
     assert NormalTestEvent in hook_registry._registered_callbacks
-    assert callback in hook_registry._registered_callbacks[NormalTestEvent]
+    assert any(e.callback is callback for e in hook_registry._registered_callbacks[NormalTestEvent])
 
 
 def test_add_multiple_callbacks_same_event(hook_registry, normal_event):
@@ -70,8 +70,8 @@ def test_add_multiple_callbacks_same_event(hook_registry, normal_event):
     hook_registry.add_callback(NormalTestEvent, callback2)
 
     assert len(hook_registry._registered_callbacks[NormalTestEvent]) == 2
-    assert callback1 in hook_registry._registered_callbacks[NormalTestEvent]
-    assert callback2 in hook_registry._registered_callbacks[NormalTestEvent]
+    assert any(e.callback is callback1 for e in hook_registry._registered_callbacks[NormalTestEvent])
+    assert any(e.callback is callback2 for e in hook_registry._registered_callbacks[NormalTestEvent])
 
 
 def test_add_hook(hook_registry):

strands-py/tests/strands/agent/test_agent.py

@@ -2629,7 +2629,7 @@ def test_agent_add_hook_delegates_to_hooks_add_callback():
     # Spy on the hooks.add_callback method
     with unittest.mock.patch.object(agent.hooks, "add_callback") as mock_add_callback:
         agent.add_hook(callback, BeforeInvocationEvent)
-        mock_add_callback.assert_called_once_with(BeforeInvocationEvent, callback)
+        mock_add_callback.assert_called_once_with(BeforeInvocationEvent, callback, order=0)
 
 
 @pytest.mark.asyncio