feat(hooks): add BeforeStreamChunkEvent for stream chunk interception

[fede-kamel]

Motivation

There's currently no way to intercept stream chunks before they're processed. Hooks fire after processing, so modifications don't affect the TextStreamEvent or final message content.

The current hook lifecycle has a gap during streaming:

BeforeModelCallEvent
    ↓
[MODEL STREAMS CHUNKS]  ← no hook here currently
    ↓
AfterModelCallEvent

BeforeModelCallEvent fires before chunks exist. AfterModelCallEvent fires after chunks have already been streamed to the user. Neither allows chunk interception.

This adds BeforeStreamChunkEvent that fires before each chunk is processed, allowing:

• Chunk modification (affects downstream processing and final message)
• Chunk filtering via skip=True (excludes from all events and final message)
• Access to invocation_state for context

Resolves #1760

Use Cases

• Content redaction: Modify sensitive data before it reaches the user
• Filtering: Skip certain chunks entirely (skip=True)
• Real-time monitoring: Observe streaming without modifying
• Content transformation: Translation, formatting, etc.

Public API Changes

New event: BeforeStreamChunkEvent

from strands.hooks import BeforeStreamChunkEvent

async def redact_chunks(event: BeforeStreamChunkEvent):
    if "contentBlockDelta" in event.chunk:
        delta = event.chunk["contentBlockDelta"]["delta"]
        if "text" in delta:
            # modify the chunk - affects TextStreamEvent and final message
            event.chunk = {"contentBlockDelta": {"delta": {"text": "[REDACTED]"}}}
            # or skip entirely
            # event.skip = True

agent.hooks.add_callback(BeforeStreamChunkEvent, redact_chunks)

Properties:

• chunk (writable) - the raw stream chunk
• skip (writable) - set True to exclude chunk from processing
• invocation_state (read-only) - context dict from invocation

Reproduction and Demonstration

Issue reproduction (main branch): https://gist.github.com/fede-kamel/9d93429a3d6d7676906e41eb60890d46

Working demonstration (feature branch): https://gist.github.com/fede-kamel/a6a78d8d35eb9707d80c2327cd17b462

Type of Change

New feature

Testing

• I ran hatch run prepare (lint, format, tests all pass)

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[fede-kamel]

@mkmeral 👍

[fede-kamel]

Integration Test Results

Rebased on latest main and tested BeforeStreamChunkEvent against multiple OpenAI models:

Model	Event Fires	Chunk Modification	Skip Chunks	invocation_state	Result
gpt-4o-mini	7 chunks	All redacted	Empty content	14 states	PASS
gpt-4o	6 chunks	All redacted	Empty content	14 states	PASS
gpt-5.2	6 chunks	All redacted	Empty content	12 states	PASS

What was tested

1. Event fires — BeforeStreamChunkEvent callback receives all chunk types (messageStart, contentBlockStart, contentBlockDelta, contentBlockStop, messageStop, metadata)
2. Chunk modification — Replacing text deltas with [REDACTED] propagates to both TextStreamEvent and final message content (no original text leaks)
3. Skip chunks — Setting skip=True on content deltas prevents processing: no TextStreamEvent yielded, final message has empty content
4. invocation_state — Custom state dict passed at invocation is accessible in every BeforeStreamChunkEvent callback

Unit tests

All 108 unit tests pass after rebase (hooks, event_loop, streaming).

[agent-of-mkmeral]

Hey @fede-kamel — solid work here, really well-iterated PR. The use cases (content redaction, monitoring, chunk filtering) are all valid.

However, we'd like to defer this for now due to the recently merged state machine design (0005) from @pgrayy. The writable chunk + skip pattern is essentially middleware behavior (intercept → transform → control flow), which 0005 explicitly assigns to the middleware layer rather than hooks/plugins. Chunk interception would be a natural first middleware use case.

Also worth noting: TS currently only has read-only ModelStreamUpdateEvent — shipping writable interception in Python first creates a parity gap that'd be cleaner to solve via the middleware approach in both SDKs.

@pgrayy — would love your take on whether this fits middleware or hooks in the 0005 design.

---

🤖 AI agent response on behalf of @mkmeral. Strands Agents. Feedback welcome!

[fede-kamel]

Hey @mkmeral @pgrayy — thanks for the thorough feedback.

Totally understand. I went through the 0005 design and I can see how chunk interception (writable chunk + skip) maps to the middleware layer rather than hooks — it's intercept → transform → control flow, which 0005 explicitly assigns to middleware. Makes sense not to ship this through the plugin layer if it'll need to migrate later.

Happy to defer this PR. We can keep it as a reference alongside the linked issue (#1762), since the use case (content redaction, chunk filtering, stream monitoring) is still genuine. Once the middleware layer lands, we can revisit implementing this through that path.

Is there a rough timeline for when middleware support might be available? Just so we know when to circle back.

[fede-kamel]

Rebased & validated against real model

Rebased on latest main (resolved conflicts with the end_on_exit=False span fix and callable hooks tests — both are preserved).

Live validation with OpenAI gpt-4o-mini

Ran all three hook capabilities against a real streaming provider:

1. Chunk capture — intercepts every chunk in the stream:

Chunks seen: 12
Chunk types: ['contentBlockDelta', 'contentBlockStart', 'contentBlockStop',
              'messageStart', 'messageStop', 'metadata']

2. Content redaction — modifies text deltas before processing:

Streamed text events: ['[REDACTED]', '[REDACTED]', '[REDACTED]', ...]
Final message text  : [REDACTED][REDACTED][REDACTED]...

The original model output never reaches the consumer — both TextStreamEvent and the final message contain only the redacted content.

3. Chunk skipping — suppresses content deltas entirely:

Text events received: 0
Final content       : []

Setting skip=True prevents the chunk from being processed at all — no events yielded, no contribution to the final message.

All three scenarios confirmed working end-to-end. Test script added in examples/test_stream_chunk_hook.py.