Python

import os, json, requests

VERIFY_AI_KEY = os.environ["VERIFY_AI_KEY"]
VERIFY_AI_BASE = "https://verify.switchlabs.dev/api"

with open("scooter.jpg", "rb") as f:
  res = requests.post(
      f"{VERIFY_AI_BASE}/v1/verify",
      headers={"X-API-Key": VERIFY_AI_KEY},
      files={"image": f},
      data={
          "policy": "scooter_parking",
          "metadata": json.dumps({"user_id": "usr_456", "trip_id": "trip_123"}),
      },
      timeout=30,
  )

res.raise_for_status()
result = res.json()
print(result["is_compliant"], result["category"], result["violation_reasons"])

import hashlib
import hmac
import json
import os
import time
import psycopg  # pip install "psycopg[binary]"

# Create this PostgreSQL table before starting the receiver:
# CREATE TABLE verifyai_webhook_inbox (
#   delivery_id text PRIMARY KEY, event text NOT NULL, payload jsonb NOT NULL,
#   received_at timestamptz NOT NULL DEFAULT now(), processed_at timestamptz
# );
# A worker polls rows WHERE processed_at IS NULL and marks them only after
# successful processing. Retry failed work; make business effects idempotent.
# If forwarding to another queue, mark processed only after durable enqueue.
# Never store a duplicate receipt separately from the work obligation.

from flask import Flask, abort, request

app = Flask(__name__)
SIGNING_SECRET = os.environ["VERIFY_AI_WEBHOOK_SECRET"]  # whsec_...


def verify_verifyai_signature(raw_body: bytes, signature_header: str, secret: str, tolerance_seconds: int = 300) -> bool:
    """X-VerifyAI-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256 of b"<t>." + raw body>."""
    parts = dict(
        part.strip().split("=", 1) for part in (signature_header or "").split(",") if "=" in part
    )
    t, v1 = parts.get("t", ""), parts.get("v1", "")
    if not (t.isascii() and t.isdigit()) or len(v1) != 64:
        return False
    # A timestamp more than five minutes from now is a replay: reject it.
    if abs(time.time() - int(t)) > tolerance_seconds:
        return False
    expected = hmac.new(secret.encode(), t.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.encode(), v1.lower().encode())


@app.post("/webhooks/verifyai")
def verifyai_webhook():
    raw_body = request.get_data()  # the exact bytes that were signed; don't parse first
    if not verify_verifyai_signature(raw_body, request.headers.get("X-VerifyAI-Signature", ""), SIGNING_SECRET):
        abort(401)
    delivery_id = request.headers.get("X-VerifyAI-Delivery")
    if not delivery_id:
        abort(400)
    try:
        payload = json.loads(raw_body)
    except ValueError:
        abort(400)
    event, data = payload["event"], payload["data"]
    if event == "webhook.test":
        return "", 200
    try:
        # The connection context commits before returning 200; errors roll back.
        with psycopg.connect(os.environ["DATABASE_URL"]) as db:
            db.execute("INSERT INTO verifyai_webhook_inbox (delivery_id, event, payload) VALUES (%s, %s, %s::jsonb) ON CONFLICT (delivery_id) DO NOTHING", (delivery_id, event, json.dumps(data)))
    except psycopg.Error:
        abort(503)  # no acknowledged delivery without a durable work obligation
    return "", 200