Integrating with FastAPI

A receipt email can take longer than an HTTP request should wait, and the work may need to survive a restart of your API server. Put the send in a Waymark workflow, then let the FastAPI route either await the result or return a queued workflow id.

This is useful when work you've put in FastAPI background tasks needs persistence, retry policies, or a run history.

Project layout

Keep the workflow in a module both processes can import. The FastAPI app imports it to start runs; the workers spawned by waymark-executor import it to execute the actions.

app/
  api.py
  models.py
  workflows.py

Define the contract

The route and the action need the order id and recipient email. Use Pydantic models to define the request and the result they share.

# app/models.py
from pydantic import BaseModel

class ReceiptRequest(BaseModel):
    order_id: str
    email: str

class ReceiptResult(BaseModel):
    order_id: str
    email_id: str

FastAPI validates the incoming request, and Waymark can serialize the same model through Postgres to the worker.

Write the workflow

The email provider call is the slow, flaky part, so it belongs in an action. Here we call Resend's HTTP API directly. This does network I/O, raises on provider errors, and returns the provider's email id.

# app/workflows.py
import os

import httpx
from waymark import action

from app.models import ReceiptRequest, ReceiptResult

@action
async def send_receipt_email(request: ReceiptRequest) -> ReceiptResult:
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.post(
            "https://api.resend.com/emails",
            headers={
                "Authorization": f"Bearer {os.environ['RESEND_API_KEY']}",
                "User-Agent": "waymark-fastapi-tutorial/1.0",
            },
            json={
                "from": os.environ["RECEIPTS_FROM_EMAIL"],
                "to": [request.email],
                "subject": f"Receipt for order {request.order_id}",
                "html": f"<p>Your receipt for order {request.order_id} is ready.</p>",
            },
        )
    response.raise_for_status()
    data = response.json()
    return ReceiptResult(order_id=request.order_id, email_id=data["id"])

The workflow gives the API a durable entrypoint for the send.

from waymark import Workflow, workflow

@workflow
class ReceiptWorkflow(Workflow):
    async def run(self, request: ReceiptRequest) -> ReceiptResult:
        return await send_receipt_email(request)

This workflow makes one action call. Waymark records its outcome; a provider error fails the workflow unless you add a retry policy.

Await short work

If the workflow is short enough for a normal HTTP request budget, await it and return the result directly.

# app/api.py
from fastapi import FastAPI

from app.models import ReceiptRequest, ReceiptResult
from app.workflows import ReceiptWorkflow

app = FastAPI()

@app.post("/receipts", response_model=ReceiptResult)
async def send_receipt(request: ReceiptRequest) -> ReceiptResult:
    return await ReceiptWorkflow().run(request)

The route waits while a Waymark worker sends the email. If the workflow fails, run() raises WorkflowFailedError at the route's call site.

Queue long work

If the work might take longer than a client should wait, return as soon as the workflow is queued. Use this for imports, report generation, slow third-party APIs, and agent runs.

from pydantic import BaseModel

class QueuedReceipt(BaseModel):
    workflow_instance_id: str

Pass _blocking=False to get the workflow id without waiting for the email.

@app.post("/receipts/async", response_model=QueuedReceipt, status_code=202)
async def queue_receipt(request: ReceiptRequest) -> QueuedReceipt:
    instance_id = await ReceiptWorkflow().run(request, _blocking=False)
    return QueuedReceipt(workflow_instance_id=instance_id)

FastAPI BackgroundTasks run in the web process, so unfinished work can be lost when that process stops. Here, Postgres holds the queued workflow for a Waymark node to pick up.

Runtime setup

The Resend action uses httpx, so install it with Waymark and your web dependencies if it is not already in your app.

uv add waymark fastapi uvicorn httpx

Your API process and waymark-executor need the same database, and the node's workers need to be able to import app.workflows: start waymark-executor from the project root, where the app package lives, or install the project into its environment. WAYMARK_USER_MODULE has the workers import it at startup, so an import error stops waymark-executor from starting, after its 15 second worker attach timeout, instead of failing the first receipt.

export WAYMARK_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/waymark
export WAYMARK_USER_MODULE=app.workflows
export RESEND_API_KEY=re_xxxxxxxxx
export RECEIPTS_FROM_EMAIL="Acme <onboarding@resend.dev>"

Run the node separately from the web server.

uv run waymark-executor

Then run FastAPI however you normally do.

uvicorn app.api:app --reload

The API process creates workflow runs. The node picks them up from Postgres, runs the workflow, and dispatches each action to one of its workers, which import app.workflows to run it.

Production habits

  • Keep HTTP-only concerns in FastAPI: auth, headers, status codes, request parsing, and response models.
  • Keep slow I/O in actions: email, billing, imports, file processing, model calls, and anything that needs retries.
  • Keep workflow inputs and outputs as Pydantic models when they are more than a couple of scalars.
  • Keep WAYMARK_USER_MODULE pointed at the modules that define your actions, so a broken import stops waymark-executor at startup instead of failing the first action.
  • Use _blocking=False when a client does not need the final answer in the same request.

Before sending real receipts, make the email action idempotent: a node failure can cause an in-flight action to run again. See Production Deployment for recovery behavior.