暂停以等待外部审批

把外部回复与正在等待的工具关联起来。模拟的 UI 会立即作答；重复的回复会被拒绝。实际上不会执行任何导出。

运行示例

在仓库根目录下：

uv run python docs/_examples/approval.py

输出包含：

Response accepted; duplicate rejected

这个程序使用确定性的离线模型，并在需要时使用临时存储。它不需要 API 密钥，由文档测试套件执行。

接入审批 UI

示例会发布一个审批请求，暂停该工具，并通过客户端发回一个关联的回复。在真实的应用中，请把模拟的回复替换成你 UI 的接受或拒绝操作。

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)

在你展示给用户的提示词旁，保存会话、请求和工具调用的标识符。针对另一个请求的回复绝不能为当前请求授权。示例会检查重复的回复被拒绝，并且不会执行真正的导出。

完整源码

下载 approval.py。

"""Round-trip an external approval through the framework's pending-event base."""

import asyncio
from collections.abc import AsyncIterator
from contextlib import aclosing

from zett_agent.agent import (
    AgentRunConfig,
    AgentRunContext,
)
from zett_agent.client import create_agent
from zett_agent.events import (
    AgentEvent,
    AgentEventType,
)
from zett_agent.extensions.external import (
    ExternalEvent,
    ExternalEventExtension,
)
from zett_agent.extensions.tool_guidelines import ToolGuidelinesExtension
from zett_agent.messages import (
    AssistantMessage,
    ToolCall,
    ToolMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)
from zett_agent.tools.base import tool


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"]},
                )
            )
        # Leaving the async-with body waits for the routed external response.


class ApprovalModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        last = request.messages[-1]
        if isinstance(last, ToolMessage):
            assert last.success
            message = AssistantMessage(content="Approval received; no export was executed in this demo.")
        else:
            message = AssistantMessage(
                tool_calls=(ToolCall("approval-1", "request_approval", {"action": "Export notes"}),)
            )
        yield ModelEvent.completed(ModelResponse(message))


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)


if __name__ == "__main__":
    asyncio.run(main())

下一步：阅读发出事件并接收用户输入了解响应路由，或阅读组合内置扩展了解 AskUserExtension 和 PlanModeExtension。