该技能文档适用于在无服务器(Serverless)环境(例如 Vercel Fluid Compute)中构建 Python 版 AI SDK 端点。无服务器环境与常规环境最大的差异在于对工具审批及其他钩子的处理:由于无法让 hook 的 future 持续存活,需要主动停止运行、保存消息,然后在后续请求中预先注册钩子解决方案再恢复运行。
一、工具审批 用 require_approval=True 标记需要审批的工具。示例中定义了一个删除文件的工具:
@ai.tool(require_approval=True) async def delete_file(path: str) -> str: return f"Deleted {path}"
二、首个请求 当出现 deferred hook 时,将其发送给客户端并调用 ai.defer_hook(...)。需要持续消费流,不要遇到第一个 hook 就 break,以便让同级工具完成或被标记为 deferred,并使 stream.messages 保持完整。
deferred_hooks = []
async with agent.run(model, messages) as stream: async for event in stream: if ( isinstance(event, ai.events.HookEvent) and event.hook.status == "pending" ): deferred_hooks.append(event.hook) ai.defer_hook(event.hook)
yield event
saved_messages = [ message.model_dump(mode="json") for message in stream.messages ] save_messages(saved_messages) save_deferred_hook_ids([hook.hook_id for hook in deferred_hooks])
三、恢复请求 加载已保存的消息,预先注册钩子解决方案,然后调用 agent.run。
messages = [ ai.messages.Message.model_validate(message) for message in load_messages() ]
for approval in approvals: ai.resolve_hook( approval.hook_id, ai.tools.ToolApproval( granted=approval.granted, reason=approval.reason, ), )
async with agent.run(model, messages) as stream: async for event in stream: yield event
save_messages([ message.model_dump(mode="json") for message in stream.messages ])
必须在 agent.run(...) 之前调用 ai.resolve_hook(...),不要让模型再次发起该工具调用。
Agent.run 会准备已保存的中断消息以便重放:已完成的同级工具结果会被复用,deferred hook 会收到预注册的解决方案,仅用于重放的事件会对调用方隐藏。
四、规则
- 使用常规的 agent.run(...);无服务器恢复通常不需要自定义循环。
- 如果确实要写自定义循环,请使用 context.resolve(...)、ToolRunner 和 context.add(...),以保证审批和重放继续正常工作。
- 自定义钩子可通过 ai.resolve_hook(hook_id, data, payload=PayloadType) 进行预注册。
- 若使用 AI SDK UI 客户端,可用 ai-python-ui-adapter 处理消息转换、审批响应和 SSE。