API Pix no FastAPI: httpx assíncrono e webhook assinado

No FastAPI, a API Pix combina com o httpx, que chama a API com await sem travar o event loop. Uma rota cria a cobrança e devolve o QR Code, outra recebe o webhook, valida a assinatura e deixa o trabalho para depois da resposta com BackgroundTasks. O código segue a documentação da API, usa Python 3.10 ou superior e a base https://app.xflowpayments.com/api/v1. O mesmo fluxo síncrono, com requests e conciliação, está em API Pix em Python, e a visão geral, no guia da API Pix.

O que você precisa

Uma chave gerada no painel, em Desenvolvedores > Chaves de API. A chave pública (pk_live_...) fica sempre visível. O segredo (sk_live_...) aparece uma única vez.
Para testar, uma chave do ambiente Teste (pk_test_... e sk_test_...), que pode ser criada antes da aprovação do cadastro.
pip install fastapi httpx uvicorn.
As variáveis de ambiente XFLOW_PUBLIC_KEY, XFLOW_SECRET_KEY e XFLOW_WEBHOOK_SECRET.

Um AsyncClient para a aplicação inteira

A autenticação é HTTP Basic, e no httpx basta auth=(chave_publica, segredo). Valores vão em centavos, como inteiros. Criar um cliente por requisição desperdiça conexões, então ele nasce no lifespan e fica em app.state:

python

# app/xflow.py
import os

import httpx


class XflowErro(Exception):
    def __init__(self, resp: httpx.Response):
        corpo = resp.json() if resp.content else {}
        self.status = resp.status_code
        self.code = corpo.get("error")
        self.request_id = corpo.get("request_id")
        self.retry_after = int(resp.headers.get("retry-after") or 0)
        super().__init__(f"{self.status} {self.code} ({self.request_id})")


def novo_cliente() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url="https://app.xflowpayments.com/api/v1",
        auth=(os.environ["XFLOW_PUBLIC_KEY"], os.environ["XFLOW_SECRET_KEY"]),
        timeout=30,
    )


async def xflow(cliente: httpx.AsyncClient, method: str, path: str, json=None, idempotency_key=None):
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    resp = await cliente.request(method, path, json=json, headers=headers)
    if resp.is_error:
        raise XflowErro(resp)
    return resp.json()

python

# app/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from .xflow import novo_cliente


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with novo_cliente() as cliente:
        app.state.xflow = cliente
        yield


app = FastAPI(lifespan=lifespan)

Criar a cobrança

python

from fastapi import Depends, HTTPException, Request

from .xflow import xflow


@app.post("/pedidos/{pedido_id}/pix")
async def gerar_pix(pedido_id: int, request: Request, usuario=Depends(usuario_logado)):
    pedido = await buscar_pedido(pedido_id, usuario)  # sua camada de dados
    if pedido is None:
        raise HTTPException(status_code=404)

    cobranca = await xflow(
        request.app.state.xflow,
        "POST",
        "/charges",
        json={
            "amountCents": pedido.total_centavos,  # inteiro: R$ 19,90 = 1990
            "customer": {"name": pedido.nome, "email": pedido.email, "document": pedido.documento},
            "description": f"Pedido #{pedido.id}",
            "external_reference": f"pedido-{pedido.id}",
        },
        idempotency_key=f"pedido-{pedido.id}",
    )
    await salvar_cobranca(pedido.id, cobranca["id"])
    return {"qr_code": cobranca["pix"]["qrCodeBase64"], "copia_e_cola": cobranca["pix"]["copyPaste"]}

usuario_logado, buscar_pedido e salvar_cobranca são da sua aplicação. Três decisões:

A rota recebe só o id do pedido. Valor, cliente e dono vêm do banco, nunca do corpo enviado pelo navegador.
Idempotency-Key por pedido. Uma nova tentativa devolve a mesma cobrança em vez de gerar outro Pix. A mesma chave com outro corpo dá 409 idempotency_conflict. Veja idempotência em API de pagamento.
A resposta é enxuta. O front-end recebe só o que mostra.

Mostrar o QR Code e o copia e cola

qr_code é um PNG em data-URL, então vai direto no src de uma imagem. copia_e_cola é o código BR Code que o pagador cola no app do banco, repassado sem alteração:

const resp = await fetch(`/pedidos/${pedidoId}/pix`, { method: "POST" });
const { qr_code, copia_e_cola } = await resp.json();
document.querySelector("#qr").src = qr_code;
document.querySelector("#copiar").onclick = () => navigator.clipboard.writeText(copia_e_cola);

Se a página sai de um Jinja2Templates, use <img src="{{ qr_code }}"> no template. Os dois campos vêm na criação e em GET /api/v1/charges/{id}, mas não na listagem.

Receber o webhook com BackgroundTasks

Cadastre a URL em Integrações > Webhooks. Ela precisa ser https:// e pública. Cada entrega é um POST JSON com os headers x-webhook-event e x-webhook-signature: sha256=..., o HMAC-SHA256 do corpo cru com o segredo do endpoint.

python

import hashlib
import hmac
import json
import os

from fastapi import BackgroundTasks, Request, Response


def assinatura_valida(corpo: bytes, recebida: str) -> bool:
    segredo = os.environ["XFLOW_WEBHOOK_SECRET"].encode()
    esperada = "sha256=" + hmac.new(segredo, corpo, hashlib.sha256).hexdigest()
    return hmac.compare_digest(esperada.encode(), recebida.encode())


async def processar_evento(evento: dict) -> None:
    data = evento["data"]
    if evento["event"] == "transaction.paid" and data["livemode"]:
        await dar_baixa(data["external_reference"], data["pix"]["endToEnd"])  # sua função


@app.post("/webhooks/xflow")
async def webhook_xflow(request: Request, tarefas: BackgroundTasks):
    corpo = await request.body()  # bytes crus, antes de qualquer parse
    if not assinatura_valida(corpo, request.headers.get("x-webhook-signature", "")):
        return Response(status_code=401)
    tarefas.add_task(processar_evento, json.loads(corpo))
    return Response(status_code=200)

O que muda no FastAPI:

Nenhum parâmetro Pydantic na rota. Um modelo no corpo faria o FastAPI converter e validar o JSON antes da assinatura. Leia await request.body() e converta depois.
hmac.compare_digest compara em tempo constante. Não use == para assinatura.
BackgroundTasks roda depois da resposta. O 200 sai na hora, e a xFlow considera o evento entregue. Se dar_baixa falhar, ninguém reenvia. Quando a baixa é crítica, grave o evento numa tabela antes de responder, ou use uma fila com retentativa.
async def ou def. Uma tarefa async def roda no event loop. Se a sua baixa usa um driver de banco síncrono, declare a tarefa com def, e o FastAPI a executa num thread pool.
Baixa idempotente. Sem 2xx, a xFlow reenvia até 3 vezes, então o mesmo evento pode chegar duas vezes. livemode: false marca evento de teste.

Os outros eventos, como estorno e saque, estão em webhook Pix.

Testar no sandbox

Com as chaves de teste nas variáveis, este script cria a cobrança e simula o pagamento:

python

# simular_pix.py
import asyncio

from app.xflow import novo_cliente, xflow


async def main():
    async with novo_cliente() as cliente:
        teste = await xflow(cliente, "POST", "/charges", json={
            "amountCents": 1990,
            "customer": {"name": "Cliente", "email": "c@exemplo.com", "document": "00000000000"},
        })
        await xflow(cliente, "POST", f"/test/charges/{teste['id']}/pay")
        paga = await xflow(cliente, "GET", f"/charges/{teste['id']}")
        print(paga["status"], paga["livemode"])  # paid False


asyncio.run(main())

A cobrança vira paid, a taxa sai das taxas reais da sua conta e o webhook chega com livemode: false. O pix.copyPaste de teste não pode ser pago, e a simulação com chave de produção responde 403 test_mode_only. Mais em sandbox da API Pix.

A rota do webhook se testa com o TestClient e um corpo assinado por você:

python

# test_webhook.py
import hashlib
import hmac
import json

from fastapi.testclient import TestClient

from app.main import app


def test_webhook_assinado(monkeypatch):
    monkeypatch.setenv("XFLOW_WEBHOOK_SECRET", "segredo-de-teste")
    corpo = json.dumps({"event": "transaction.pending", "data": {"livemode": False}}).encode()
    sig = "sha256=" + hmac.new(b"segredo-de-teste", corpo, hashlib.sha256).hexdigest()
    resp = TestClient(app).post("/webhooks/xflow", content=corpo, headers={"x-webhook-signature": sig})
    assert resp.status_code == 200

Para testar a criação da cobrança sem rede, crie o AsyncClient com transport=httpx.MockTransport(handler), que devolve respostas montadas por você.

Erros e limite de requisições

Um handler de exceção registra o request_id e devolve ao front-end uma mensagem simples:

python

import logging

from fastapi.responses import JSONResponse

from .xflow import XflowErro

log = logging.getLogger(__name__)


@app.exception_handler(XflowErro)
async def erro_xflow(request: Request, erro: XflowErro):
    log.warning("xFlow: %s", erro)  # status, código e request_id
    headers = {"Retry-After": str(erro.retry_after)} if erro.retry_after else None
    status = 503 if erro.status == 429 else 502
    return JSONResponse({"erro": "Não foi possível gerar o Pix agora."}, status_code=status, headers=headers)

O request_id localiza a chamada em Integrações > Logs da API, que guarda cada requisição autenticada por 30 dias. Os códigos mais comuns:

401 unauthorized: chave inválida ou revogada, ou IP fora da allowlist.
403 account_inactive: conta em análise ou bloqueada. Siga com a chave de teste.
429 rate_limited: o limite é por conta. Para repetir, espere erro.retry_after segundos com await asyncio.sleep, nunca time.sleep, que trava o loop. Repita com a mesma Idempotency-Key.
502 charge_failed: falha ao gerar a cobrança no provedor. Repita com backoff.

Toda resposta autenticada traz x-ratelimit-remaining. A tabela completa está em erros da API Pix.

Antes de ir para produção

Troque as chaves de teste por pk_live_/sk_live_ só nas variáveis do servidor de produção.
Cadastre a URL de produção do webhook e guarde o segredo do endpoint.
Decida onde o evento fica guardado se a tarefa de fundo falhar.
Mantenha o filtro de livemode e a baixa idempotente.
Registre o request_id dos erros e, se quiser, ative a allowlist de IP em Chaves de API.

Se a assinatura não bater, confira os valores no validador de assinatura de webhook. A versão com Django está em API Pix no Django, e arquivos prontos em Python, em exemplos de código.

Perguntas frequentes

Por que usar httpx em vez de requests no FastAPI?

Porque o requests bloqueia o event loop dentro de uma rota async def: enquanto a API responde, o worker não atende mais ninguém. O httpx.AsyncClient faz a mesma chamada com await. Crie um cliente só, no lifespan, e reutilize.

Posso declarar um modelo Pydantic no corpo da rota do webhook?

Não é o ideal. O FastAPI converte e valida o corpo antes da sua função rodar, ainda sem a assinatura conferida, e responde 422 a um corpo fora do modelo. Leia await request.body(), valide o HMAC e só então converta.

BackgroundTasks garante o processamento do webhook Pix?

Não. A tarefa roda depois que a resposta 200 sai, e a xFlow só reenvia quando não recebe 2xx. Se a tarefa falhar ou o processo reiniciar, o evento se perde. Para baixa crítica, grave o evento no banco antes de responder ou use uma fila.

Como testar a API Pix no FastAPI sem pagar?

Com chaves de teste (pk_test_/sk_test_), crie a cobrança e chame POST /api/v1/test/charges/{id}/pay. A cobrança vira paid e o webhook chega com livemode: false. Nos testes automatizados, o httpx.MockTransport substitui a API.

API Pix no FastAPI

O que você precisa

Um AsyncClient para a aplicação inteira

Criar a cobrança

Mostrar o QR Code e o copia e cola

Receber o webhook com BackgroundTasks

Testar no sandbox

Erros e limite de requisições

Antes de ir para produção

Perguntas frequentes

Continue lendo