ExternalEventExtension#

Build extensions, choose hooks, and manage request-scoped state. See the guide.

class zett_agent.extensions.external.ExternalEventExtension(*, response_event_name, correlation_field)[源代码]#

Base class for extensions that suspend until an external response arrives.

Use this base for tools that need a UI decision before execution can finish. The subclass owns the tool, outbound UI event, and business validation. The base owns pending Futures, routing, duplicate rejection, thread-safe wake-up, and cleanup. It does not send HTTP responses or render UI components.

参数:

  • response_event_name (str | Collection[str]) -- Accepted inbound name, or a collection of names.

  • correlation_field (str) -- Payload key identifying the operation, usually tool_call_id. Its value must equal the ID passed to the wait helper.

The configuration passed separately to accept(config, event) selects the waiting session and optional request ID. Agent does not inject routing identifiers into the payload.

Lifecycle:

+---------------------------+     +---------------------------+
| before_tool()             |     | UI / external caller      |
+---------------------------+     +---------------------------+
| enter async with:         |     |                           |
| register pending Future   |     |                           |
| context.emit(CUSTOM)      |---->| display question          |
| leave body: await Future  |     | emit_external_event(      |
|                           |<----|     event, config=config) |
| accept(): claim response, |     +---------------------------+
| schedule Future completion|
| stage response by context |
+-------------+-------------+
              |
              v
+---------------------------+
| tool executes             |
| _take_external_event()    |
| validate and return result|
+---------------------------+

Entering the async context registers the wait BEFORE the UI event is emitted, so even an immediate response is safe. The await happens when leaving the async-with body, not on entry. It suspends this task without blocking other sessions. The tool executes only after the response has been staged.

Example:

from zett_agent.agent import AgentRunContext
from zett_agent.events import (
    AgentEvent,
    AgentEventType,
)
from zett_agent.extensions.external import ExternalEventExtension
from zett_agent.messages import ToolCall
from zett_agent.tools.base import tool

class ConfirmationExtension(ExternalEventExtension):
    def __init__(self) -> None:
        super().__init__(
            response_event_name="confirmation_response",
            correlation_field="tool_call_id",
        )

    async def on_tool(self, context: AgentRunContext) -> None:
        @tool
        async def confirm() -> bool:
            '''Ask the user whether to proceed.

            Snippet:
                confirm()

            Guidelines:
                - Use when proceeding requires an explicit user decision.
            '''
            response = self._take_external_event(context)
            approved = response.payload.get("approved")
            if not isinstance(approved, bool):
                raise ValueError("approved must be a boolean")
            return approved

        context.register_tool(confirm)

    async def before_tool(self, context: AgentRunContext, call: ToolCall) -> None:
        if call.name != "confirm":
            return
        async with self._wait_for_external_event(context, call.id):
            await context.emit(AgentEvent(
                AgentEventType.CUSTOM,
                session_id=context.session_id,
                name="confirmation_requested",
                payload={"tool_call_id": call.id, "question": "Proceed?"},
            ))

Usage:

from zett_agent.agent import (
    Agent,
    AgentRunConfig,
)
from zett_agent.extensions.external import ExternalEvent

config = AgentRunConfig(session_id="session-42", request_id="request-1")
agent = await Agent.create(model, config=config, extensions=[ConfirmationExtension()])

# Consume agent.stream(...) and forward confirmation_requested to the UI.
# A separate UI callback returns the exact tool_call_id from that event.
accepted_by = agent.emit_external_event(
    ExternalEvent(
        name="confirmation_response",
        payload={"tool_call_id": "call-7", "approved": True},
    ),
    config=config,
)
# ["ConfirmationExtension"] if accepted; [] for a stale/duplicate response.

Acceptance means delivery was claimed, not that tool execution succeeded. Payload business validation belongs to the subclass. A missing config, mismatched request, unknown correlation ID, or duplicate response is rejected. Different sessions may reuse the same tool_call_id without sharing responses.

Cancellation clears staged responses and cancels pending Futures; on_error wakes pending waits with the error. Overrides of on_event/on_error must call super() to preserve cleanup. Future mutation is scheduled on its owning loop when accept is called from another thread. Keep that loop alive until request cleanup has finished. This base supports sequential tools: one staged response per AgentRunContext, not concurrent external-response tools in the same request.

_wait_for_external_event(context, correlation_id, *, response_event_name=None)[源代码]#

Register on entry; await and stage the response when the body exits.

Yield the outbound AgentEvent inside this async-with body. Optionally restrict response_event_name to one of the configured inbound names. The pending entry is removed on every exit, including cancellation.

参数:

  • context (AgentRunContext)

  • correlation_id (str)

  • response_event_name (str | None)

返回类型:

AsyncGenerator[None]

_take_external_event(context)[源代码]#

Pop this invocation's staged response exactly once inside its tool.

Call only after _wait_for_external_event has finished successfully. Raises RuntimeError if no response is staged or it was already consumed.

参数:

context (AgentRunContext)

返回类型:

ExternalEvent

accept(config, event)[源代码]#

Claim a matching response and wake its waiter on the Future's loop.

Usually called by Agent.emit_external_event, not directly by the UI. True means this extension accepted delivery; False leaves other extensions free to recognize the event. Neither argument is modified.

参数:
返回类型:

bool

async on_event(context, event)[源代码]#

Cancel active waits when their request enters CANCELLED.

参数:
返回类型:

None

async on_error(context, error)[源代码]#

Complete active waits with the request's original error.

参数:
返回类型:

None

async after_model(context, response)#

Run after the complete assistant message is appended.

参数:
返回类型:

None

async after_run(context, result)#

Run after a successful final answer is appended.

参数:
返回类型:

None

async after_tool(context, call, result, error)#

Inspect or modify a tool result before it is appended.

error is the original execution exception for a failed result and None after successful execution. Changes made to result are included in context, persistence, and the next model request. Raising prevents the result from being appended.

参数:
返回类型:

None

async after_turn(context, result)#

Run once the turn's assistant message and tool results are settled.

result is the assistant message that opened the turn: either the final answer, or the tool-calling message whose ToolMessages were appended. Every requested call is settled first, including calls that failed, and calls that steering skipped, which arrive as unsuccessful or skipped ToolMessages.

Steering that interrupts a turn is accepted while the runtime settles that batch, so its UserMessage is already appended when this hook runs and the steering turn follows as the next turn.

A turn that never completed, because the request failed or was cancelled, has no after_turn call; use on_error and RunCancelledEvent for those paths.

参数:
返回类型:

None

async before_model(context, request)#

Inspect the current request before preprocessing a primary model call.

Request fields are frozen. Change messages through AgentRunContext.add_message() or AgentRunContext.replace_messages(), local tools through context.tools, and provider tools through context.server_tools; the runtime rebuilds these fields before the next hook and before calling the provider. This is a shallow request view, not an immutable copy of each message or tool definition.

示例

Inspect the tools offered in this iteration:

async def before_model(self, context, request):
    offered = {definition.name for definition in request.tools}
    if "write_file" in offered:
        context.add_message(
            SystemMessage(content="Read existing files before editing."),
        )
参数:
返回类型:

None

async before_run(context)#

Run after the transformed user input is appended and published.

参数:

context (AgentRunContext)

返回类型:

None

async before_tool(context, call)#

Run before one requested tool is invoked.

参数:
返回类型:

None

async before_turn(context)#

Run before this turn's model call.

Runs after before_run() on the first turn and after the previous turn's after_turn() on later ones. Model request assembly, compaction, and before_model() follow this hook, so context edited here reaches the provider.

参数:

context (AgentRunContext)

返回类型:

None

property name: str#

Unique registration identity, defaulting to the concrete class name.

Override with a class attribute or assign an instance name to register independently configured instances of the same extension type.

async on_compact(context)#

Summarize restored context when a caller asks for compaction.

Agent.compact restores the session through on_tool and on_state exactly as a request does, appends no input message, and makes no primary model call: the only work a compaction pass does is what this hook performs. The default does nothing, so an extension with nothing to compact pays nothing.

示例

Usage:

async def on_compact(self, context):
    await self.compact(context)
参数:

context (AgentRunContext)

返回类型:

None

async on_message(context)#

Transform context.input_message after state restoration, before append.

Replace the pending UserMessage to implement externally selected modes. Do not append or persist it here: the runtime appends the final message exactly once after all input hooks finish. before_run sees that message through AgentRunContext.add_message() or AgentRunContext.replace_messages(), and persistence subscribers have already received it.

参数:

context (AgentRunContext)

返回类型:

None

async on_model_request(context, request, call_next)#

Wrap one provider request and its complete streamed response.

Pass request (or a dataclass replacement) to call_next. The default closes the inner stream so cancellation reaches the provider.

参数:
返回类型:

AsyncIterator[ModelEvent]

async on_state(context)#

Restore history and initialize state before any input conversion.

示例

Usage:

async def on_state(self, context):
    context.add_message(SystemMessage(content="Use concise answers."), index=0)
参数:

context (AgentRunContext)

返回类型:

None

async on_success(context, result)#

Run once after all after_run hooks succeed, before RUN_COMPLETED.

This callback observes a completed request, not an intermediate model step. Failed or cancelled requests do not trigger it. Callback errors propagate through on_error and prevent RUN_COMPLETED from being emitted.

示例

Usage:

async def on_success(self, context, result):
    await save_answer(context.session_id, result.content)
参数:
返回类型:

None

async on_tool(context)#

Register local or provider-hosted tools before context restoration.

All extensions finish this hook before the first on_state() call, so prompt extensions can reliably inspect context.tools and context.server_tools as complete request-scoped registries.

示例

Usage:

async def on_tool(self, context):
    context.register_tool(read_file)
    context.register_server_tool(ServerToolDefinition(type="web_search"))
参数:

context (AgentRunContext)

返回类型:

None

async on_tool_call(context, call, call_next)#

Wrap one registered local tool handler and return its ToolResult.

An extension may adjust call.arguments, return without calling the next layer, transform its result, or raise into normal tool-failure handling.

参数:
返回类型:

ToolResult

property pending: bool#

Return whether this extension is participating in an active request.

The Agent updates this state for every registered extension, including extensions whose hooks are currently idle. One extension instance may be shared by multiple Agents and sessions; pending remains true until all requests using that instance have finished, failed, or been cancelled. Persistent data retained between requests does not count.