--- title: Build durable workflows in Python on Vercel description: Keep long-running Python jobs alive through crashes and deploys with steps, retries, sleeps, and hooks. url: "https://vercel.com/kb/guide/python-and-durable-workflows" published: 2026-10-26 last_updated: 2026-10-26 authors: Ben Sabic install_vercel_plugin: npx plugins add vercel/vercel-plugin --- When someone signs up, your app creates their profile and sends a welcome email. Three days later, it checks whether they finished setup and sends a reminder if they still need one. Use the Python Workflow SDK to continue that process after the signup request ends. It records completed steps and schedules the three-day wait so the workflow can resume when the follow-up is due. Your FastAPI route starts the workflow, which calls your application's database and email functions. The SDK retries failed steps and uses saved results to resume interrupted work. > The Python SDK is in beta. APIs and behavior may change before general availability. ## What is the Python Workflow SDK? The [Python Workflow SDK](https://pypi.org/project/vercel-workflow/) adds durable execution to async Python functions. It uses [Vercel Workflows](https://vercel.com/docs/workflows), the platform behind the TypeScript SDK, to store execution history and resume interrupted work. You define the process with Python control flow, including loops and exception handling. The workflow coordinates the order of operations while separate step functions perform database writes and API calls. ### Workflows and steps Every workflow is made of two kinds of function: | | Workflow | Step | | ------------ | -------------------------------------- | -------------------------------------------- | | Decorator | `@wf.workflow` | `@wf.step` | | Job | Decides what happens next | Does one unit of real work | | Side effects | None: no network, clock, or randomness | Yes: database writes, API calls, model calls | | On failure | Replays from its event log | Retries automatically | | Can call | Steps, `sleep()`, hooks | Your own code and libraries | Your route starts a run and can return before the workflow finishes. When the workflow awaits a step, that step executes separately and the workflow suspends without consuming compute. The step's result is saved in the event log so the workflow can use it when execution resumes. ### How a run survives failure The SDK records workflow and step inputs and outputs in an event log. During recovery, it replays the workflow body and returns saved results for completed steps. An interrupted step can run again if its result was not recorded, even when an external action already succeeded. Pass only the data the next operation needs. Read provider credentials inside steps from server-side configuration instead of returning them to the workflow. On Vercel, a run stays pinned to the deployment that started it. New deploys don't change the code under a run that's already in flight. ### Two rules for workflow code Replay requires the workflow body to make the same decisions from the same recorded history. Use `workflow.now()` or `workflow.time_ns()` for time and `workflow.random()` for seeded randomness. Keep module globals unchanged, especially when sandboxes are shared between runs. Network requests, database writes, and file operations belong in steps. Steps can use ordinary system APIs, but external writes must tolerate retries because the SDK records their results only after execution succeeds. ## Prerequisites Before you begin, you need: - A [Vercel account](https://vercel.com/signup). - Python 3.12 or a newer version supported by the [Vercel Python runtime](https://vercel.com/docs/functions/runtimes/python). - Working knowledge of `async`/`await` and [FastAPI](https://fastapi.tiangolo.com/). - An application with user records, an email provider, and request authentication. For local development, you also need [uv](https://docs.astral.sh/uv/), Node.js 24+, and the [Vercel CLI](https://vercel.com/docs/cli). The database, email, and authentication functions below are application adapters that you must supply. Their expected behavior is described beside each example. The Workflow SDK itself does not depend on FastAPI. ## Step 1: Set up the project Create the project with Python 3.12 and install its dependencies: ```bash uv init --python 3.12 onboarding-workflow cd onboarding-workflow uv add vercel-workflow fastapi pydantic ``` Append this to `pyproject.toml` without replacing its generated fields: ```toml [[tool.vercel.workflows]] entrypoint = "app.workflows:wf" ``` The `module:object` entrypoint refers to the registry exported in Step 3. Commit `pyproject.toml` and `uv.lock` to deploy the resolved versions. The umbrella `vercel` package works too, since it depends on `vercel-workflow` and exposes the same `vercel.workflow` module. Install `vercel` if you also use other Vercel SDKs such as Sandbox or Blob. Next, create the registry in `app/workflow.py`. Every workflow and step in your project registers against this one object: ```python from vercel import workflow wf = workflow.Workflows( sandbox_policy=workflow.SandboxPolicy(share_sandboxes=True), ) ``` Sharing sandboxes reduces startup work by reusing a sandbox across runs. Module globals are shared too, so store per-run state in workflow variables and step results. Link the folder to your Vercel project for development and deployment: ```bash vercel link ``` Create these files as you work through Steps 2 through 4. The client and authentication modules adapt your existing application services: ```plaintext onboarding-workflow/ ├── pyproject.toml ├── uv.lock └── app/ ├── __init__.py ├── auth.py # your request authentication ├── clients.py # your database and email adapters ├── main.py # FastAPI app ├── workflow.py # shared registry ├── steps/ │ ├── __init__.py │ └── onboarding.py └── workflows/ ├── __init__.py # exports wf and imports workflow modules └── onboarding.py ``` ## Step 2: Write the steps Save the onboarding steps in `app/steps/onboarding.py`. Configure `app/clients.py` with these async adapters before running the example: | Adapter | Required behavior | | --------------------------------- | ------------------------------------------------------------------------------------------------ | | `db.users.get(user_id)` | Return an existing user with an email field, or `None` | | `db.profiles.upsert(user_id=...)` | Create a profile once under a unique user ID and preserve existing setup progress | | `db.profiles.get(user_id)` | Return the profile with its current setup-complete flag | | `email.send(...)` | Send the named template, enforce the supplied idempotency key, and return an HTTP-style response | Install the adapters' client packages and set credentials on the server. Seed a user for local tests and use a test email provider or stub. ```python from vercel.workflow import get_step_metadata from app.workflow import wf from app.clients import db, email @wf.step async def create_profile(*, user_id: str) -> dict: user = await db.users.get(user_id) await db.profiles.upsert(user_id=user_id) return {"email": user.email} @wf.step async def send_email(*, to: str, template: str) -> None: step = get_step_metadata() response = await email.send( to=to, template=template, idempotency_key=step.step_id, ) response.raise_for_status() @wf.step async def has_finished_setup(*, user_id: str) -> bool: profile = await db.profiles.get(user_id) return profile.setup_complete ``` The step ID stays the same across retries of that call. Forward it to an email provider that supports idempotency and choose a deduplication window that covers your retry period. If your provider lacks that feature, your adapter needs a delivery strategy that handles uncertain send outcomes. The step returns the email address because database calls cannot run in the workflow body. Step 5 handles missing users and rate limits. ## Step 3: Write the workflow Save the workflow in `app/workflows/onboarding.py`. It sends the welcome email, waits three days, and checks whether a reminder is still needed: ```python from datetime import timedelta from vercel import workflow from app.workflow import wf from app.steps.onboarding import create_profile, has_finished_setup, send_email @wf.workflow async def onboard_user(*, user_id: str) -> str: user = await create_profile(user_id=user_id) await send_email(to=user["email"], template="welcome") await workflow.sleep(timedelta(days=3)) if await has_finished_setup(user_id=user_id): return "activated" await send_email(to=user["email"], template="setup-nudge") return "nudged" ``` The durable sleep suspends the run until the follow-up is due without consuming compute during the wait. Call it in the workflow body; calling it inside a step raises a `RuntimeError`. Sleep accepts a `timedelta`, seconds as an integer or float, duration strings such as `"3 days"`, or a timezone-aware `datetime` for an absolute wake-up time. In `app/workflows/__init__.py`, export the registry and import the workflow module so its definitions register: ```python from app.workflow import wf from app.workflows import onboarding __all__ = ["onboarding", "wf"] ``` _This is the_ `app.workflows:wf` _entrypoint from Step 1._ ## Step 4: Start the workflow from FastAPI Start the workflow after your application has created the user record. The start call waits until the run is created and queued, then returns a handle without waiting for the three-day process to finish. In `app/main.py`, your `current_user_id` dependency must reject unauthenticated requests and return the user's ID. Save each run's owner in the database and restrict status requests to that user: ```python from fastapi import Depends, FastAPI from vercel import workflow from app.auth import current_user_id from app.clients import db from app.workflows.onboarding import onboard_user app = FastAPI() @app.post("/api/onboarding", status_code=202) async def start_onboarding(user_id: str = Depends(current_user_id)): run = await workflow.start(onboard_user, user_id=user_id) await db.workflow_runs.save(run_id=run.run_id, user_id=user_id) return {"run_id": run.run_id} @app.get("/api/onboarding/{run_id}") async def onboarding_status( run_id: str, user_id: str = Depends(current_user_id), ): await db.workflow_runs.require_owner(run_id=run_id, user_id=user_id) run = workflow.get_run(run_id, type=str) return {"status": await run.status()} ``` Each call to the start endpoint creates a new run. In a signup integration, deduplicate repeated signup events and persist the run mapping as part of your application's job-dispatch process. Step retries do not deduplicate separate calls to start a workflow. _Vercel detects a FastAPI_ `app` _in_ `app/main.py` _with no extra configuration._ Save the run ID to retrieve its handle with `workflow.get_run()`. Set `type=str` so the SDK validates this workflow's result as a string. The status is one of `pending`, `running`, `completed`, `failed`, or `cancelled`. Awaiting the return value waits for completion, so use status polling for this three-day workflow instead of holding the original HTTP request open. ## Step 5: Handle retries and permanent failures By default, a step can retry three times after its initial attempt. Ordinary exceptions use that retry budget, while `FatalError` ends the step immediately. Use retries for transient failures such as network errors; fail immediately when retrying cannot correct the input. Choose the error type according to whether another attempt can succeed: | Raise | What happens | Use it for | | ------------------------------------------ | ----------------------------------- | ----------------------------------------- | | Any other exception | Retries with the default timing | Unexpected errors | | `RetryableError("...", retry_after="30s")` | Retries, but not before the delay | Rate limits, `503`s, timeouts | | `FatalError("...")` | Fails the step now, with no retries | Missing records, bad input, auth failures | Replace `app/steps/onboarding.py` with the following version. It keeps all three steps and reads the email provider's retry delay when present: ```python from datetime import timezone from email.utils import parsedate_to_datetime from vercel.workflow import FatalError, RetryableError, get_step_metadata from app.workflow import wf from app.clients import db, email def retry_delay(response): value = response.headers.get("Retry-After", "") if value.isdigit(): return int(value) try: retry_at = parsedate_to_datetime(value) except (TypeError, ValueError, OverflowError): return 30 return retry_at if retry_at.tzinfo else retry_at.replace(tzinfo=timezone.utc) @wf.step async def create_profile(*, user_id: str) -> dict: user = await db.users.get(user_id) if user is None: raise FatalError(f"User {user_id} not found") await db.profiles.upsert(user_id=user_id) return {"email": user.email} @wf.step(max_retries=5) async def send_email(*, to: str, template: str) -> None: step = get_step_metadata() response = await email.send( to=to, template=template, idempotency_key=step.step_id, ) if response.status_code == 429: raise RetryableError( "Email provider rate limit", retry_after=retry_delay(response), ) response.raise_for_status() @wf.step async def has_finished_setup(*, user_id: str) -> bool: profile = await db.profiles.get(user_id) if profile is None: raise FatalError(f"Profile for {user_id} not found") return profile.setup_complete ``` The email step allows five retries after its initial attempt. Its idempotency key remains unchanged across those attempts. The helper honors a valid `Retry-After` header and falls back to 30 seconds when the header is absent or invalid. Raising `RetryableError` changes the next attempt's timing without increasing the retry budget. Classify other provider errors in your adapter according to the provider's API contract. The SDK also exposes execution metadata and cancellation controls: - Step metadata includes the run ID, stable step ID, attempt number, and the first attempt's start time. Use it for logging and elapsed-time checks. - Mark a step with `cancellable=True` when the workflow needs to request cancellation. The deadline example below explains how the step must cooperate. ## Step 6: Run it locally, then deploy ### Run it locally Configure your application's database and email credentials for local development, seed a test user, and use your authentication flow to obtain that user's session or token. Start the app with the Vercel CLI: ```bash vercel dev ``` The local backend stores workflow records in `.workflow-data/`. Add that directory to `.gitignore` because it contains workflow inputs and results. Set the delay to 30 seconds for this test. If your adapter uses bearer tokens, start the run from another terminal with this authenticated request: ```bash curl --fail-with-body -X POST http://localhost:3000/api/onboarding \ -H "Authorization: Bearer $APP_TEST_TOKEN" ``` The response contains a run ID. Use it with the status endpoint to confirm that the run moves to a completed or failed state. ### Watch the run Open the workflow web UI from the project directory in a second terminal: ```bash npx workflow web ``` Inspect the run's step results and its scheduled wait. To test retries without sending real email, make your email stub return a rate-limit response once, then succeed. Confirm that the next attempt uses the same idempotency key and that a missing user ends the run without repeated retries. Keep the local server running because pending deliveries stay in memory even though event records persist on disk. Restarting the whole server can lose those deliveries and is not a reliable recovery test. For terminal inspection, list runs and inspect the ID you received: ```bash npx workflow inspect runs npx workflow inspect run "$RUN_ID" ``` ### Deploy to Vercel Restore the three-day delay and commit the project files and lockfile. In the project's Environment Variables settings, enable access to System Environment Variables so the runtime can identify the Vercel deployment. Follow the [Vercel World setup](https://workflow-sdk.dev/worlds/vercel) to confirm Fluid compute is enabled. Configure the database and email credentials in Preview, then deploy: ```bash vercel deploy ``` Vercel installs the locked dependencies, detects FastAPI, and registers the configured workflows. Test the preview with a test account and email provider before using production credentials. Check that unauthenticated requests fail and that another user cannot read the run. Hosted runs use managed storage and queues with separate run data for each environment. View the preview in the Vercel dashboard or hosted Workflow UI: ```bash npx workflow web --backend vercel ``` After the preview checks pass, configure the Production credentials and deploy: ```bash vercel deploy --prod ``` ## Use cases The following extensions add approval events, deadlines, and progress updates. Each example belongs in its own module under `app/workflows/`. Import any module you use from `app/workflows/__init__.py` so its workflows and steps register with the same registry. | Task | Workflow feature | | ----------------------------------------- | -------------------------- | | Retry a failed model call | Steps with bounded retries | | Wait for a person or an external callback | Hooks | | Schedule a follow-up | Durable sleep | | Request cancellation after a deadline | Cancellable steps | | Show progress during a run | Typed streams | ### AI agents with human approval Define the approval workflow in `app/workflows/approval.py`. The content module supplies your model-call and publishing steps, each registered with the same registry. Publishing must be safe to retry. Before starting a run, create a unique approval ID and store its association with the draft and authorized reviewers. Pass that ID to the workflow and never reuse it for another review cycle: ```python import typing import pydantic from vercel import workflow from app.workflow import wf from app.steps.content import generate_draft, publish_draft, refine_draft class Approval(pydantic.BaseModel, workflow.BaseHook): decision: typing.Literal["approved", "changes"] notes: str | None = None @wf.workflow async def review_draft(*, topic: str, approval_id: str) -> None: draft = await generate_draft(topic=topic) async for event in Approval.wait(token=approval_id): if event.decision == "approved": await publish_draft(draft=draft) break draft = await refine_draft(draft=draft, notes=event.notes) ``` The hook waits for decisions sent to the approval ID. Requesting changes triggers another draft and another wait; approval triggers publication. Add this route to the existing FastAPI app in `app/main.py`. Your approvals adapter must reject unauthorized reviewers and closed review cycles before returning the stored approval ID: ```python from fastapi import Depends from app.auth import current_user_id from app.clients import db from app.workflows.approval import Approval @app.post("/api/drafts/{draft_id}/review") async def review( draft_id: str, approval: Approval, user_id: str = Depends(current_user_id), ): record = await db.approvals.require_reviewer( draft_id=draft_id, user_id=user_id, ) hook = await approval.resume(record.approval_id) return {"run_id": hook.run_id} ``` FastAPI validates the decision payload, while the application checks who may approve the draft. Handle duplicate submissions in the approval record so a retry cannot become an extra review event. For a payment provider or Slack callback, verify the provider's signature and deduplicate its event ID before resuming a hook. Model-call steps can use the [AI SDK for Python](https://ai-python.dev/). ### Long-running jobs with a deadline Save this example in `app/workflows/reports.py` and import it in the registry package. Your app must supply the reporting adapter. Mark the step as cancellable so a ten-minute timeout can request that it stop: ```python import asyncio from datetime import timedelta from app.workflow import wf from app.clients import reporting_service @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"} ``` The workflow waits for the step to terminate before continuing. If the step suppresses cancellation, it can still return normally. Let cancellation propagate after cleanup, and use the provider's cancellation API if the step started a remote job that must also stop. ### Live progress in your UI Save this in `app/workflows/streaming.py` and import it in the registry package. Its step writes typed progress messages for server-side readers: ```python import pydantic from vercel import workflow from app.workflow import wf 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) ``` The workflow passes the stream to a step that writes messages. Check the caller's access before reading the run's messages on the server: ```python from vercel import workflow from app.workflows.streaming import Progress async def read_progress(run_id: str): run = workflow.get_run(run_id) async for progress in run.readable(type=Progress): print(progress.message) ``` The final successful writer closes the stream so readers know it has ended. If the workflow fails before that point, your UI should use the run's terminal status to stop waiting. Closing the stream unconditionally during a failed attempt would prevent a retry from continuing it. Retries can write the same progress message again. For production events, include a stable event ID so consumers can deduplicate repeated updates. ### When to use something else Use an ordinary function for work that fits in the request and needs no durable recovery. Choose [Vercel Queues](https://vercel.com/docs/queues) for independent message consumers. For a fixed schedule, use [Cron Jobs](https://vercel.com/docs/cron-jobs). The cron route can start a workflow when each scheduled execution needs durable steps or a later follow-up. ## Next steps - Use the [Python Workflow reference](https://workflow-sdk.dev/docs/getting-started/python) for deterministic helpers, hooks, and stream namespaces. - Review [Vercel World configuration](https://workflow-sdk.dev/worlds/vercel) for hosted execution and environment settings. - Follow [How to ship a FastAPI app on Vercel](https://vercel.com/kb/guide/ship-a-fastapi-app-on-vercel) for the surrounding application deployment. - Explore the [notebook app](https://github.com/vercel-labs/notebook-factory) for a Python application that runs its assistant in Vercel Workflows. ## Related Documentation - [Vercel Workflows](/docs/workflow) ## More Vercel Workflows guides - [Run dynamic code in a durable workflow](/kb/guide/run-dynamic-code-in-a-durable-workflow): Run dynamic JavaScript with Run SDK and Workflow SDK. Pause a local refund demo for human approval and resume it after a server restart. - [Running Durable Agents on Vercel](/kb/guide/durable-ai-agents): Design durable AI agents on Vercel with patterns for execution recovery, human approval, safe retries, deployment changes, and persistent files. - [Run recurring security reviews with deepsec on Vercel](/kb/guide/deepsec-reviews-on-vercel): Learn how to run periodic security scans on your code with deepsec. Set up deepsec to review GitHub repositories and deliver findings to GitHub and Slack using Vercel Connect, Workflows, Sandbox, and AI Gateway.