菜单
Use when adding durable execution to AI SDK for Python, building durable agent loops, or serializing messages across workflow steps.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/vercel-labs/seal --skill ai-python-durable-execution命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
Use when building custom agent loops. Modify tool dispatch, history management, hooks, control flow.
Use for implementing custom providers in AI SDK for Python.
Use when building serverless AI SDK for Python endpoints, handling hook approvals, deferring hooks, or resuming runs across requests.
基于 SOC 职业分类
| name | ai-python-durable-execution |
| description | Use when adding durable execution to AI SDK for Python, building durable agent loops, or serializing messages across workflow steps. |
| metadata | {"sdk-version":"0.2.1"} |
Use durable execution when an agent run must survive restarts, worker moves, or long waits.
The SDK does not provide durability by itself. Build a custom Agent.loop, and
put side effects inside durable steps:
Keep the workflow replayable. Durable steps should take JSON inputs and return JSON outputs.
Serialize messages like this:
data = message.model_dump(mode="json")
message = ai.messages.Message.model_validate(data)
A durable model step should drain ai.stream(...) inside the step and return one
complete assistant Message.
@workflow.step
async def llm_step(
model_data: dict[str, object],
messages_data: list[dict[str, object]],
tools_data: [[, ]],
) -> [, ]:
model = ai.Model.model_validate(model_data)
messages = [
ai.messages.Message.model_validate(message)
message messages_data
]
tools = [ai.Tool.model_validate(tool) tool tools_data]
ai.stream(model, messages, tools=tools) stream:
_event stream:
stream.message :
RuntimeError()
stream.message.model_dump(mode=)
Prefer wrapping the tool body in the durable step:
@ai.tool
@workflow.step
async def ask_mothership(question: str) -> str:
response = await mothership_client.ask(question)
return response.summary
If the workflow system needs separate activity dispatch, schedule a zero-arg
callable that returns ai.tool_result(...). Do not call tool.fn directly.
Use the model step result as a complete message. Do not wrap it in ai.Stream,
ai.events.replay_message_events, or ai.util.merge, those utilities are
used for fluent dispatch in non-durable applications, which is impossible
in a workflow setting since streams are considered side-effects.
class DurableAgent(ai.Agent):
async def loop(self, context: ai.Context):
while context.keep_running():
result = await llm_step(
context.model.model_dump(mode="json"),
[m.model_dump(mode="json") for m in context.messages],
[t.model_dump(mode="json") for t in context.tools],
)
assistant_message = ai.messages.Message.model_validate(result)
context.add(assistant_message)
async with ai.ToolRunner() as runner:
for tool_call in assistant_message.tool_calls:
runner.schedule(context.resolve(tool_call))
async for event in runner.events():
yield event
context.add(runner.get_tool_message())
This pattern does not stream model tokens to the caller. That is usually the right tradeoff for durable workflows, because many durable systems do not support async generators. You can build a queue-based side channel for streaming; however, that kind of stream can't be used to dispatch tools and affect control flow directly.