发出事件并接收用户输入#

向 UI 发送进度#

当工作需要可见的进度时，请在生命周期钩子里使用 context.emit。这里的负载包含第一个扩展教程中的模型步骤计数器：

    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]},
            )
        )

应用程序通过 on_custom_event收到这个确切的事件。请选择一个稳定的带命名空间的名称，并把负载键的说明当作应用协议来记录。如果 UI 会通过 SSE 或 WebSocket 转发这些值，请使用与 JSON 兼容的值。

观察内部状态而不展示它#

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 记录消息的时序和模型用量。订阅者不必通过观察文本在 UI 中到达来推断时序。不要把缓存读取重复计为输入 token，也不要把推理计为额外的输出 token。运行观察用量与耗时查看完整示例。

要发布自定义的内部通知，请定义一个小小的 ExtensionEvent 子类，然后调用 await context.publish(event)。每个扩展的 on_event 都会按优先级顺序收到它。做出反应时请使用不同的事件类型，不要递归地再次发布同一个事件。内部事件不会自动流式传输。

传入 target="ExtensionName"可以只把事件投递给注册名匹配的那个扩展。目标缺失或有歧义会抛出 ValueError；省略 target则保持按顺序广播的行为。

暂停工具以等待外部审批#

这种行为请继承 ExternalEventExtension。它负责锁、待处理的 Future、路由、重复拒绝、对事件循环安全的唤醒以及清理。你的子类负责工具模式、对外发出的事件，以及校验收到的应答。

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"]},
                )
            )

等待在事件发出之前就已经注册好，所以 UI 立即回复也是安全的。async-with 退出时会等待响应，它不会阻塞事件循环。只有回复被暂存之后，真正的工具才会执行并消费它。

通过 Agent 返回应答#

应用程序用配置来路由这份回复，与业务负载分开：

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)

在这个离线演示里，应答会被自动接受。在真实应用中，请先展示问题，等用户真正做出选择之后，再发出响应。工具本身不会执行所提议的操作。

接受与校验#

emit_external_event 的返回值是接受该事件的扩展名列表。
重复的应答、会话不匹配、提供的请求 ID 不匹配、路由已过期或响应名称不对，都会被基类拒绝。
接受只表示投递成功；工具仍会把 approved校验为 bool。
RunCancelledEvent 会取消待处理的等待；on_error 会用该错误把它们唤醒。
对一个已被接受的应答调用两次，不会让工具执行两次。

运行说明和预期输出参见暂停以等待外部审批。受保护的子类辅助方法契约会列在ExternalEventExtension的 API 页面里，因为扩展作者需要它们，尽管它们的名字以下划线开头。

internal continuation 与 user steering#

当扩展需要一次有边界的后续模型步骤时，请发布 InternalMessageEvent(AgentMessage(content=...))。GoalExtension 用它来在评审之后请求更多工作。不要直接修改循环队列。紧急的 user steering 是另一条内置通道，优先级高于内部输入。这两种机制都不会中断正在接收模型响应的 HTTP 请求。