feat(hooks): add BeforeStreamChunkEvent for true stream chunk interception

Add BeforeStreamChunkEvent hook that fires BEFORE each stream chunk is
processed, enabling true interception capabilities:

- Monitor streaming progress in real-time
- Modify chunk content before processing (affects final message)
- Skip chunks entirely by setting skip=True (excluded from final message)
- Implement content transformation (e.g., redaction, translation)

Implementation details:
- Added ChunkInterceptor callback type to streaming.py
- Modified process_stream() to invoke interceptor BEFORE processing
- Modified stream_messages() to accept chunk_interceptor parameter
- event_loop.py creates interceptor that invokes BeforeStreamChunkEvent

When skip=True:
- The chunk is not processed at all
- No events (ModelStreamChunkEvent, TextStreamEvent) are yielded
- The chunk does not contribute to the final message

When chunk is modified:
- The modified chunk is used for all downstream processing
- TextStreamEvent will contain the modified text
- The final message will contain the modified content

Author: fede-kamel

src/strands/event_loop/event_loop.py

@@ -15,7 +15,7 @@
 
 from opentelemetry import trace as trace_api
 
-from ..hooks import AfterModelCallEvent, BeforeModelCallEvent, MessageAddedEvent
+from ..hooks import AfterModelCallEvent, BeforeModelCallEvent, BeforeStreamChunkEvent, MessageAddedEvent
 from ..telemetry.metrics import Trace
 from ..telemetry.tracer import Tracer, get_tracer
 from ..tools._validator import validate_and_prepare_tools
@@ -39,7 +39,7 @@
     MaxTokensReachedException,
     StructuredOutputException,
 )
-from ..types.streaming import StopReason
+from ..types.streaming import StopReason, StreamEvent
 from ..types.tools import ToolResult, ToolUse
 from ._recover_message_on_max_tokens_reached import recover_message_on_max_tokens_reached
 from ._retry import ModelRetryStrategy
@@ -327,6 +327,18 @@ async def _handle_model_execution(
                 tool_specs = [tool_spec] if tool_spec else []
             else:
                 tool_specs = agent.tool_registry.get_all_tool_specs()
+
+            # Create chunk interceptor that invokes BeforeStreamChunkEvent hook
+            async def chunk_interceptor(chunk: StreamEvent) -> tuple[StreamEvent, bool]:
+                """Intercept chunks and invoke BeforeStreamChunkEvent hook."""
+                stream_chunk_event = BeforeStreamChunkEvent(
+                    agent=agent,
+                    chunk=chunk,
+                    invocation_state=invocation_state,
+                )
+                await agent.hooks.invoke_callbacks_async(stream_chunk_event)
+                return stream_chunk_event.chunk, stream_chunk_event.skip
+
             try:
                 async for event in stream_messages(
                     agent.model,
@@ -337,6 +349,7 @@ async def _handle_model_execution(
                     tool_choice=structured_output_context.tool_choice,
                     invocation_state=invocation_state,
                     cancel_signal=agent._cancel_signal,
+                    chunk_interceptor=chunk_interceptor,
                 ):
                     yield event
 

src/strands/event_loop/streaming.py

@@ -5,7 +5,7 @@
 import threading
 import time
 import warnings
-from collections.abc import AsyncGenerator, AsyncIterable
+from collections.abc import AsyncGenerator, AsyncIterable, Awaitable, Callable
 from typing import Any
 
 from ..models.model import Model
@@ -42,6 +42,10 @@
 
 logger = logging.getLogger(__name__)
 
+# Type for chunk interceptor callback
+# Takes a chunk and returns (modified_chunk, skip) where skip=True means don't process this chunk
+ChunkInterceptor = Callable[[StreamEvent], Awaitable[tuple[StreamEvent, bool]]]
+
 
 def _normalize_messages(messages: Messages) -> Messages:
     """Remove or replace blank text in message content.
@@ -387,13 +391,17 @@ async def process_stream(
     chunks: AsyncIterable[StreamEvent],
     start_time: float | None = None,
     cancel_signal: threading.Event | None = None,
+    chunk_interceptor: ChunkInterceptor | None = None,
 ) -> AsyncGenerator[TypedEvent, None]:
     """Processes the response stream from the API, constructing the final message and extracting usage metrics.
 
     Args:
         chunks: The chunks of the response stream from the model.
         start_time: Time when the model request is initiated
         cancel_signal: Optional threading.Event to check for cancellation during streaming.
+        chunk_interceptor: Optional callback to intercept and modify chunks before processing.
+            The callback receives a chunk and returns (modified_chunk, skip). If skip is True,
+            the chunk is not processed or yielded.
 
     Yields:
         The reason for stopping, the constructed message, and the usage metrics.
@@ -427,6 +435,12 @@ async def process_stream(
             )
             return
 
+        # Invoke chunk interceptor BEFORE processing if provided
+        if chunk_interceptor is not None:
+            chunk, skip = await chunk_interceptor(chunk)
+            if skip:
+                continue
+
         # Track first byte time when we get first content
         if first_byte_time is None and ("contentBlockDelta" in chunk or "contentBlockStart" in chunk):
             first_byte_time = time.time()
@@ -464,6 +478,7 @@ async def stream_messages(
     system_prompt_content: list[SystemContentBlock] | None = None,
     invocation_state: dict[str, Any] | None = None,
     cancel_signal: threading.Event | None = None,
+    chunk_interceptor: ChunkInterceptor | None = None,
     **kwargs: Any,
 ) -> AsyncGenerator[TypedEvent, None]:
     """Streams messages to the model and processes the response.
@@ -478,6 +493,9 @@ async def stream_messages(
             system prompt data.
         invocation_state: Caller-provided state/context that was passed to the agent when it was invoked.
         cancel_signal: Optional threading.Event to check for cancellation during streaming.
+        chunk_interceptor: Optional callback to intercept and modify chunks before processing.
+            The callback receives a chunk and returns (modified_chunk, skip). If skip is True,
+            the chunk is not processed or yielded.
         **kwargs: Additional keyword arguments for future extensibility.
 
     Yields:
@@ -497,5 +515,5 @@ async def stream_messages(
         invocation_state=invocation_state,
     )
 
-    async for event in process_stream(chunks, start_time, cancel_signal):
+    async for event in process_stream(chunks, start_time, cancel_signal, chunk_interceptor):
         yield event

src/strands/hooks/__init__.py

@@ -41,6 +41,7 @@ def log_end(self, event: AfterInvocationEvent) -> None:
     BeforeModelCallEvent,
     BeforeMultiAgentInvocationEvent,
     BeforeNodeCallEvent,
+    BeforeStreamChunkEvent,
     BeforeToolCallEvent,
     MessageAddedEvent,
     MultiAgentInitializedEvent,
@@ -50,6 +51,7 @@ def log_end(self, event: AfterInvocationEvent) -> None:
 __all__ = [
     "AgentInitializedEvent",
     "BeforeInvocationEvent",
+    "BeforeStreamChunkEvent",
     "BeforeToolCallEvent",
     "AfterToolCallEvent",
     "BeforeModelCallEvent",

src/strands/hooks/events.py

@@ -15,7 +15,7 @@
 from ..types.agent import AgentInput
 from ..types.content import Message, Messages
 from ..types.interrupt import _Interruptible
-from ..types.streaming import StopReason
+from ..types.streaming import StopReason, StreamEvent
 from ..types.tools import AgentTool, ToolResult, ToolUse
 from .registry import BaseHookEvent, HookEvent
 
@@ -303,6 +303,48 @@ def should_reverse_callbacks(self) -> bool:
         return True
 
 
+@dataclass
+class BeforeStreamChunkEvent(HookEvent):
+    """Event triggered before each stream chunk is processed.
+
+    This event is fired for each chunk received from the model BEFORE the chunk
+    is processed for message building or yielded as stream events. Hook providers
+    can use this event to:
+
+    - Monitor streaming progress in real-time
+    - Modify chunk content before processing (affects final message and all events)
+    - Filter/skip chunks entirely by setting skip=True
+    - Implement content transformation (e.g., redaction, translation)
+
+    When skip=True:
+        - The chunk is not processed at all
+        - No events (ModelStreamChunkEvent, TextStreamEvent, etc.) are yielded
+        - The chunk does not contribute to the final message
+
+    When chunk is modified:
+        - The modified chunk is used for all downstream processing
+        - TextStreamEvent will contain the modified text
+        - The final message will contain the modified content
+
+    Performance Note:
+        This event fires for every stream chunk, so callbacks should execute
+        quickly to avoid impacting streaming latency.
+
+    Attributes:
+        chunk: The raw stream event from the model. Can be modified by hooks
+            to transform content before processing.
+        skip: When True, the chunk is skipped entirely (not processed or yielded).
+        invocation_state: State passed through agent invocation.
+    """
+
+    chunk: StreamEvent
+    invocation_state: dict[str, Any] = field(default_factory=dict)
+    skip: bool = False
+
+    def _can_write(self, name: str) -> bool:
+        return name in ["chunk", "skip"]
+
+
 # Multiagent hook events start here
 @dataclass
 class MultiAgentInitializedEvent(BaseHookEvent):

tests/strands/agent/hooks/test_events.py

@@ -10,6 +10,7 @@
     AgentInitializedEvent,
     BeforeInvocationEvent,
     BeforeModelCallEvent,
+    BeforeStreamChunkEvent,
     BeforeToolCallEvent,
     MessageAddedEvent,
 )
@@ -260,3 +261,58 @@ def test_after_invocation_event_resume_accepts_various_input_types(agent):
     # None to stop
     event.resume = None
     assert event.resume is None
+
+
+@pytest.fixture
+def before_stream_chunk_event(agent):
+    chunk = {"contentBlockDelta": {"delta": {"text": "Hello"}}}
+    return BeforeStreamChunkEvent(
+        agent=agent,
+        chunk=chunk,
+        invocation_state={"test": "state"},
+    )
+
+
+def test_before_stream_chunk_event_should_not_reverse_callbacks(before_stream_chunk_event):
+    """Test that BeforeStreamChunkEvent does not reverse callbacks."""
+    assert before_stream_chunk_event.should_reverse_callbacks is False
+
+
+def test_before_stream_chunk_event_can_write_chunk(before_stream_chunk_event):
+    """Test that BeforeStreamChunkEvent.chunk is writable."""
+    new_chunk = {"contentBlockDelta": {"delta": {"text": "Modified"}}}
+    before_stream_chunk_event.chunk = new_chunk
+    assert before_stream_chunk_event.chunk == new_chunk
+
+
+def test_before_stream_chunk_event_can_write_skip(before_stream_chunk_event):
+    """Test that BeforeStreamChunkEvent.skip is writable."""
+    assert before_stream_chunk_event.skip is False
+    before_stream_chunk_event.skip = True
+    assert before_stream_chunk_event.skip is True
+
+
+def test_before_stream_chunk_event_cannot_write_agent(before_stream_chunk_event):
+    """Test that BeforeStreamChunkEvent.agent is not writable."""
+    with pytest.raises(AttributeError, match="Property agent is not writable"):
+        before_stream_chunk_event.agent = Mock()
+
+
+def test_before_stream_chunk_event_cannot_write_invocation_state(before_stream_chunk_event):
+    """Test that BeforeStreamChunkEvent.invocation_state is not writable."""
+    with pytest.raises(AttributeError, match="Property invocation_state is not writable"):
+        before_stream_chunk_event.invocation_state = {}
+
+
+def test_before_stream_chunk_event_skip_defaults_to_false(agent):
+    """Test that BeforeStreamChunkEvent.skip defaults to False."""
+    chunk = {"contentBlockDelta": {"delta": {"text": "Hello"}}}
+    event = BeforeStreamChunkEvent(agent=agent, chunk=chunk)
+    assert event.skip is False
+
+
+def test_before_stream_chunk_event_invocation_state_defaults_to_empty(agent):
+    """Test that BeforeStreamChunkEvent.invocation_state defaults to empty dict."""
+    chunk = {"contentBlockDelta": {"delta": {"text": "Hello"}}}
+    event = BeforeStreamChunkEvent(agent=agent, chunk=chunk)
+    assert event.invocation_state == {}