Python

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

Getting started

Add the vercel-workflow package and workflow entrypoint to pyproject.toml:

[project]
requires-python = ">=3.12"
dependencies = ["vercel-workflow"]

[[tool.vercel.workflows]]
entrypoint = "app.workflows:wf"

The workflow entrypoint uses the module:object format and points to the exported Workflows registry.

Workflows

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(
    sandbox_policy=workflow.SandboxPolicy(share_sandboxes=True),
)

share_sandboxes=True reuses a sandbox across workflow runs for faster startup. Module globals are shared between those runs, so workflow code should not mutate global state.

from app.workflow import wf
from app.steps.generate_draft import generate_draft, summarize_draft

@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"]

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.

Steps

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 executes separately from the workflow orchestrator. While the step executes, the workflow suspends without consuming resources. When the step completes, the workflow resumes automatically where it left off.

Cancellable steps

Pass cancellable=True when a workflow may need to stop a running step. This makes normal Python asyncio cancellation apply to the step: if a task awaiting the step is cancelled in the workflow, then the task running the step will be cancelled as well. For example, asyncio.timeout() cancels the step if it does not finish within the allotted time:

import asyncio
from datetime import timedelta

from app.workflow import wf


@wf.step(cancellable=True)
async def generate_report(*, account_id: str) -> dict[str, str]:
    return await reporting_service.generate(account_id=account_id)


@wf.workflow
async def report_workflow(*, account_id: str) -> dict[str, str]:
    try:
        async with asyncio.timeout(timedelta(minutes=10).total_seconds()):
            return await generate_report(account_id=account_id)
    except TimeoutError:
        return {"status": "timed_out"}

Cancellation is a request, so the workflow waits for the step to terminate before continuing. If the step suppresses its cancellation, it can still return normally.

Starting a workflow

Call workflow.start() from server-side code to start a workflow. It returns a Run that you can use to identify the run, check its status, and wait for its result:

from app.workflows.ai_content_workflow import ai_content_workflow
from vercel import workflow

@app.post("/api/generate")
async def generate_content(*, topic: str):
    run = await workflow.start(ai_content_workflow, topic=topic)

    print(run.run_id)
    print(await run.status())

    # Wait until the workflow completes and return its result.
    return await run.return_value()

Starting a workflow only waits until the run has been created and queued. Await return_value() to wait for the workflow to finish, or save its run_id and recreate the handle later with workflow.Run(run_id).

Sleep

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

from datetime import timedelta

from app.workflow import wf
from vercel import workflow

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

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

    refined = await refine_draft(draft)

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

The parameter accepts four forms:

The string form accepts one or more <value><unit> pairs. Supported units:

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

asyncio.sleep() may also be used to sleep.

Deterministic workflow helpers

Workflow bodies must be fully deterministic, and so are not allowed to perform operations like reading the system clock or generating randomness using the default generator. Deterministic replacements are provided for use in workflow bodies:

• workflow.now() is a deterministic substitute for datetime.datetime.now. It returns the time that the last workflow event occurred at.
• workflow.time_ns() is a deterministic substitute for time.time_ns.
• workflow.random() returns a random.Random instance with a seed based on the run id.

These helpers can only be called from a workflow body. Steps should use the normal system functions.

Hooks

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

import pydantic
from app.workflow import wf
from vercel import workflow

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) -> None:
    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"""

    hook = await approval.resume("draft-123")
    print(f"Resumed workflow run: {hook.run_id}")
    return {"ok": True}

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

Streaming

Steps can stream progress while a workflow is running. Streams are associated with workflow runs, can be acquired from inside a workflow or inside a step, and can be passed to workflows and steps as arguments. If a type is specified when getting the stream (or on the type annotation of a step or workflow it is passed to), then Pydantic will be used to encode and decode the type.

import pydantic

from app.workflow import wf
from vercel import workflow


class Progress(pydantic.BaseModel):
    message: str


@wf.step
async def write_progress(
    writable: workflow.WorkflowWritable[Progress],
):
    for message in ["Drafting", "Reviewing", "Complete"]:
        await writable.write(Progress(message=message))

    await writable.close()

@wf.workflow
async def streaming_workflow():
    writable = workflow.get_writable(type=Progress)
    await write_progress(writable)

Read and validate the typed values from the returned Run as they arrive:

from app.workflows.streaming import Progress, streaming_workflow
from vercel import workflow

@app.post("/api/stream")
async def stream_progress():
    run = await workflow.start(streaming_workflow)

    async for progress in run.readable(type=Progress):
        print(progress.message)

Streams are not closed automatically. It can be manually closed when it is finished so that readers know to terminate.

readable and get_writable also take a namespace parameter, allowing each run to have many distinct streams.

Next steps

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