Python

Build durable workflows and AI agents in Python with the Vercel SDK.

You can build durable workflows in Python using the vercel Python SDK. Your workflow code can pause, resume, and maintain state, just like the JavaScript and TypeScript Workflow SDK.

Install the vercel package:

Configure experimentalServices in your vercel.json:

{
  "experimentalServices": {
    "ai_content_workflow": {
      "type": "worker",
      "entrypoint": "app/workflows/ai_content_workflow.py",
      "topics": ["__wkf_*"]
    }
  }
}

A workflow is a stateful function that coordinates multi-step logic over time. Create a Workflows instance and use the @wf.workflow decorator to mark a function as durable:

from vercel import workflow

wf = workflow.Workflows()

from app.workflow import wf

@wf.workflow
async def ai_content_workflow(*, topic: str):
    draft = await generate_draft(topic=topic)
    summary = await summarize_draft(draft=draft)

    return {
        "draft": draft,
        "summary": summary,
    }

Under the hood, the workflow compiles into a route that orchestrates execution. All inputs and outputs are recorded in an event log. If a deploy or crash happens, the system replays execution deterministically from where it stopped.

A step is a stateless function that runs a unit of durable work inside a workflow. Use @wf.step to mark a function as a step:

import random
from app.workflow import wf

@wf.step
async def generate_draft(*, topic: str):
    return await ai_generate(prompt=f"Write a blog post about {topic}")

@wf.step
async def summarize_draft(*, draft: str):
    summary = await ai_summarize(text=draft)

    # Simulate a transient error. The step automatically retries.
    if random.random() < 0.3:
        raise Exception("Transient AI provider error")

    return summary

Each step compiles into an isolated route. While the step executes, the workflow suspends without consuming resources. When the step completes, the workflow resumes automatically where it left off.

Sleep pauses a workflow for a specified duration without consuming compute resources:

from vercel import workflow

@wf.workflow
async def ai_refine_workflow(*, draft_id: str):
    draft = await fetch_draft(draft_id)

    await workflow.sleep("7 days")  # Wait 7 days to gather more signals.

    refined = await refine_draft(draft)

    return {
        "draft_id": draft_id,
        "refined": refined,
    }

The sleep call pauses the workflow and consumes no resources. The workflow resumes automatically when the time expires.

A hook lets a workflow wait for external events such as user actions, webhooks, or third-party API responses.

Define a hook model with Pydantic and workflow.BaseHook:

from vercel import workflow

class Approval(BaseModel, workflow.BaseHook):
    """Human approval for AI-generated drafts"""

    decision: Literal["approved", "changes"]
    notes: str | None = None

@wf.workflow
async def ai_approval_workflow(*, topic: str):
    draft = await generate_draft(topic=topic)

    # Wait for human approval events
    async for event in Approval.wait(token="draft-123"):
        if event.decision == "approved":
            await publish_draft(draft)
            break

        revised = await refine_draft(draft, event.notes)
        await publish_draft(revised)

Resume the workflow when data arrives:

@app.post("/api/resume")
async def resume(approval: Approval):
    """Resume the workflow when an approval is received"""

    await approval.resume("draft-123")
    return {"ok": True}

When a hook receives data, the workflow resumes automatically. You don't need polling, message queues, or manual state management.

For comprehensive documentation, examples, and the latest updates, visit the official Vercel Workflow Python documentation.

Learn more about the Foundations.
Check Errors if you encounter issues.
Explore the API Reference.

推荐订阅源

Workflow SDK Documentation