[FEATURE] Async Iterator - Typed Events

[pgrayy]

Problem Statement

Currently, the events we pass to the callback handlers and also yield from our async iterator are not strongly typed. This makes it challenging for users to build conditions off the events for performing actions (e.g. printing model text chunks to screen). We should formalize the structure and strongly types these event payloads.

Proposed Solution

Solution is to implement interfaces for the Agent event payloads. Could be a dataclass, TypedDict, etc.

Use Case

To configure reliable conditions on the events for performing actions real-time (e.g., printing model text to screen in a chat application of Strands).

Alternatives Solutions

No response

Additional Context

No response

[pgrayy]

Related issue: #276

[zastrowm]

Note that #250 called out:

I would like the ability to filter events for easier consumption
Add a parameter to agent to filter events (all, messages, blocks for example)
With this in place, code will be simpler

if "delta" in event:

or

if "message" in event:

as it will yield only the wanted events

[wzxxing]

Edit: please ignore my comment, I realized typed dict can be a good option too.

---

Want to +1 on to option of using dataclass, the reason being the better support in the match expression.

For TypedDict, the match is not so powerful, perhaps we can have a tagged Union like

class Event(TypedDict, total=False):
    tool_call: ToolCallEvent # typed dict
    tool_result: ToolResultEvent # typed dict

def event_processor(event: Event):
    match event:
        case {"tool_call": tool_call}:
            ...
        case {"tool_result": tool_result}:
            ...

It's simple but that's it.

However, with dataclasses, we can achieve more:

1. instance checking
2. matching the nested fields typed dict can match nested field too.

The following example, the PartStartEvent is generic in its part field, it can be thinking or text, different purposes and useful differentiation.

    match event:
        case PartStartEvent(index, TextPart() as part):
            return TextChunk(f"text-message-{index}", part.content)
        case PartStartEvent(index, ThinkingPart() as part):
            return ThinkingChunk(f"thinking-{index}", part.content)

The benefit is that typeddict is relatively easy to write, users do not have to import the type definition to match. Constructing can be easily done by the dictionary literals, no need to use the constructor. But as a framework user, they probably never need to manually construct the events. So I think it is fair to trade the complexity of constructing events inside of the framework for users convenience of matching dataclass.

[zastrowm]

Related to #148