Backend em FastAPI

A rota que emite a credencial e o receptor de webhook, num servidor Python com FastAPI.

A mesma integração do Express, em Python. O que muda é como o framework entrega o corpo cru ao handler de webhook.

Stack	Python 3.12 · FastAPI · httpx
Você terá	uma rota que emite a credencial e um endpoint de webhook com assinatura conferida

Confira o exemplo

Os arquivos desta POC, com o destino de cada um.

Emitir a credencial

main.py

import os
import httpx
from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()
API = "https://api.legitimuz.com/public/verifications"

@app.post("/api/verifications")
async def create_verification(user=Depends(authenticate)):
    registration = await db.registration_by_user(user.id)

    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.post(
            API,
            headers={
                "X-API-Key": os.environ["LEGITIMUZ_API_KEY"],
                "Content-Type": "application/json",
            },
            json={
                "schema_version": "1.0",
                "ref_id": registration.id,
                "document": {"type": "cpf", "number": registration.cpf},
                "flow_public_id": os.environ["LEGITIMUZ_FLOW_ID"],
            },
        )

    if response.is_error:
        raise HTTPException(status_code=502, detail="legitimuz_unavailable")

    data = response.json()
    await db.link(registration.id, data["verification"]["public_id"])

    # Só a `entry` volta ao cliente. O resto fica no servidor.
    return {"entry": data["entry"]}

Receber o desfecho

webhooks.py

import hashlib
import hmac
import json
import os
import time

from fastapi import Request, Response

TOLERANCE_SECONDS = 300

def is_signature_valid(raw_body: bytes, header: str, secret: str) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    try:
        timestamp = int(parts["t"])
        received = parts["v1"]
    except (KeyError, ValueError):
        return False

    if abs(int(time.time()) - timestamp) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(
        secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected, received)

@app.post("/api/webhooks/legitimuz")
async def receive(request: Request):
    # `await request.body()` devolve os BYTES originais. Um modelo Pydantic aqui já reserializaria
    # o JSON, e a assinatura passaria a falhar sempre.
    raw_body = await request.body()
    header = request.headers.get("x-legitimuz-signature", "")

    if not is_signature_valid(raw_body, header, os.environ["LEGITIMUZ_WEBHOOK_SECRET"]):
        return Response(status_code=401)

    delivery = request.headers.get("x-legitimuz-delivery", "")
    if not await db.record_if_new(delivery):
        return Response(status_code=200)

    await queue.publish(json.loads(raw_body))

    return Response(status_code=200)

Não declare o corpo como modelo Pydantic na rota de webhook. O FastAPI faria o parse e o request.body() viria vazio ou reserializado, e a conferência da assinatura falharia em toda entrega.

Antes de rodar

.env

LEGITIMUZ_API_KEY=<SUA_CHAVE>
LEGITIMUZ_WEBHOOK_SECRET=<SEGREDO_DO_ENDPOINT>
LEGITIMUZ_FLOW_ID=<FLOW_PUBLIC_ID>

Variável	Onde achar
LEGITIMUZ_API_KEY	Integrações → Segurança → Chaves de API. Aparece uma vez
LEGITIMUZ_WEBHOOK_SECRET	Integrações → Segurança → Webhooks, na criação do endpoint
LEGITIMUZ_FLOW_ID	Solução KYC → Fluxos, no menu da linha, em Copiar ID do fluxo

Use uma integração sandbox. Nenhum dos três valores vai para o browser ou para o app.

O que esta POC não faz

POC é código para entender o fluxo, não para copiar em produção. Em todas elas, authenticate() é um stub, db é um objeto de mentira e não há migration, observabilidade nem retentativa própria.

• authenticate é uma dependência de mentira; troque pela sua sessão real.
• queue é um stub. Veja a POC de fila e worker.
• Sem retry na chamada de criação. O catálogo de quando repetir está em erros.

Próximo passo

Fila e worker

Processar o desfecho fora do request.

Segurança dos webhooks

A conferência em quatro linguagens.