Emit events and accept user input#

Send progress to a UI#

Use context.emit from a lifecycle hook when work needs visible progress. Here the payload contains a model-step counter from the first-extension tutorial:

    async def before_model(self, context: AgentRunContext, request: ModelRequest) -> None:
        self.active[context] += 1
        await context.emit(
            AgentEvent(
                AgentEventType.CUSTOM,
                session_id=context.session_id,
                name="note.model_step",
                payload={"step": self.active[context]},
            )
        )

The application receives the exact event through on_custom_event. Choose a stable namespaced name and document payload keys as an application protocol. Use JSON-compatible values if the UI forwards them over SSE or WebSocket.

Observe internal state without displaying it#

class UsageObserver(AgentExtension):
    """Retain immutable observations for a demo; use a bounded sink in production."""

    def __init__(self) -> None:
        self.records: list[tuple[str, MessageAppendedEvent]] = []

    async def on_event(self, context: AgentRunContext, event: ExtensionEvent) -> None:
        if isinstance(event, MessageAppendedEvent) and event.usage is not None:
            self.records.append((context.session_id, event))

MessageAppendedEvent records message timing and model usage. The subscriber does not need to infer timing by watching text arrive in the UI. Do not double count cache reads as input tokens or reasoning as additional output tokens. Run Observe usage and timing to inspect the complete example.

To publish a custom internal notification, define a small ExtensionEvent subclass and await context.publish(event). Every extension’s on_event receives it in priority order. Use a different event type when reacting, rather than recursively publishing the same event again. Internal events are not streamed automatically.

Pass target="ExtensionName" to deliver only to the extension with that registered name. A missing or ambiguous target raises ValueError; omitting target preserves ordered broadcast behavior.

Pause a tool for external approval#

Subclass ExternalEventExtension for this behavior. It owns the lock, pending Future, routing, duplicate rejection, loop-safe wake-up, and cleanup. Your subclass owns the tool schema, outbound event, and validation of the returned answer.

class ApprovalExtension(ExternalEventExtension):
    def __init__(self) -> None:
        super().__init__(response_event_name="approval.response", correlation_field="tool_call_id")

    async def on_tool(self, context: AgentRunContext) -> None:
        @tool
        def request_approval(action: str) -> str:
            """Ask the user whether a proposed action should proceed.

            Args:
                action: Human-readable proposed action; this tool does not execute it.

            Snippet:
                request_approval(action="Export notes")

            Guidelines:
                - Ask before performing an action that requires user approval.
            """
            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" if approved else "rejected"

        context.register_tool(request_approval)

    async def before_tool(self, context: AgentRunContext, call: ToolCall) -> None:
        if call.name != "request_approval":
            return
        async with self._wait_for_external_event(context, call.id):
            await context.emit(
                AgentEvent(
                    AgentEventType.CUSTOM,
                    session_id=context.session_id,
                    name="approval.requested",
                    payload={"tool_call_id": call.id, "action": call.arguments["action"]},
                )
            )

The wait is registered before the event is emitted, so an immediate UI reply is safe. The async-with exit waits for the response; it does not block the event loop. Only after the reply is staged does the actual tool execute and consume it.

Return the answer through the Agent#

The application routes the reply using config, separately from business payload:

async def main() -> None:
    config = AgentRunConfig(session_id="approval-demo", request_id="request-1")
    client = await create_agent(
        ApprovalModel(), config=config, extensions=[ApprovalExtension(), ToolGuidelinesExtension()]
    )
    # stream() is annotated AsyncIterator but is an async generator, which aclosing needs.
    async with aclosing(client.stream("Export my notes")) as events:  # type: ignore[type-var]
        async for event in events:
            if event.type == AgentEventType.CUSTOM and event.name == "approval.requested":
                assert event.payload is not None
                print(f"Question: {event.payload['action']}?")
                # A real application sends this from a UI/API callback, not automatically.
                response = ExternalEvent(
                    name="approval.response",
                    payload={"tool_call_id": event.payload["tool_call_id"], "approved": True},
                )
                accepted = client.agent.emit_external_event(response, config=config)
                assert accepted == ["ApprovalExtension"]
                assert client.agent.emit_external_event(response, config=config) == []
                print("Response accepted; duplicate rejected")
            elif event.type == AgentEventType.RUN_COMPLETED:
                assert event.message is not None
                print(event.message.content)

In this offline demonstration the answer is accepted automatically. In a real application, render the question and wait for an actual user choice before emitting the response. The tool does not execute the proposed action itself.

Acceptance and validation#

  • The return value of emit_external_event is a list of accepting extension names.

  • A duplicate, mismatched session, mismatched supplied request ID, stale route, or wrong response name is rejected by the base.

  • Acceptance claims delivery; the tool still validates approved as a bool.

  • RunCancelledEvent cancels pending waits; on_error wakes them with the error.

  • Calling an accepted reply twice does not run the tool twice.

Use Pause for external approval for run instructions and expected output. Protected subclass helper contracts are included on the ExternalEventExtension API page because extension authors need them, despite their underscore names.

Internal continuation versus user steering#

Publish InternalMessageEvent(AgentMessage(content=...)) when an extension needs a bounded follow-on model step. GoalExtension uses this to request more work after review. Do not mutate the loop queue directly. Urgent user steering is a separate built-in channel and has priority over internal input. Neither mechanism interrupts an HTTP request in the middle of receiving a model response.