iterative stream messages #213

pgrayy · 2025-06-12T21:04:39Z

Description

We are currently working on support for an iterative async stream method on the agent class (#83). As part of this work, we need to yield underlying events of the model stream. This PR thus converts the stream_messages function to a generator.

Related Issues

#83

Type of Change

Testing

hatch fmt --linter
hatch fmt --formatter
hatch test --all
Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli
hatch run test-lint

Checklist

I have read the CONTRIBUTING document
I have added 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
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.

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

pgrayy · 2025-06-12T21:08:06Z

src/strands/event_loop/event_loop.py

+                    callback_handler(**inputs)
+            else:
+                stop_reason, message, usage, metrics = event["stop"]
+                kwargs.setdefault("request_state", {})


I will progressively move the callback_handler invocations upward in the call stack as we convert the event loop to an event generator.

pgrayy · 2025-06-12T21:09:32Z

src/strands/event_loop/event_loop.py

+            for event in stream_messages(model, system_prompt, messages, tool_config):
+                if "callback" in event:
+                    inputs = {**event["callback"], **(kwargs if "delta" in event else {})}
+                    callback_handler(**inputs)


Yes this is a bit messy but I am trying to maintain backwards compatibility with these changes. Ultimately, we should formalize and strongly type callback payloads.

Note, this callback invocation was copied out of stream_messages.

Is the strongly typed callback payloads already tracked? If not, can you add it to the async iterator stuff (we can always break it out later)

pgrayy · 2025-06-12T21:10:55Z

src/strands/event_loop/streaming.py

        event: Delta event.
        state: The current state of message processing.
-        callback_handler: Callback for processing events as they happen.
-        **kwargs: Additional keyword arguments to pass to the callback handler.


Since we are now yielding events from stream_messages, we no longer need to pass around callback_handler and kwargs. We can invoke the callback_handler in the event loop where we iterate over stream_messages.

pgrayy · 2025-06-12T21:12:32Z

src/strands/event_loop/streaming.py


        state["current_tool_use"]["input"] += delta_content["toolUse"]["input"]
-        callback_handler(delta=delta_content, current_tool_use=state["current_tool_use"], **kwargs)
+        callback_event["callback"] = {"delta": delta_content, "current_tool_use": state["current_tool_use"]}


This is a temporary format to maintain backwards compatibility. We are planning on formalizing and strongly typing the construction of our agent events for consistency and ease of use.

pgrayy · 2025-06-12T21:14:24Z

src/strands/event_loop/streaming.py

        elif "contentBlockDelta" in chunk:
-            state = handle_content_block_delta(chunk["contentBlockDelta"], state, callback_handler, **kwargs)
+            state, callback_event = handle_content_block_delta(chunk["contentBlockDelta"], state)
+            yield callback_event


Again, keeping for backwards compatibility, but really we should consider getting rid of this extra yield. It is redundant since we have the yield chunk at the start of each loop.

Can you clarify and/or elaborate on what the two events are? Are they truly duplciates?

pgrayy · 2025-06-12T21:14:47Z

tests-integ/test_mcp_client.py

-    condition=os.environ.get("GITHUB_ACTIONS") == 'true',
-    reason="streamable transport is failing in GitHub actions, debugging if linux compatibility issue"
+    condition=os.environ.get("GITHUB_ACTIONS") == "true",
+    reason="streamable transport is failing in GitHub actions, debugging if linux compatibility issue",


Auto linted.

zastrowm · 2025-06-13T18:52:18Z

src/strands/event_loop/event_loop.py

+            for event in stream_messages(model, system_prompt, messages, tool_config):
+                if "callback" in event:
+                    inputs = {**event["callback"], **(kwargs if "delta" in event else {})}
+                    callback_handler(**inputs)


Is the strongly typed callback payloads already tracked? If not, can you add it to the async iterator stuff (we can always break it out later)

zastrowm · 2025-06-13T18:57:04Z

src/strands/event_loop/event_loop.py

-            )
+            for event in stream_messages(model, system_prompt, messages, tool_config):
+                if "callback" in event:
+                    inputs = {**event["callback"], **(kwargs if "delta" in event else {})}


Can we add a comment indicating what's going on/what the change is.

I believe it's effectively:

In the migration to async-iterator, we converted all events that were previously passed to callback into yielded events that have the callback key, with the properties that we emitted which we now have to unwrap and combine with kwargs for backwards compatibility

Also, this is clever, so props for that

zastrowm · 2025-06-13T18:58:46Z

src/strands/event_loop/event_loop.py

+                    inputs = {**event["callback"], **(kwargs if "delta" in event else {})}
+                    callback_handler(**inputs)
+            else:
+                stop_reason, message, usage, metrics = event["stop"]


So stop right now is not an event that callback gets passed. Given that we don't need backwards compability here, what would you think about turning this into a typed event?

E.g:

class StopStreamEvent(dict):

could also be a follow-up PR

zastrowm · 2025-06-13T19:00:44Z

src/strands/event_loop/streaming.py

        elif "contentBlockDelta" in chunk:
-            state = handle_content_block_delta(chunk["contentBlockDelta"], state, callback_handler, **kwargs)
+            state, callback_event = handle_content_block_delta(chunk["contentBlockDelta"], state)
+            yield callback_event


Can you clarify and/or elaborate on what the two events are? Are they truly duplciates?

zastrowm · 2025-06-13T19:03:07Z

src/strands/event_loop/event_loop.py

I presume that test_event_loop has enough UT coverage verifying the callbacks of events are as expected? At which point we have a lot of confidence that we maintained backwards compatibility.

* build(a2a): add a2a deps and mitigate otel conflict

pgrayy commented Jun 12, 2025

View reviewed changes

pgrayy added the approved-for-integ-test Indicator that a pull request has been approved to run integration tests label Jun 12, 2025

pgrayy mentioned this pull request Jun 12, 2025

[FEATURE] Implement native async iterator support #83

Closed

zastrowm assigned pgrayy Jun 13, 2025

zastrowm self-requested a review June 13, 2025 18:45

zastrowm previously approved these changes Jun 13, 2025

View reviewed changes

pgrayy dismissed zastrowm’s stale review via d397649 June 18, 2025 12:42

pgrayy temporarily deployed to auto-approve June 18, 2025 12:43 — with GitHub Actions Inactive

pgrayy had a problem deploying to auto-approve June 18, 2025 12:50 — with GitHub Actions Failure

pgrayy and others added 5 commits June 18, 2025 12:51

iterative stream messages

5d21ebb

tests - event loop - callback

a5e9f40

event loop - clarifying comment about handler args

9a70470

chore: allow custom tracer provider in Agent (strands-agents#207)

7725624

build(a2a): add a2a deps and mitigate otel conflict (strands-agents#232)

c382891

* build(a2a): add a2a deps and mitigate otel conflict

pgrayy force-pushed the async branch from 4e756b7 to c382891 Compare June 18, 2025 12:53

pgrayy had a problem deploying to auto-approve June 18, 2025 12:53 — with GitHub Actions Failure

pgrayy closed this Jun 18, 2025

pgrayy deleted the async branch July 14, 2025 13:18

iterative stream messages #213

iterative stream messages #213

Uh oh!

Conversation

pgrayy commented Jun 12, 2025

Description

Related Issues

Type of Change

Testing

Checklist

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pgrayy Jun 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

pgrayy Jun 12, 2025 •

edited

Loading