Stream to a terminal or UI

You can watch a request happen in two ways: register a dispatcher for the events you care about, or consume the stream yourself when you need to cancel or forward it.

Use callbacks for display

Subclass AgentEventDispatcher and override only the events you want to render. Bind it through create_agent(..., event_dispatcher=...); client.run() still returns the final answer.

class ConsoleEvents(AgentEventDispatcher):
    async def on_reasoning_delta_event(self, event: AgentEvent) -> None:
        print(f"Reasoning: {event.delta}")

    async def on_tool_started_event(self, event: AgentEvent) -> None:
        call = event.tool_calls[0]
        print(f"Tool: {call.name} {call.arguments}")

    async def on_tool_completed_event(self, event: AgentEvent) -> None:
        assert event.message is not None
        print(f"Result: {event.message.content}")

    async def on_text_delta_event(self, event: AgentEvent) -> None:
        print(event.delta)

Two rules keep the display correct:

• delta is incremental, while message on a completion event is complete. Appending both to the same buffer duplicates the answer.

• Tool arguments can arrive as incomplete fragments. Only complete tool calls are executed, and the runtime does that — your UI never runs them.

TOOL_STARTED can describe a batch: iterate event.tool_calls instead of assuming there is one call. Completion and failure events identify the finished call; correlate them by ID, since parallel calls can finish out of order.

Consume the stream yourself for control

from contextlib import aclosing

async with aclosing(client.stream("Inspect the project")) as events:
    async for event in events:
        await send_to_ui(event)

For a simple terminal, match only the fields associated with each event:

from zett_agent.events import AgentEventType

async with aclosing(client.stream("What is 20 plus 22?")) as events:
    async for event in events:
        if event.type == AgentEventType.TEXT_DELTA:
            print(event.delta, end="", flush=True)
        elif event.type == AgentEventType.TOOL_STARTED:
            for call in event.tool_calls:
                print(f"\nCalling {call.name} ({call.id})")
        elif event.type == AgentEventType.RUN_COMPLETED:
            final_message = event.message

client must already have a model and any tools it needs. The final message is the authoritative answer; deltas are for presentation. A provider may emit an answer only in its final response, so a UI should reconcile the final content instead of assuming text deltas always exist. If a dispatcher is bound, it has already handled each event before you receive it here.

Use aclosing even when you break out of the loop early: it stops the request and waits for cleanup. Callbacks are awaited in order, so slow callbacks apply backpressure; hand work to a bounded application queue instead of spawning unlimited tasks.

What you will see

Event	Meaning
TEXT_DELTA	A fragment of the answer text.
REASONING_DELTA	A fragment of the model’s reasoning, when the model exposes it.
TOOL_STARTED / TOOL_COMPLETED / TOOL_FAILED	A requested tool call and how it finished.
TOOL_CALL_DELTA	Partial tool arguments, for progress display only.
COMPACTION_*	Older context is being summarized before the next model step.
MODEL_COMPLETED	One complete model response, including usage.
MODEL_STOPPED	The model did not finish normally: it was refused, cut off, interrupted, or paused for good. Follows MODEL_COMPLETED and carries stop; see Know why the model stopped.
RUN_COMPLETED	The request finished successfully.
CUSTOM	An event your own extension emitted.

Provider-hosted tools add SERVER_TOOL_* events; steering and internal continuations add their own pairs. The full list is on AgentEventType.

Track model usage

Read usage from event.response.usage on MODEL_COMPLETED. It describes one model step, not the whole user request. A tool round trip often produces two or more such events, so sum their counters to measure the primary loop:

tokens = 0
async with aclosing(client.stream("Explain the result.")) as events:
    async for event in events:
        if event.type == AgentEventType.MODEL_COMPLETED and event.response is not None:
            tokens += event.response.usage.total_tokens

total_tokens is input plus output; reasoning and cache counters are already included and must not be added again. Missing usage is not proof of zero cost. Separate compaction or delegated models can incur additional usage outside these primary model events. See Observe usage and timing for recorded timings and Connect a model provider for cache reporting.

Failures are exceptions

RUN_COMPLETED means the request ran to its end, which is not the same as the model finishing its answer: a refusal ends a request and still completes it. Check MODEL_STOPPED or message.stop for that; see Know why the model stopped. Provider errors, extension errors, and cancellation propagate as exceptions — they are never converted into a synthetic completion event. Wrap the stream when the UI needs to show a failure, and see Cancel and reuse an agent for cleanup and reuse.

Next: keep the conversation in Keep and restore conversations, or read Three kinds of events for the difference between UI events, internal notifications, and external input.