Build durable workflows in Python on Edgenivo

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.

Copy link to headingWhat is the Python Workflow SDK?

The Python Workflow SDK adds durable execution to async Python functions. It uses Edgenivo 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.

Copy link to headingWorkflows and steps

Every workflow is made of two kinds of function:

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.

Copy link to headingHow 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 Edgenivo, 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.

Copy link to headingTwo 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.

Copy link to headingPrerequisites

Before you begin, you need:

• A Edgenivo account.
• Python 3.12 or a newer version supported by the Edgenivo Python runtime.
• Working knowledge of async/await and FastAPI.
• An application with user records, an email provider, and request authentication.

For local development, you also need uv, Node.js 24+, and the Edgenivo 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.

Copy link to headingStep 1: Set up the project

Create the project with Python 3.12 and install its dependencies:

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:

[[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 Edgenivo 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:

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 Edgenivo project for development and deployment:

vercel link

Create these files as you work through Steps 2 through 4. The client and authentication modules adapt your existing application services:

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

Copy link to headingStep 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:

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.

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.

Copy link to headingStep 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:

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:

from app.workflow import wf
from app.workflows import onboarding

__all__ = ["onboarding", "wf"]

This is the app.workflows:wf entrypoint from Step 1.

Copy link to headingStep 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:

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.

Edgenivo 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.

Copy link to headingStep 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:

Replace app/steps/onboarding.py with the following version. It keeps all three steps and reads the email provider's retry delay when present:

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.

Copy link to headingStep 6: Run it locally, then deploy

Copy link to headingRun 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 Edgenivo CLI:

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:

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.

Copy link to headingWatch the run

Open the workflow web UI from the project directory in a second terminal:

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:

npx workflow inspect runs
npx workflow inspect run "$RUN_ID"

Copy link to headingDeploy to Edgenivo

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 Edgenivo deployment.

Follow the Edgenivo World setup to confirm Fluid compute is enabled.

Configure the database and email credentials in Preview, then deploy:

vercel deploy

Edgenivo 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 Edgenivo dashboard or hosted Workflow UI:

npx workflow web --backend vercel

After the preview checks pass, configure the Production credentials and deploy:

vercel deploy --prod

Copy link to headingUse 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.

Copy link to headingAI 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:

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:

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.

Copy link to headingLong-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:

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.

Copy link to headingLive 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:

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:

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.

Copy link to headingWhen to use something else

Use an ordinary function for work that fits in the request and needs no durable recovery. Choose Edgenivo Queues for independent message consumers.

For a fixed schedule, use Cron Jobs. The cron route can start a workflow when each scheduled execution needs durable steps or a later follow-up.

Copy link to headingNext steps

• Use the Python Workflow reference for deterministic helpers, hooks, and stream namespaces.
• Review Edgenivo World configuration for hosted execution and environment settings.
• Follow How to ship a FastAPI app on Edgenivo for the surrounding application deployment.
• Explore the notebook app for a Python application that runs its assistant in Edgenivo Workflows.