Callback Receiver Example

FastAPI example implementation for receiving QTG v3 callbacks

Callback Receiver Example (FastAPI, v3)

The example below is a minimal sample that receives and verifies the v3 callback sent by qtg.

Runnable example files:

• Receiver server: examples/callback_receiver_fastapi.py
• Local test sender: examples/send_signed_callback.py

from fastapi import FastAPI, HTTPException, Request, status

from qtg.callback_auth import verify_callback

app = FastAPI()

CALLBACK_SECRET = "your-callback-secret"


def record_nonce(nonce: str) -> bool:
    # In production, a shared DB + unique constraint is recommended.
    ...


@app.post("/qtg/callback")
async def qtg_callback(request: Request):
    body = await request.body()
    headers = {k.lower(): v for k, v in request.headers.items()}

    ok, reason = verify_callback(
        headers=headers,
        body_bytes=body,
        secret=CALLBACK_SECRET,
        max_age_seconds=300,
        nonce_recorder=record_nonce,
    )
    if not ok:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"invalid callback signature: {reason}",
        )

    payload = await request.json()
    return {"ok": True, "movement_id": payload.get("movement_id")}

Delivered Headers (v3)

• X-QTG-Callback-Timestamp
• X-QTG-Callback-Nonce
• X-QTG-Callback-Signature-Version: v3
• X-QTG-Callback-Signature

Signature Rule (v3)

Canonical string:

{timestamp}
{nonce}
{sha256(body_bytes)}

signature:

HMAC_SHA256(callback_hmac_secret, canonical)

Local Run Quickstart

1. Run the receiver server

MG_CALLBACK_HMAC_SECRET=dev-secret \
uv run uvicorn examples.callback_receiver_fastapi:app --reload --port 8200

2. Send a signed callback

MG_CALLBACK_HMAC_SECRET=dev-secret \
uv run python examples/send_signed_callback.py \
  --url http://localhost:8200/qtg/callback \
  --movement-id sample-1 \
  --state COMPLETED

Important Notes

• This example shows a minimal nonce ledger.
• If the receiver runs in multiple instances in production, you must use nonce recording based on a shared DB/store + unique constraint.