"j&SSKJr SSKrSSKJrJrJr SSKJr SSK J r J r J r SSK r SSKJrJr SSKJr SSKJr SS KJrJr S S KJr \ (aSS KJrJr \"S S\\55rg)) annotationsN) AsyncIterable AsyncIterator Coroutine) dataclass) TYPE_CHECKINGAnycast)MemoryObjectReceiveStreamMemoryObjectSendStream)_utils)AgentStreamEvent) AgentDepsT RunContext)AbstractCapability)EventStreamHandlerEventStreamProcessorcb\rSrSr%SrS\S'S Sjr\S Sj5r\ S Sj5r Sr g ) ProcessEventStreamuA capability that forwards the agent's event stream to a user-provided async handler. The handler receives the stream of [`AgentStreamEvent`][pydantic_ai.messages.AgentStreamEvent]s emitted during classic model streaming and tool execution, or the shared and realtime-only events emitted by a realtime session. Two forms are supported: - An [`EventStreamHandler`][pydantic_ai.agent.EventStreamHandler] — an `async def` returning `None`. Events are forwarded to the handler while also being passed through unchanged to the rest of the capability chain, so multiple handlers (and the top-level `event_stream_handler` argument) can all see the same stream without changing each other's view. A handler that returns early stops receiving events but does not affect downstream consumers; a handler that raises propagates the exception to the rest of the run. Events are delivered synchronously, so a slow handler back-pressures the rest of the stream. - An `EventStreamProcessor` — an async generator yielding [`AgentStreamEvent`][pydantic_ai.messages.AgentStreamEvent]s. The events it yields replace the inner stream for downstream wrappers and consumers, so it can modify, drop, or add events. This replacement is global, not a private view for event-stream handlers: the run has one event stream and a processor shapes all of it. Dropping or rewriting a [`PartDeltaEvent`][pydantic_ai.messages.PartDeltaEvent] therefore also changes what [`stream_text()`][pydantic_ai.result.StreamedRunResult.stream_text] yields to a `run_stream()` caller. Some events are also control signals: [`FinalResultEvent`][pydantic_ai.messages.FinalResultEvent] is what tells [`agent.run_stream()`][pydantic_ai.agent.AbstractAgent.run_stream] that the final output has started, so dropping it makes `run_stream()` wait for the whole model response before handing back the result instead of streaming it. Filter deliberately. None of this changes the run's output: the [`ModelResponse`][pydantic_ai.messages.ModelResponse] is accumulated from the raw model stream before a processor sees the events, so [`stream_output()`][pydantic_ai.result.StreamedRunResult.stream_output] and the final validated output are unaffected (dropping events can only change when a partial snapshot is emitted, not its content). Use the observer form if you only want to watch events. In a realtime session, this is likewise only a consumer-facing view. Transforming or dropping events does not affect session history or tool execution. When this capability is registered, `agent.run()` and [`AgentRun.next()`][pydantic_ai.run.AgentRun.next] automatically enable streaming so the handler fires without requiring an explicit `event_stream_handler` argument. The handler sees the same events however the run is driven, including under [`agent.iter()`][pydantic_ai.agent.Agent.iter] and when you stream a node yourself with `node.stream()`. !!! note "Durable execution" Under the durable-execution capabilities ([`TemporalDurability`][pydantic_ai.durable_exec.temporal.TemporalDurability], [`DBOSDurability`][pydantic_ai.durable_exec.dbos.DBOSDurability], [`PrefectDurability`][pydantic_ai.durable_exec.prefect.PrefectDurability]), this capability's handler always runs in workflow or flow code and must be deterministic because it re-runs on workflow replay. Tool-call and final-output events arrive live; model events are the real captured events replayed after each model-request activity, step, or task completes. For handler I/O that must run exactly once inside a durable boundary, pass `event_stream_handler=` to the durability capability instead. zIEventStreamHandlerFunc[AgentDepsT] | EventStreamProcessorFunc[AgentDepsT]handlerc^^ ^ ^ # URTU5n[U[5(aUShvN nU7v M[SU5R 5 [SUR5m [ R "5unm SUU U 4Sjjn[R"U"55nSn[U5m UIShvN Sn S U 4Sjjn [R"U "55nU (a[R"X4[RS9IShvN UR5(apUR5(dUR5cSn OG[R "U5IShvN [R""T 5IShvN UIShvN UIShvN nU (aUR'U5IShvN U7v GMGN gGNNNkNNNF!UIShvN f=fNM![$a O2f=fN>![ R([ R*4a Sn Ndf=fSSS5IShvN Os!,IShvN (df  O\=f![,aL [R "U/[/SU45Q76IShvN [R""T 5IShvN ef=fUIShvN g7f) NzCoroutine[Any, Any, None]z"EventStreamHandlerFunc[AgentDepsT]c># TIShvN T"TT5IShvN SSS5IShvN gN&NN !,IShvN (df  g=f7fN)ctxobserverreceive_streamsj/home/mande/repo/quber/.venv/lib/python3.13/site-packages/pydantic_ai/capabilities/process_event_stream.py run_handler=ProcessEventStream.wrap_run_event_stream..run_handlerrs2%~sN333&~~3&~~~sFA1A 737 A5A7AAA A ATc6># [T5IShvN $N7fr)anext)stream_iteratorsr pull_next;ProcessEventStream.wrap_run_event_stream..pull_nexts!&!7777s ) return_whenF)returnNone)r)r)r isinstancerr closeanyiocreate_memory_object_streamasyncio create_taskaiterwaitFIRST_COMPLETEDdone cancelled exceptionr cancel_and_drainaclose_if_supportedStopAsyncIterationsendBrokenResourceErrorClosedResourceError BaseExceptionfilter)selfrstreamprobeevent send_streamr! handler_task next_task handler_aliver&rrr%s ` @@@r wrap_run_event_stream(ProcessEventStream.wrap_run_event_streamZsb S&) e] + +$ e (%0668risX"CC!++ R14(  e+J7e erN