流式展示一次工具往返

把一个 Python 函数变成工具，并在你的应用中显示它的进度。模型请求 add，收到 42，然后给出最终回答。

运行示例

在仓库根目录下：

uv run python docs/_examples/streaming_tools.py

输出包含：

Answer: 42

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

可复用的三个部分

定义能力。类型注解描述参数，文档字符串告诉模型何时调用它：

@tool
def add(left: int, right: int) -> int:
    """Add two integers exactly.

    Args:
        left: First operand.
        right: Second operand.

    Snippet:
        add(left=20, right=22)

    Guidelines:
        - Use this tool for exact integer addition.
    """
    return left + right

显示进度。事件回调会收到推理、工具调用、工具结果以及增量的回答文本：

class ConsoleEvents(AgentEventDispatcher):
    async def on_reasoning_delta_event(self, event: AgentEvent) -> None:
        print(f"Reasoning: {event.delta}")

    async def on_tool_started_event(self, event: AgentEvent) -> None:
        call = event.tool_calls[0]
        print(f"Tool: {call.name} {call.arguments}")

    async def on_tool_completed_event(self, event: AgentEvent) -> None:
        assert event.message is not None
        print(f"Result: {event.message.content}")

    async def on_text_delta_event(self, event: AgentEvent) -> None:
        print(event.delta)

连接到客户端。当你准备好发出真实请求时，把 MathModel() 替换成你的模型提供商：

async def main() -> None:
    client = await create_agent(MathModel(), tools=[add], event_dispatcher=ConsoleEvents())
    reply = await client.run("What is 20 plus 22?")
    assert reply.content == "Answer: 42"

预设脚本的模型总是调用工具。真实的模型会自行决定是否调用，而且即使内部进行了推理，也可能不会返回可见的推理内容。

完整源码

下载 streaming_tools.py。

"""Run a typed tool and display reasoning, tool results, and final text."""

import asyncio
from collections.abc import AsyncIterator

from zett_agent.client import create_agent
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.events import AgentEvent
from zett_agent.messages import (
    AssistantMessage,
    ToolCall,
    ToolMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)
from zett_agent.tools.base import tool


@tool
def add(left: int, right: int) -> int:
    """Add two integers exactly.

    Args:
        left: First operand.
        right: Second operand.

    Snippet:
        add(left=20, right=22)

    Guidelines:
        - Use this tool for exact integer addition.
    """
    return left + right


class MathModel:
    """Script a model/tool/model round trip without making a network request."""

    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        if isinstance(request.messages[-1], ToolMessage):
            answer = f"Answer: {request.messages[-1].content}"
            yield ModelEvent.text(answer)
            yield ModelEvent.completed(ModelResponse(AssistantMessage(content=answer)))
        else:
            assert any(definition.name == "add" for definition in request.tools)
            yield ModelEvent.reasoning("Use exact arithmetic.")
            yield ModelEvent.completed(
                ModelResponse(AssistantMessage(tool_calls=(ToolCall("sum-1", "add", {"left": 20, "right": 22}),)))
            )


class ConsoleEvents(AgentEventDispatcher):
    async def on_reasoning_delta_event(self, event: AgentEvent) -> None:
        print(f"Reasoning: {event.delta}")

    async def on_tool_started_event(self, event: AgentEvent) -> None:
        call = event.tool_calls[0]
        print(f"Tool: {call.name} {call.arguments}")

    async def on_tool_completed_event(self, event: AgentEvent) -> None:
        assert event.message is not None
        print(f"Result: {event.message.content}")

    async def on_text_delta_event(self, event: AgentEvent) -> None:
        print(event.delta)


async def main() -> None:
    client = await create_agent(MathModel(), tools=[add], event_dispatcher=ConsoleEvents())
    reply = await client.run("What is 20 plus 22?")
    assert reply.content == "Answer: 42"


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

下一步：阅读把流式输出接入终端或界面了解事件处理和取消，或阅读定义并注册工具了解模式、错误和安全执行。