AgentEventDispatcher

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

class zett_agent.dispatcher.AgentEventDispatcher

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.

示例

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.

备注

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

参见

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

async dispatch(event)

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().

参数:

event (AgentEvent)

返回类型:

None

async on_compaction_started_event(event)

Handle the start of context compaction.

参数:

event (AgentEvent)

返回类型:

None

async on_compaction_text_delta_event(event)

Handle a streamed checkpoint text fragment.

参数:

event (AgentEvent)

返回类型:

None

async on_compaction_reasoning_delta_event(event)

Handle a streamed compaction reasoning fragment.

参数:

event (AgentEvent)

返回类型:

None

async on_compaction_completed_event(event)

Handle completion of context compaction.

参数:

event (AgentEvent)

返回类型:

None

async on_model_started_event(event)

Handle the start of a primary model call.

参数:

event (AgentEvent)

返回类型:

None

async on_text_delta_event(event)

Handle an answer-content fragment in event.delta.

参数:

event (AgentEvent)

返回类型:

None

async on_content_started_event(event)

Handle the boundary immediately before the first nonempty text delta.

参数:

event (AgentEvent)

返回类型:

None

async on_content_completed_event(event)

Handle normal completion of an opened answer-content segment.

参数:

event (AgentEvent)

返回类型:

None

async on_reasoning_started_event(event)

Handle the boundary immediately before the first nonempty reasoning delta.

参数:

event (AgentEvent)

返回类型:

None

async on_reasoning_completed_event(event)

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

参数:

event (AgentEvent)

返回类型:

None

async on_reasoning_delta_event(event)

Handle a reasoning fragment in event.delta.

参数:

event (AgentEvent)

返回类型:

None

async on_tool_call_delta_event(event)

Handle a streamed tool-call fragment.

参数:

event (AgentEvent)

返回类型:

None

async on_server_tool_started_event(event)

Handle the start of a provider-hosted tool invocation.

参数:

event (AgentEvent)

返回类型:

None

async on_server_tool_input_delta_event(event)

Handle a provider-hosted tool input fragment.

参数:

event (AgentEvent)

返回类型:

None

async on_server_tool_completed_event(event)

Handle successful provider-hosted tool completion.

参数:

event (AgentEvent)

返回类型:

None

async on_server_tool_failed_event(event)

Handle a structured provider-hosted tool failure.

参数:

event (AgentEvent)

返回类型:

None

async on_model_completed_event(event)

Handle the complete model response.

参数:

event (AgentEvent)

返回类型:

None

async on_model_stopped_event(event)

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

参数:

event (AgentEvent)

返回类型:

None

async on_tool_started_event(event)

Handle the start of tool execution.

参数:

event (AgentEvent)

返回类型:

None

async on_tool_completed_event(event)

Handle a successful tool result.

参数:

event (AgentEvent)

返回类型:

None

async on_tool_failed_event(event)

Handle a failed tool result.

参数:

event (AgentEvent)

返回类型:

None

async on_tool_skipped_event(event)

Handle a tool call skipped because steering took over.

参数:

event (AgentEvent)

返回类型:

None

async on_steering_started_event(event)

Handle the start of steering-message processing.

参数:

event (AgentEvent)

返回类型:

None

async on_steering_completed_event(event)

Handle completion of steering-message processing.

参数:

event (AgentEvent)

返回类型:

None

async on_steering_interrupted_event(event)

Handle interrupted steering-message processing.

参数:

event (AgentEvent)

返回类型:

None

async on_internal_message_started_event(event)

Handle the start of internal-message processing.

参数:

event (AgentEvent)

返回类型:

None

async on_internal_message_completed_event(event)

Handle completion of internal-message processing.

参数:

event (AgentEvent)

返回类型:

None

async on_internal_message_interrupted_event(event)

Handle interrupted internal-message processing.

参数:

event (AgentEvent)

返回类型:

None

async on_run_completed_event(event)

Handle the final successful answer in event.message.

参数:

event (AgentEvent)

返回类型:

None

async on_custom_event(event)

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

参数:

event (AgentEvent)

返回类型:

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.

示例

Use a model or tool from a plain function:

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

参数:

runtime (SyncRuntime | None)

返回类型:

SyncObject[Self]