Add Agent `event_stream_handler` function as alternative to `Agent.iter`

[samuelcolvin]

@dmontagu has the plan.

[dmontagu]

Inspired by the goal of better temporal support, I want to provide a way to handle streaming responses from an LLM without needing to expose the stream externally to the call to agent.run. (In particular, I want to provide a way to do stream-handling inside an activity as part of the agentic loop, without moving the entire agentic loop into the activity, so you can still have tool calls be robust etc.)

My proposal for how to do this is an API roughly along these lines (note that we'll probably change the Event types in some ways, at the least providing info about when the request index changes):

from dataclasses import dataclass
from typing import AsyncIterable

from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import AgentStreamEvent, HandleResponseEvent


@dataclass
class Deps:
    stream_destination_url: str | None


async def handle_event_stream(ctx: RunContext[Deps], event_stream: AsyncIterable[AgentStreamEvent | HandleResponseEvent]):
    with connect_to(ctx.deps.stream_destination_url) as handle_event:
        async for event in event_stream:
            handle_event(event)  # e.g., send the event to a redis stream or similar


agent = Agent(instructions='...', deps_type=Deps, stream_handler=handle_event_stream)

The idea is that when you call agent.run or iterate (via agent.iter), if you have an event_handler, we will default to using the streaming version of the APIs under the hood and send the event stream to the event handler. This will be compatible with doing the streaming in a separate process/etc. from the process controlling the agentic loop, the way you need to for temporal activities, etc.

We'll still keep the existing stream APIs (or something close to them) to allow you to do more flexible stream handling if you want to do this and run the full agentic loop in an activity (or just want to do things in memory, without temporal/etc. at all).

One other thing that I think we need to do with this is change the run_stream API to expose an event stream to end users that can be used to stream over the output of each step in the agentic loop, NOT just the final step.

[Wh1isper]

Thanks, I'm actually enjoying and using the current flow control in depth, very flexible and glad to see that there won't be any particularly big breaking changes.