Workflows with Python
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.
Add the vercel package and workflow entrypoint to pyproject.toml:
[project]
requires-python = ">=3.12"
dependencies = ["vercel"]
[[tool.vercel.workflows]]
entrypoint = "app.workflows:wf"The workflow entrypoint uses the module:object format and points to the
exported Workflows registry.
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,
}Export the registry from the workflow package and import the module containing your workflow so its definitions are registered:
from app.workflow import wf
from app.workflows import ai_content_workflow
__all__ = ["ai_content_workflow", "wf"]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 summaryEach 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
from app.workflow import wf
@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 parameter accepts three forms:
| Form | Description | Example |
|---|---|---|
str | Human-readable duration string | "2 days", "1w", "1h 30m" |
int or float | Milliseconds from now | 5000 (5 seconds) |
datetime.datetime | Absolute wake-up time (must be timezone-aware) | datetime(2025, 1, 1, tzinfo=UTC) |
The string form accepts one or more <value><unit> pairs. Supported units:
| Duration | Unit |
|---|---|
| Milliseconds | ms |
| Seconds | s, second, seconds |
| Minutes | m, minute, minutes |
| Hours | h, hour, hours |
| Days | d, day, days |
| Weeks | w, week, weeks |
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:
import typing, pydantic
from vercel import workflow
from app.workflow import wf
class Approval(pydantic.BaseModel, workflow.BaseHook):
"""Human approval for AI-generated drafts"""
decision: typing.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:
from app.workflows.approval import Approval
@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}Was this helpful?
