集成到应用#

Zett Agent 提供模型/工具循环，界面、身份认证、会话路由、资源生命周期和各项策略则由你的应用负责。本页说明从脚本走向服务时需要处理哪些边界。

选择调用方式#

消费请求的三种方式#

API

适用场景

await client.run(...)

你需要最终答案。已绑定的分发器仍会收到进度事件。

client.stream(...)

你需要转发事件、提前停止，或者实现交互式控制。

create_agent_sync(...)

你的宿主是阻塞式的 Python 代码。参见 在同步代码中使用。

对于 HTTP 流或 WebSocket，请把事件转换为你自己的传输格式。AgentEvent 可能携带 Python 消息、异常或模型提供商的载荷，它并不是应用层的 JSON 协议。只挑选界面需要的字段转发，并避免外泄密钥或原始异常对象。

设置请求期限#

在 Python 3.10+ 上，用 asyncio.wait_for 为整个被等待的运行设置上限：

import asyncio

try:
    reply = await asyncio.wait_for(client.run("Review the change."), timeout=60)
except asyncio.TimeoutError:
    print("The request timed out.")
else:
    print(reply.content)

超时会取消尚未完成的工作，并等待取消清理结束。清理可能让总耗时超过期限。模型提供商的重试和工具等待同样占用这个期限；max_iterations 并不是超时设置。

取消并不等于回滚。工具可能已经改动了外部系统，而运行在工作线程中的同步函数无法通过取消它的等待者被强制停止。请为会修改状态的工具单独设置超时、事务规则和幂等控制。绝不要仅仅因为界面连接断开就自动重放整个请求。

断开连接时关闭流#

from contextlib import aclosing

async with aclosing(client.stream("Inspect the project.")) as events:
    async for event in events:
        await send_to_ui(event)

send_to_ui 是你自己的应用传输层。如果它在连接断开时抛出异常，上下文管理器会关闭这个尚未完成的请求。循环提前退出时也会进行同样的清理。在 Web 框架中，请传播请求取消，而不要任其留下脱离的生成任务意外运行。

挂在这个客户端上的分发器会在事件产出 之前 被调用。不要自己再次分发该事件。回调按顺序等待执行，可能拖慢生成速度；对于较慢的输出端，请使用有界队列。对于 GUI，请把控件更新调度到 GUI 线程上执行，而不是在异步或后台循环的回调中直接操作控件。

区分失败边界#

什么会终止请求？#

情况

行为

工具处理函数抛出普通异常

发出 TOOL_FAILED 事件，并向模型返回一个不成功的工具结果。模型可以继续运行。

模型提供商、协议、生命周期钩子或回调失败

向外传播该异常。不要假设一定会收到 RUN_COMPLETED 事件。

请求被取消或流提前关闭

取消尚未完成的工作并释放会话。不会合成成功的最终答案。

模型调用预算耗尽

抛出 AgentIterationLimitError。只有在决定如何处理未完成的工作之后，才能使用新的用户请求。

在 finally 中关闭加载指示器。处理错误时要保留取消语义：不要把 CancelledError 转换成普通的成功回复。对于用户可见的错误，显示一条安全的提示信息，并把详细的诊断信息写入受访问控制的输出目标。工具处理函数抛出的异常会以文本形式对模型可见，因此在离开敏感集成前要对它们做脱敏处理。

路由并发对话#

不同的会话可以在同一个运行时上并发运行：

from zett_agent.agent import AgentRunConfig

replies = await asyncio.gather(
    client.run("First conversation", config=AgentRunConfig(session_id="chat-a")),
    client.run("Second conversation", config=AgentRunConfig(session_id="chat-b")),
)

针对 同一个会话 的重叠请求会被拒绝。请在你的应用中排队，或者把支持的外部输入发送给正在进行的请求；不要为这个对话再启动第二个模型循环。client.agent.is_conflict(id) 适合展示，但运行时的准入检查才是权威。检查状态请用 client.agent.get_state(session_id)，而不是便捷的 state 视图，后者反映的是最近一次启动的请求。

这层保护只属于单个运行时实例，它不是跨进程或跨客户端的分布式锁。在多工作进程的服务中，请把会话路由到唯一的活动工作进程，或者在应用的共享协调层中强制串行化。会话 ID 只是标识符，不代表授权：在运行、列出或删除某个会话之前，请先校验该用户的访问权限。

如果回调要服务多个会话，请按 event.session_id 进行路由。把扩展状态保持在请求作用域内，并确保共享工具在并发下是安全的；参见 状态、并发与清理 和 定义并注册工具。

显式管理资源#

在应用启动时创建模型提供商和持久化存储，在请求使用期间保持它们打开，并在关闭时关闭它们：

from zett_agent.client import create_agent
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.extensions.tool_guidelines import ToolGuidelinesExtension

history = SQLiteSessionExtension("sessions.sqlite")
try:
    client = await create_agent(
        model,
        extensions=[history, ToolGuidelinesExtension()],
    )
    await serve_application(client)
finally:
    await history.close()
    await model.aclose()

model 是你配置好的提供商；serve_application 是你服务器的请求生命周期。关闭流程应先停止接纳新请求，并在关闭资源之前处理完正在进行的请求。运行时会关闭它自己创建的流，但不会关闭你提供的提供商或存储对象。MCP 传输在作用域上是例外：连接 MCP 工具 会按请求打开和关闭它们。

一个离线集成示例#

async def main() -> None:
    model = ApplicationModel()
    client = await create_agent(model)
    replies = await asyncio.gather(
        client.run("A", config=AgentRunConfig(session_id="chat-a")),
        client.run("B", config=AgentRunConfig(session_id="chat-b")),
    )
    assert [reply.content for reply in replies] == ["A1", "B1"]
    print("Separate sessions: A1, B1")
    assert (await client.run("A", config=AgentRunConfig(session_id="chat-a"))).content == "A2"

    waiting = asyncio.create_task(client.run("wait", config=AgentRunConfig(session_id="timeout")))
    await model.started.wait()
    try:
        await asyncio.wait_for(waiting, timeout=0)
    except asyncio.TimeoutError:
        pass
    else:
        raise AssertionError("The waiting request should time out")
    assert model.closed.is_set() and not client.agent.is_conflict("timeout")
    state = client.agent.get_state("timeout")
    assert state is not None and state.phase == AgentPhase.CANCELLED
    # Cancellation releases the session; it does not remove the recorded input.
    assert (await client.run("retry", config=AgentRunConfig(session_id="timeout"))).content == "retry2"
    print("Timeout cleaned up; session reused.")

在源码检出目录中运行 uv run python docs/_examples/application.py：

Separate sessions: A1, B1
Timeout cleaned up; session reused.

完整程序 无需网络请求即可验证独立的历史记录和取消清理。

部署之前#

  • 锁定并测试你的应用所依赖的模型与端点组合。

  • 在代码中实施工具访问控制，而不只是写在提示词里；同时隔离 shell 和文件访问。

  • 设置请求期限、模型提供商重试策略和单个工具的限额。

  • 串行化每个对话，并选择一个可恢复历史记录的扩展。

  • 确定哪些提示词、图像、工具结果和追踪数据可以存储或记录日志。

  • 测试成功、工具失败、连接断开、取消以及重启后的恢复。

下一步：需要快速回归测试请看 测试你的智能体，需要了解历史记录和保留策略请看 保存并恢复对话。