API Pix em Python

Em Python, a API Pix se integra com a biblioteca requests e poucas linhas: um POST cria a cobrança, a resposta já traz o QR Code e o copia e cola, e uma rota do seu servidor recebe o webhook de pagamento. O código abaixo segue a documentação da API, roda em Python 3.9 ou superior e usa a base https://app.xflowpayments.com/api/v1. Para entender a integração completa antes do código, leia o guia da API Pix.

Antes de começar

• Gere uma chave em Desenvolvedores > Chaves de API, no painel. A chave pública (pk_live_...) fica sempre visível. O segredo (sk_live_...) aparece uma única vez.
• Para testar, crie uma chave do ambiente Teste (pk_test_... e sk_test_...). Ela pode ser criada antes da aprovação do cadastro.
• Instale a dependência: pip install requests.
• Defina XFLOW_PUBLIC_KEY, XFLOW_SECRET_KEY e XFLOW_WEBHOOK_SECRET como variáveis de ambiente.

O segredo fica no servidor. Não o coloque em código versionado nem em nada que rode no navegador.

Um cliente para a API

A autenticação é HTTP Basic: usuário é a chave pública, senha é o segredo. Com requests, basta auth=(publica, segredo). Valores monetários são inteiros em centavos.

# xflow.py
import os
import requests

BASE = "https://app.xflowpayments.com/api/v1"
AUTH = (os.environ["XFLOW_PUBLIC_KEY"], os.environ["XFLOW_SECRET_KEY"])


class XflowErro(Exception):
    def __init__(self, status, corpo, retry_after=None):
        self.status = status
        self.code = corpo.get("error")
        self.request_id = corpo.get("request_id")
        self.retry_after = retry_after
        super().__init__(f"{status} {self.code}: {corpo.get('message')} ({self.request_id})")


def xflow(method, path, json=None, idempotency_key=None):
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    resp = requests.request(method, BASE + path, auth=AUTH, json=json, headers=headers, timeout=30)
    corpo = resp.json() if resp.content else {}
    if resp.status_code >= 400:
        raise XflowErro(resp.status_code, corpo, resp.headers.get("retry-after"))
    return corpo

Sem dependências, com urllib

Se não puder instalar pacotes, a biblioteca padrão resolve. A diferença é montar o header Authorization à mão:

import base64
import json
import os
import urllib.request

cred = f"{os.environ['XFLOW_PUBLIC_KEY']}:{os.environ['XFLOW_SECRET_KEY']}".encode()
req = urllib.request.Request(
    "https://app.xflowpayments.com/api/v1/balance",
    headers={"Authorization": "Basic " + base64.b64encode(cred).decode()},
)
with urllib.request.urlopen(req, timeout=30) as resp:
    print(json.load(resp))  # {"available": ..., "reserved": ..., "currency": "BRL"}

Com urllib, uma resposta de erro vira a exceção urllib.error.HTTPError. O corpo dela, lido com .read(), traz o mesmo JSON de erro da API.

Toda resposta traz um request_id (req_...), no corpo e no header x-request-id. Registre esse valor quando der erro: ele localiza a chamada em Integrações > Logs da API, onde cada requisição autenticada fica 30 dias com status, duração e código de erro. A tabela de códigos está em erros da API Pix.

Criar a cobrança Pix

from xflow import xflow

pedido_id = "1234"

cobranca = xflow(
    "POST",
    "/charges",
    json={
        "amountCents": 1990,  # R$ 19,90
        "customer": {
            "name": "Cliente Exemplo",
            "email": "cliente@exemplo.com",
            "document": "00000000000",  # CPF ou CNPJ, só dígitos
        },
        "description": f"Pedido #{pedido_id}",
        "external_reference": f"pedido-{pedido_id}",
    },
    idempotency_key=f"pedido-{pedido_id}",
)

print(cobranca["id"], cobranca["status"])  # clx... pending

A Idempotency-Key deve ser única por pedido. Se a conexão cair e você repetir a chamada com a mesma chave, a API devolve a mesma cobrança em vez de gerar outro Pix. Com a mesma chave e outro corpo, a resposta é 409 idempotency_conflict. Veja o porquê em idempotência em API de pagamento.

Salve o id devolvido junto do pedido. Ele serve para consultar o status em GET /api/v1/charges/{id} e, se o cliente desistir, para pedir a devolução em POST /api/v1/charges/{id}/refund, detalhada em estorno Pix por API.

O external_reference (até 120 caracteres) liga a cobrança ao seu pedido. Ele volta nas respostas e webhooks e filtra a listagem: GET /api/v1/charges?external_reference=pedido-1234. Valor fora do mínimo ou do máximo da conta retorna 400 validation.

Mostrar o QR Code e o copia e cola

Em um template Jinja, por exemplo:

<img src="{{ cobranca.pix.qrCodeBase64 }}" alt="QR Code Pix" width="240" height="240">
<textarea readonly>{{ cobranca.pix.copyPaste }}</textarea>

pix.qrCodeBase64 é um PNG em data-URL, pronto para o src. pix.copyPaste é o payload BR Code que o pagador cola no app do banco. Repasse sem alterar. Os dois campos vêm na criação e em GET /api/v1/charges/{id}, mas não na listagem. Para exibição em outros canais, veja QR Code Pix dinâmico por API.

Receber o webhook com Flask

Cadastre a URL em Integrações > Webhooks. Ela tem que 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.

# webhook.py
import hashlib
import hmac
import json
import os

from flask import Flask, abort, request

app = Flask(__name__)
SEGREDO = os.environ["XFLOW_WEBHOOK_SECRET"].encode()


def assinatura_valida(raw, header):
    esperado = "sha256=" + hmac.new(SEGREDO, raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(esperado.encode(), (header or "").encode())


@app.post("/webhooks/xflow")
def webhook():
    raw = request.get_data()  # corpo cru, antes de qualquer parse
    if not assinatura_valida(raw, request.headers.get("x-webhook-signature")):
        abort(401)

    evento = json.loads(raw)
    data = evento["data"]
    try:
        if evento["event"] == "transaction.paid" and data["livemode"]:
            dar_baixa_no_pedido(data["external_reference"], data["pix"]["endToEnd"])  # sua função
    except Exception:
        return "", 500  # sem 2xx, a entrega é reenviada
    return "", 200

No FastAPI, o corpo cru vem de await request.body(). O resto é igual. Os cuidados:

• Valide antes do json.loads. Usar request.get_json() e serializar de novo muda os bytes e quebra o HMAC.
• Compare bytes com hmac.compare_digest, que não vaza tempo de comparação.
• Responda 2xx. Sem isso, a xFlow reenvia até 3 vezes. A sua baixa precisa tolerar o mesmo evento duas vezes.
• Descarte livemode: false em produção: são eventos gerados com chave de teste.

No webhook, o valor vem em amount, e transaction.paid traz também net e fee. payer e pix.endToEnd vêm do PSP na confirmação, para conciliar com o extrato. Todos os eventos estão em webhook Pix.

Testar no sandbox

Com as chaves de teste nas variáveis, o mesmo código roda sem mover dinheiro. Toda resposta traz livemode: false, nenhum provedor Pix é chamado e o pix.copyPaste é um marcador que não pode ser pago.

from xflow import xflow

teste = xflow("POST", "/charges", json={
    "amountCents": 1990,
    "customer": {"name": "Cliente", "email": "c@exemplo.com", "document": "00000000000"},
})

xflow("POST", f"/test/charges/{teste['id']}/pay")

paga = xflow("GET", f"/charges/{teste['id']}")
print(paga["status"], paga["livemode"])  # paid False

A cobrança vira paid, a taxa é calculada com as taxas reais da sua conta e o webhook transaction.paid é disparado com livemode: false. A simulação só aceita chave de teste. Com chave de produção, volta 403 test_mode_only. Saques não rodam em teste (403 test_mode_unsupported). O ambiente está detalhado em sandbox da API Pix.

Conciliar as cobranças pagas

Para fechar o dia, liste as cobranças pagas de um período. A paginação é por cursor: limit vai de 1 a 100 e, enquanto has_more for verdadeiro, você repete a chamada com starting_after igual ao id do último item.

from urllib.parse import urlencode
from xflow import xflow


def cobrancas_pagas(de, ate):
    cursor = None
    while True:
        filtros = {"status": "paid", "from": de, "to": ate, "limit": 100}
        if cursor:
            filtros["starting_after"] = cursor
        pagina = xflow("GET", "/charges?" + urlencode(filtros))
        yield from pagina["data"]
        if not pagina["has_more"] or not pagina["data"]:
            break
        cursor = pagina["data"][-1]["id"]


for c in cobrancas_pagas("2026-10-01", "2026-10-31"):
    print(c["external_reference"], c["amountCents"])

Cada item traz payer e pix.endToEnd, que a documentação indica para conciliar o recebimento com o extrato bancário.

Antes de ir para produção

1. Troque as chaves de teste pelas de produção (pk_live_/sk_live_) só no ambiente de produção.
2. Cadastre a URL de produção do webhook e guarde o segredo do endpoint.
3. Mantenha o descarte de eventos com livemode: false.
4. Registre o request_id de toda resposta com erro.
5. Se quiser restringir a origem das chamadas, ative a allowlist de IP em Chaves de API. Fora da lista, a resposta é 401.

Limite de requisições e retentativa

Um 429 rate_limited traz o header retry-after, em segundos. Espere esse tempo e tente de novo com backoff:

import time
from xflow import XflowErro, xflow


def com_retentativa(method, path, tentativas=3, **kwargs):
    for tentativa in range(tentativas):
        try:
            return xflow(method, path, **kwargs)
        except XflowErro as erro:
            if erro.status != 429 or tentativa == tentativas - 1:
                raise
            time.sleep(int(erro.retry_after or 2 ** tentativa))

O limite é por conta. Toda resposta autenticada traz x-ratelimit-limit, x-ratelimit-remaining e x-ratelimit-reset, então dá para desacelerar antes de tomar o 429. Prefira o webhook ao polling: consultar GET /api/v1/charges/{id} em laço gasta o mesmo limite. O fluxo em outras linguagens está em API Pix em Node.js e API Pix em PHP.