AgentEventDispatcher#

Distinguish UI streams, extension notifications, and external replies. See the guide.

class zett_agent.dispatcher.AgentEventDispatcher[source]#

Dispatch streamed events to optional asynchronous callbacks.

Override only the callbacks your UI or application needs. Each callback receives the original event, including its relevant message, tool calls, or custom payload. Awaiting dispatch preserves stream order and applies backpressure; callback failures and cancellation propagate to the caller. This class does not change the Agent lifecycle or consume its stream.

Callbacks must use async def and run on the Agent’s event loop, including when called through SyncAgent. No subclass methods are wrapped or moved to worker threads. Offload blocking I/O explicitly; GUI callbacks must schedule visual updates on their UI thread.

Examples

Usage:

from contextlib import aclosing

class ConsoleEvents(AgentEventDispatcher):
    async def on_text_delta_event(self, event: AgentEvent) -> None:
        print(event.delta, end="", flush=True)

    async def on_custom_event(self, event: AgentEvent) -> None:
        print(event.name, event.payload)

handler = ConsoleEvents()
async with aclosing(agent.stream("Inspect this project")) as events:
    async for event in events:
        await handler.dispatch(event)

Stream errors and cancellation are raised by the stream itself, not synthesized as callbacks. Internal extension events are a separate API.

Note

Dispatch uses AgentEventType, not event.name. Extension-defined names are handled inside on_custom_event().

See also

stream() dispatches events automatically; AgentEvent documents type-specific fields.

async dispatch(event)[source]#

Await the callback named on_<event type>_event exactly once.

Routing uses the finite AgentEventType enum, never a custom event name. All enum values have explicit methods for completion and discovery in editors. Custom names remain data handled by on_custom_event().

Parameters:: event (AgentEvent)
Return type:: None

async on_compaction_started_event(event)[source]#

Handle the start of context compaction.

Parameters:: event (AgentEvent)
Return type:: None

async on_compaction_text_delta_event(event)[source]#

Handle a streamed checkpoint text fragment.

Parameters:: event (AgentEvent)
Return type:: None

async on_compaction_reasoning_delta_event(event)[source]#

Handle a streamed compaction reasoning fragment.

Parameters:: event (AgentEvent)
Return type:: None

async on_compaction_completed_event(event)[source]#

Handle completion of context compaction.

Parameters:: event (AgentEvent)
Return type:: None

async on_model_started_event(event)[source]#

Handle the start of a primary model call.

Parameters:: event (AgentEvent)
Return type:: None

async on_text_delta_event(event)[source]#

Handle an answer-content fragment in event.delta.

Parameters:: event (AgentEvent)
Return type:: None

async on_content_started_event(event)[source]#

Handle the boundary immediately before the first nonempty text delta.

Parameters:: event (AgentEvent)
Return type:: None

async on_content_completed_event(event)[source]#

Handle normal completion of an opened answer-content segment.

Parameters:: event (AgentEvent)
Return type:: None

async on_reasoning_started_event(event)[source]#

Handle the boundary immediately before the first nonempty reasoning delta.

Parameters:: event (AgentEvent)
Return type:: None

async on_reasoning_completed_event(event)[source]#

Handle reasoning ending before content, tool arguments, or the final response.

Parameters:: event (AgentEvent)
Return type:: None

async on_reasoning_delta_event(event)[source]#

Handle a reasoning fragment in event.delta.

Parameters:: event (AgentEvent)
Return type:: None

async on_tool_call_delta_event(event)[source]#

Handle a streamed tool-call fragment.

Parameters:: event (AgentEvent)
Return type:: None

async on_server_tool_started_event(event)[source]#

Handle the start of a provider-hosted tool invocation.

Parameters:: event (AgentEvent)
Return type:: None

async on_server_tool_input_delta_event(event)[source]#

Handle a provider-hosted tool input fragment.

Parameters:: event (AgentEvent)
Return type:: None

async on_server_tool_completed_event(event)[source]#

Handle successful provider-hosted tool completion.

Parameters:: event (AgentEvent)
Return type:: None

async on_server_tool_failed_event(event)[source]#

Handle a structured provider-hosted tool failure.

Parameters:: event (AgentEvent)
Return type:: None

async on_model_completed_event(event)[source]#

Handle the complete model response.

Parameters:: event (AgentEvent)
Return type:: None

async on_model_stopped_event(event)[source]#

Handle a model step that was refused, cut off, interrupted, or paused for good.

Parameters:: event (AgentEvent)
Return type:: None

async on_tool_started_event(event)[source]#

Handle the start of tool execution.

Parameters:: event (AgentEvent)
Return type:: None

async on_tool_completed_event(event)[source]#

Handle a successful tool result.

Parameters:: event (AgentEvent)
Return type:: None

async on_tool_failed_event(event)[source]#

Handle a failed tool result.

Parameters:: event (AgentEvent)
Return type:: None

async on_tool_skipped_event(event)[source]#

Handle a tool call skipped because steering took over.

Parameters:: event (AgentEvent)
Return type:: None

async on_steering_started_event(event)[source]#

Handle the start of steering-message processing.

Parameters:: event (AgentEvent)
Return type:: None

async on_steering_completed_event(event)[source]#

Handle completion of steering-message processing.

Parameters:: event (AgentEvent)
Return type:: None

async on_steering_interrupted_event(event)[source]#

Handle interrupted steering-message processing.

Parameters:: event (AgentEvent)
Return type:: None

async on_internal_message_started_event(event)[source]#

Handle the start of internal-message processing.

Parameters:: event (AgentEvent)
Return type:: None

async on_internal_message_completed_event(event)[source]#

Handle completion of internal-message processing.

Parameters:: event (AgentEvent)
Return type:: None

async on_internal_message_interrupted_event(event)[source]#

Handle interrupted internal-message processing.

Parameters:: event (AgentEvent)
Return type:: None

async on_run_completed_event(event)[source]#

Handle the final successful answer in event.message.

Parameters:: event (AgentEvent)
Return type:: None

async on_custom_event(event)[source]#

Handle extension-defined events using event.name and event.payload.

Parameters:: event (AgentEvent)
Return type:: None

sync(*, runtime=None)#

Return a context-managed synchronous view; optionally share a runtime.

Omitting runtime creates a new background thread for this view. Passing an existing runtime reuses its loop and leaves its shutdown to the caller. This API is for service entry points, not Extension lifecycle hooks.

Examples

Use a model or tool from a plain function:

with read_file.sync() as read:
    result = read({"path": "README.md"})

Parameters:: runtime (SyncRuntime | None)
Return type:: SyncObject[Self]