Build a verification form with FastAPI

python3 -m venv .venv
source .venv/bin/activate
python -m pip install messagebird-sdk "fastapi[standard]"
export BIRD_API_KEY="your-server-api-key"
export DEMO_VERIFY_EMAIL="you@example.com"
export APP_ORIGIN="http://127.0.0.1:8000"
export APP_SECRET="your-generated-application-secret"

import hashlib
import hmac
import json
import os
import re
import sqlite3
import threading
import time
import uuid
from contextlib import asynccontextmanager, closing
from pathlib import Path

from bird import APIStatusError, Bird
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse
from starlette.concurrency import run_in_threadpool

DB = os.environ.get("VERIFY_STATE_DB", "verify-state.sqlite3")
ORIGIN = os.environ["APP_ORIGIN"]
SECRET = os.environ["APP_SECRET"].encode()
DESTINATION = os.environ["DEMO_VERIFY_EMAIL"]
FORM = Path("verify-form.html").read_text()
LOCK = threading.Lock()
os.umask(0o077)


@asynccontextmanager
async def lifespan(app):
    with Bird(api_key=os.environ["BIRD_API_KEY"], max_retries=0) as bird:
        app.state.bird = bird
        yield


app = FastAPI(lifespan=lifespan)


def database():
    db = sqlite3.connect(DB, timeout=5)
    db.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, data TEXT NOT NULL)")
    db.commit()
    return db


def read(db, token):
    row = db.execute("SELECT data FROM sessions WHERE id = ?", (token,)).fetchone()
    return json.loads(row[0]) if row else None


def save(db, token, attempt):
    db.execute("INSERT OR REPLACE INTO sessions VALUES (?, ?)", (token, json.dumps(attempt)))
    db.commit()


def response(body, status=200):
    return JSONResponse(body, status_code=status, headers={"Cache-Control": "no-store"})


def is_uuid(value):
    return isinstance(value, str) and re.fullmatch(
        r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", value
    ) is not None


@app.get("/verify", response_class=HTMLResponse)
def show(request: Request):
    with LOCK, closing(database()) as db:
        token = request.cookies.get("verify_session")
        attempt = read(db, token) if is_uuid(token) else None
        if not attempt or attempt["expires"] <= time.time():
            rows = db.execute("SELECT id, data FROM sessions").fetchall()
            for old_id, data in rows:
                if json.loads(data)["expires"] <= time.time():
                    db.execute("DELETE FROM sessions WHERE id = ?", (old_id,))
            db.commit()
            if db.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] >= 100:
                return response({"error": "Local session limit reached."}, 503)
            token = str(uuid.uuid4())
            attempt = {"to": {"email": DESTINATION}, "action": "demo-signup",
                       "expires": time.time() + 3600, "operations": {}}
            save(db, token, attempt)
    result = HTMLResponse(FORM, headers={"Cache-Control": "no-store"})
    result.set_cookie("verify_session", token, max_age=3600, httponly=True,
                      samesite="strict", secure=ORIGIN.startswith("https://"))
    return result


def execute(bird, token, key, data):
    with LOCK, closing(database()) as db:
        attempt = read(db, token) if is_uuid(token) else None
        if not attempt or attempt["expires"] <= time.time():
            return response({"error": "Open a fresh demo session."}, 401)
        if attempt.get("completed"):
            return response({"resolved": True, "completed": True, "action": attempt["action"]})
        if not is_uuid(key):
            return response({"resolved": True, "error": "Supply an Idempotency-Key UUID."}, 400)
        operation = data.get("operation") if isinstance(data, dict) else None
        if operation not in ("start", "check", "resend", "next"):
            return response({"resolved": True, "error": "Choose a supported operation."}, 400)
        code = data.get("code", "") if operation == "check" else ""
        if not isinstance(code, str) or (operation == "check" and not re.fullmatch(r"[0-9]{4,8}", code)):
            return response({"resolved": True, "error": "Enter the 4–8 digit code."}, 400)
        if operation != "start" and not attempt.get("verification_id"):
            return response({"resolved": True, "error": "Start this verification first."}, 409)
        digest = hmac.new(SECRET, json.dumps([operation, code]).encode(), hashlib.sha256).hexdigest()
        prior = attempt["operations"].get(key)
        if prior and not hmac.compare_digest(prior["digest"], digest):
            return response({"error": "This key belongs to a different request."}, 409)
        if prior and "response" in prior:
            return response(prior["response"]["body"], prior["response"]["status"])
        if attempt.get("pending") not in (None, key):
            return response({"error": "Resolve the earlier request before changing the step or code."}, 409)
        if not prior and len(attempt["operations"]) >= 50:
            return response({"resolved": True, "error": "Local attempt limit reached."}, 429)
        attempt["operations"][key] = {"digest": digest}
        attempt["pending"] = key
        save(db, token, attempt)
        status = 200
        try:
            options = {"idempotency_key": key}
            if operation == "check":
                result = bird.verify.verifications.check(to=attempt["to"], code=code, options=options)
                value = result.model_dump(mode="json")
                if value["verification"]["id"] != attempt["verification_id"]:
                    status = 409
                    body = {"resolved": True, "error": "The result belongs to a different verification."}
                else:
                    attempt["completed"] = value["success"] and value["verification"]["status"] == "verified"
                    body = {"resolved": True, "completed": attempt["completed"],
                            "action": attempt["action"], "reason": value.get("reason"),
                            "attempts_remaining": value.get("attempts_remaining"),
                            "verification_status": value["verification"]["status"]}
            else:
                method = bird.verify.verifications.next_channel if operation == "next" else bird.verify.verifications.create
                result = method(to=attempt["to"], options=options)
                value = result.model_dump(mode="json")
                attempt["verification_id"] = value["id"]
                body = {"resolved": True, "completed": False, "verification_status": value["status"],
                        "expires_at": value["expires_at"], "last_channel": value.get("last_channel")}
        except APIStatusError as error:
            if error.status_code not in (404, 422, 429):
                return response({"error": "Outcome unknown. Retry the same key, step and code."}, 503)
            status = error.status_code
            messages = {404: "No active verification found. Review the earlier outcome; this does not prove a failed code.",
                        422: "The request or next channel was refused; review your configuration.",
                        429: "Wait before requesting another code or check."}
            body = {"resolved": True, "error": error.code, "message": messages[status]}
        except Exception:
            return response({"error": "Outcome unknown. Retry the same key, step and code."}, 503)
        attempt["operations"][key]["response"] = {"status": status, "body": body}
        attempt.pop("pending", None)
        save(db, token, attempt)
        return response(body, status)


@app.post("/verify")
async def submit(request: Request):
    if request.headers.get("origin") != ORIGIN:
        return response({"error": "Origin refused."}, 403)
    raw = bytearray()
    async for chunk in request.stream():
        raw.extend(chunk)
        if len(raw) > 8192:
            return response({"resolved": True, "error": "Request too large for this form."}, 413)
    try:
        data = json.loads(raw)
    except (ValueError, UnicodeDecodeError):
        return response({"resolved": True, "error": "Invalid JSON."}, 400)
    return await run_in_threadpool(execute, request.app.state.bird,
                                  request.cookies.get("verify_session"),
                                  request.headers.get("idempotency-key"), data)

<!doctype html>
<html lang="en">
  <meta charset="utf-8" /><meta
    name="viewport"
    content="width=device-width, initial-scale=1"
  /><title>Verify a destination · Bird example</title>
  <style>
    body {
      font: 16px/1.6 system-ui;
      max-width: 42rem;
      margin: 8vh auto;
      padding: 24px;
      color: #173c30;
      background: #fafaf7;
    }
    button,
    input,
    select {
      font: inherit;
      padding: 12px;
      border: 1px solid #bbcbbf;
      border-radius: 8px;
      margin: 6px;
    }
    pre {
      white-space: pre-wrap;
      padding: 20px;
      background: #eaf0e9;
      border-radius: 12px;
    }
  </style>
  <main>
    <h1>Verify your destination.</h1>
    <p>
      This local example sends a real verification to the destination configured on the server. It
      only completes a demo application action.
    </p>
    <form>
      <label
        >Step
        <select name="operation">
          <option value="start">Send a code</option>
          <option value="check">Check code</option>
          <option value="resend">Resend</option>
          <option value="next">Next channel</option>
        </select></label
      ><label>Code <input name="code" inputmode="numeric" autocomplete="one-time-code" /></label
      ><button>Continue</button>
    </form>
    <pre role="status">Ready</pre>
  </main>
  <script>
    const form = document.querySelector("form"),
      output = document.querySelector("pre");
    let attempt;
    form.addEventListener("submit", async (e) => {
      e.preventDefault();
      const body = {
        operation: form.elements.operation.value,
        code: form.elements.operation.value === "check" ? form.elements.code.value : "",
      };
      const fingerprint = JSON.stringify(body);
      if (attempt && attempt.fingerprint !== fingerprint) {
        output.textContent = "Resolve the previous request with the same step and code first.";
        return;
      }
      attempt ??= { fingerprint, key: crypto.randomUUID() };
      form.querySelector("button").disabled = true;
      try {
        const r = await fetch(location.pathname, {
          method: "POST",
          headers: { "Content-Type": "application/json", "Idempotency-Key": attempt.key },
          body: attempt.fingerprint,
        });
        const data = await r.json();
        output.textContent = JSON.stringify(data, null, 2);
        if (data.resolved) attempt = undefined;
      } catch {
        output.textContent = "Outcome unknown. Retry the same step and code without reloading.";
      } finally {
        form.querySelector("button").disabled = false;
      }
    });
  </script>
</html>

uvicorn verify_app:app --host 127.0.0.1 --port 8000 --workers 1