Agentes Autônomos

Construa agentes de IA que processam pagamentos com auth, HMAC, idempotência e reconciliação

Guia para construir agentes autônomos end-to-end que interagem com a BSPAY de forma segura: auth + HMAC, idempotência, reconciliação por webhook, decisão automatizada de retry e confirmação humana em rotas críticas.

Arquitetura

Um agente robusto tem 5 camadas:

1. Auth

Cache de token Bearer com TTL 50 min. Renova proativo.

2. HMAC

Assinatura de body para cashout/transfer/conversion.

3. Idempotência

external_id UUID v4 por intenção (não por tentativa).

4. Webhooks

Validação HMAC + dedupe por transaction_id.

5. Confirmação humana

Wrapper que bloqueia rotas com $$ até confirm=True.

6. Reconciliação

Job periódico via transactions/list cobre webhooks perdidos.

Mapeamento de intenção → endpoint

IntençãoEndpointMétodoAuth
"Cobrar / receber"/v2/transactions/cashinPOSTBearer
"Endereço fixo de depósito"/v2/transactions/walletPOSTBearer
"Pagar / sacar"/v2/transactions/cashoutPOSTHMAC
"Transferir interno"/v2/internal_transfers/paymentPOSTHMAC
"Cotação spot"/v2/conversions/ratePOSTBearer
"Simular conversão"/v2/conversions/simulatePOSTBearer
"Converter moeda"/v2/conversions/newPOSTHMAC
"Saldo / perfil / limites / taxas"/v2/account/{balance,info/profile,limits,fees}GETBearer
"Extrato / consultar tx"/v2/account/transactions/listPOSTBearer
"Disputas (MED)"/v2/account/infractions, /account/infractions/{detail,reply}POST/GETBearer

Agente Python — produção-ready

Cliente completo com auth, HMAC, retry, e reconciliação.

import os, json, time, uuid, hmac, hashlib, base64, logging, threading
from typing import Optional, Any
import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("bspay-agent")

BASE = "https://api.bspay.co"


class BSPayError(Exception):
    def __init__(self, code: str, message: str, details: Optional[dict] = None, retryable: bool = False):
        self.code, self.message, self.details, self.retryable = code, message, details or {}, retryable
        super().__init__(f"{code}: {message}")


# ============================================================
# Camada 1 + 2: Auth + HMAC
# ============================================================
class BSPayClient:
    def __init__(self, client_id: str, client_secret: str, signing_key: str, callback_secret: Optional[str] = None):
        self._client_id      = client_id
        self._client_secret  = client_secret
        self._signing_key    = signing_key
        self.callback_secret = callback_secret  # validar webhooks

        self._token: Optional[str] = None
        self._token_exp: float = 0
        self._lock = threading.Lock()

    def _ensure_token(self):
        with self._lock:
            if self._token and time.time() < self._token_exp - 60:
                return
            creds = base64.b64encode(f"{self._client_id}:{self._client_secret}".encode()).decode()
            r = requests.post(
                f"{BASE}/v2/oauth/token",
                headers={"Authorization": f"Basic {creds}", "Content-Type": "application/json"},
                json={"grant_type": "client_credentials"}, timeout=10,
            )
            r.raise_for_status()
            data = r.json()
            self._token     = data["access_token"]
            self._token_exp = time.time() + data["expires_in"]
            log.info("token renovado (expira em %ds)", data["expires_in"])

    def _bearer_headers(self, extra: Optional[dict] = None) -> dict:
        self._ensure_token()
        h = {"Authorization": f"Bearer {self._token}", "Content-Type": "application/json"}
        if extra: h.update(extra)
        return h

    def _signed_headers(self, body: str) -> dict:
        ts    = str(int(time.time()))
        nonce = str(uuid.uuid4())
        sig   = hmac.new(self._signing_key.encode(), f"{ts}.{nonce}.{body}".encode(), hashlib.sha256).hexdigest()
        return self._bearer_headers({"X-Signature": sig, "X-Timestamp": ts, "X-Nonce": nonce})

    # ----------------------------------------------------------
    # Wrapper de requisição com retry exponencial + classificação de erro
    # ----------------------------------------------------------
    def _request(self, method: str, path: str, *, json_body: Optional[dict] = None,
                 raw_body: Optional[str] = None, signed: bool = False, max_retries: int = 4) -> dict:
        for attempt in range(max_retries):
            try:
                if signed:
                    body_str = raw_body or json.dumps(json_body or {}, separators=(",", ":"))
                    r = requests.request(method, f"{BASE}{path}",
                                         headers=self._signed_headers(body_str), data=body_str, timeout=30)
                else:
                    r = requests.request(method, f"{BASE}{path}",
                                         headers=self._bearer_headers(), json=json_body, timeout=30)

                if r.status_code == 429:
                    delay = int(r.headers.get("Retry-After", "5"))
                    log.warning("rate limited — esperando %ds", delay); time.sleep(delay); continue
                if r.status_code in (502, 503, 504):
                    delay = 2 ** attempt
                    log.warning("provider %d — backoff %ds", r.status_code, delay); time.sleep(delay); continue

                payload = r.json()

                if payload.get("success"):
                    return payload.get("data", payload)

                err = payload.get("error", {})
                code = err.get("code", "UNKNOWN")

                # Token expirou no meio? Renova e retenta
                if code in ("TOKEN_EXPIRED", "INVALID_TOKEN") and attempt < max_retries - 1:
                    self._token = None
                    continue

                raise BSPayError(code, err.get("message", ""), err.get("details"), err.get("retryable", False))

            except requests.exceptions.RequestException as e:
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt); continue
                raise BSPayError("NETWORK_ERROR", str(e), retryable=True)

        raise BSPayError("MAX_RETRIES_EXCEEDED", f"{method} {path}")

    # ============================================================
    # Account — read
    # ============================================================
    def balance(self) -> dict:                  return self._request("GET", "/v2/account/balance")
    def profile(self) -> dict:                  return self._request("GET", "/v2/account/info/profile")
    def limits(self) -> dict:                   return self._request("GET", "/v2/account/limits")
    def fees(self) -> dict:                     return self._request("GET", "/v2/account/fees")

    # ============================================================
    # Cashin (sem HMAC)
    # ============================================================
    def cashin(self, *, amount: float, currency: str, external_id: str,
               chain: Optional[str] = None, payer: Optional[dict] = None,
               postback_url: Optional[str] = None, split: Optional[list] = None) -> dict:
        body = {"amount": amount, "currency": currency, "external_id": external_id}
        if chain:        body["chain"] = chain
        if payer:        body["payer"] = payer
        if postback_url: body["postback_url"] = postback_url
        if split:        body["split"] = split
        return self._request("POST", "/v2/transactions/cashin", json_body=body)

    def wallet(self, *, currency: str, chain: Optional[str] = None) -> dict:
        body = {"currency": currency}
        if chain: body["chain"] = chain
        return self._request("POST", "/v2/transactions/wallet", json_body=body)

    # ============================================================
    # Cashout (HMAC)
    # ============================================================
    def cashout_pix(self, *, external_id: str, amount: float, key: str, key_type: str,
                    description: str = "", postback_url: Optional[str] = None) -> dict:
        body = {"external_id": external_id, "amount": amount, "currency": "BRL",
                "key": key, "key_type": key_type, "description": description}
        if postback_url: body["postback_url"] = postback_url
        return self._request("POST", "/v2/transactions/cashout",
                             raw_body=json.dumps(body, separators=(",", ":")), signed=True)

    def cashout_crypto(self, *, external_id: str, amount: float, currency: str,
                       key: str, network: str) -> dict:
        body = {"external_id": external_id, "amount": amount, "currency": currency,
                "key": key, "network": network}
        return self._request("POST", "/v2/transactions/cashout",
                             raw_body=json.dumps(body, separators=(",", ":")), signed=True)

    def cashout_spei(self, *, external_id: str, amount: float, key: str,
                     name: str, bank_code: Optional[str] = None) -> dict:
        body = {"external_id": external_id, "amount": amount, "currency": "MXN",
                "key": key, "name": name}
        if bank_code: body["bank_code"] = bank_code
        return self._request("POST", "/v2/transactions/cashout",
                             raw_body=json.dumps(body, separators=(",", ":")), signed=True)

    # ============================================================
    # Transfer + Conversion (HMAC)
    # ============================================================
    def transfer_internal(self, *, username: str, amount: float, currency: str,
                          description: str = "", external_id: Optional[str] = None) -> dict:
        body = {"username": username, "amount": amount, "currency": currency, "description": description,
                "external_id": external_id or str(uuid.uuid4())}
        return self._request("POST", "/v2/internal_transfers/payment",
                             raw_body=json.dumps(body, separators=(",", ":")), signed=True)

    def conversion_rate(self, *, amount: float, base_currency: str, destination_currency: str) -> dict:
        return self._request("POST", "/v2/conversions/rate",
                             json_body={"amount": amount, "base_currency": base_currency,
                                        "destination_currency": destination_currency})

    def conversion_simulate(self, *, amount: float, base_currency: str, destination_currency: str) -> dict:
        return self._request("POST", "/v2/conversions/simulate",
                             json_body={"amount": amount, "base_currency": base_currency,
                                        "destination_currency": destination_currency})

    def conversion_execute(self, *, amount: float, base_currency: str, destination_currency: str,
                           external_id: Optional[str] = None) -> dict:
        body = {"amount": amount, "base_currency": base_currency,
                "destination_currency": destination_currency,
                "external_id": external_id or str(uuid.uuid4())}
        return self._request("POST", "/v2/conversions/new",
                             raw_body=json.dumps(body, separators=(",", ":")), signed=True)

    # ============================================================
    # Listagem + reconciliação
    # ============================================================
    def list_transactions(self, **filters) -> dict:
        return self._request("POST", "/v2/account/transactions/list", json_body=filters)

    def find_by_external_id(self, external_id: str) -> Optional[dict]:
        data = self.list_transactions(external_id=external_id, page=1, page_size=1)
        items = data.get("items", [])
        return items[0] if items else None

    def validate_pix(self, pix_key: str) -> dict:

    # ============================================================
    # Disputes (MED)
    # ============================================================
    def list_infractions(self, **filters) -> dict:
        return self._request("POST", "/v2/account/infractions", json_body=filters)

    def infraction_detail(self, infraction_id: str) -> dict:
        return self._request("GET", f"/v2/account/infractions/detail?id={infraction_id}")

    # ============================================================
    # Webhook validation
    # ============================================================
    def verify_webhook(self, *, raw_body: bytes, signature: str, timestamp: int,
                       max_skew_seconds: int = 300) -> bool:
        if not self.callback_secret:
            raise RuntimeError("callback_secret não configurado")
        if abs(int(time.time()) - timestamp) > max_skew_seconds:
            return False
        expected = hmac.new(self.callback_secret.encode(), raw_body, hashlib.sha256).hexdigest()
        return hmac.compare_digest(expected, signature)


# ============================================================
# Camada 5: Confirmação humana — wrapper bloqueante
# ============================================================
DANGEROUS_OPS = {"cashout_pix", "cashout_crypto", "cashout_spei",
                 "transfer_internal", "conversion_execute"}

def with_confirmation(client: BSPayClient):
    """Decora client para exigir confirm=True em rotas perigosas."""
    class Confirmed:
        def __getattr__(self, name):
            attr = getattr(client, name)
            if name in DANGEROUS_OPS:
                def wrapper(*args, confirm: bool = False, **kw):
                    if not confirm:
                        raise PermissionError(
                            f"{name}() requer confirm=True após validação humana. "
                            f"Args: {args} {kw}"
                        )
                    return attr(*args, **kw)
                return wrapper
            return attr
    return Confirmed()


# ============================================================
# Exemplo de uso
# ============================================================
if __name__ == "__main__":
    bspay = BSPayClient(
        client_id       = os.environ["BSPAY_CLIENT_ID"],
        client_secret   = os.environ["BSPAY_CLIENT_SECRET"],
        signing_key     = os.environ["BSPAY_SIGNING_KEY"],
        callback_secret = os.environ["BSPAY_CALLBACK_SECRET"],
    )
    safe = with_confirmation(bspay)

    # 1) Receber: cashin PIX (sem HMAC)
    cobranca = safe.cashin(
        amount=100.0, currency="BRL", external_id=str(uuid.uuid4()),
        payer={"name": "João Silva", "document": "12345678901"},
        postback_url="https://meuapp.com/webhook/bspay",
    )
    log.info("QR Code: %s", cobranca["payment_info"]["qrcode"])

    # 2) Pagar: cashout PIX (HMAC) — bloqueado sem confirm=True
    try:
        safe.cashout_pix(external_id=str(uuid.uuid4()), amount=50.0,
                         key="12345678901", key_type="cpf", description="Saque")
    except PermissionError as e:
        log.warning("operação requer confirmação humana: %s", e)
        # Após validação UI/CLI/Slack:
        result = safe.cashout_pix(external_id=str(uuid.uuid4()), amount=50.0,
                                  key="12345678901", key_type="cpf",
                                  description="Saque", confirm=True)
        log.info("cashout aceito: %s", result["transaction_id"])

Decisão automatizada por código de erro

CódigoAção do agente
TOKEN_EXPIRED, INVALID_TOKENRenovar token, retentar
RATE_LIMIT_EXCEEDED (429)Esperar Retry-After segundos
INVALID_TIMESTAMP (HMAC)Sincronizar relógio, regenerar timestamp+nonce
REPLAY_DETECTED (HMAC)Gerar novo nonce e retentar
INSUFFICIENT_BALANCE, INSUFFICIENT_FUNDSAvisar usuário; não retentar
BELOW_MIN_LIMIT, EXCEEDS_MAX_LIMITMostrar details.min_amount/max_amount; sugerir valor válido
LIMIT_EXCEEDEDOperação bloqueada hoje; reagendar pra amanhã
UNSUPPORTED_CHAINMostrar details.supported; permitir trocar chain
DUPLICATE_EXTERNAL_IDBuscar tx existente via find_by_external_id() e retornar
ACCOUNT_UNDER_REVIEWPausar agente; notificar admin
PROVIDER_ERROR (502), SERVICE_UNAVAILABLE (503)Backoff exponencial até 4 tentativas
INTERNAL_CONFLICT (409)Retentar com nonce novo

Reconciliação por webhook

Webhooks podem chegar duplicados ou faltar. Implemente reconciliação periódica:

import schedule

def reconcile_pending():
    """Job a cada 1h: busca tx pending criadas há mais de 30min e atualiza status."""
    pending = bspay.list_transactions(status="pending", page_size=100)
    for tx in pending.get("items", []):
        # ... compare com sua DB; se webhook foi perdido, atualize
        pass

schedule.every().hour.do(reconcile_pending)

Boas práticas críticas

Confirmação humana é OBRIGATÓRIA

Use o with_confirmation() wrapper acima. Não confie só no system prompt — adicione validação no código que efetivamente chama a API.

external_id por intenção, não por tentativa

Gere UUID v4 quando o usuário inicia a operação. Em retries, reutilize o mesmo UUID. Isso garante idempotência mesmo se o agente re-chamar a tool.

Cache de token global (Redis)

Em deployment com várias réplicas (K8s, Lambdas), use Redis pra compartilhar o token. Senão cada réplica re-autentica e estoura o rate limit do /oauth/token (30/min/IP).

Webhook = fonte de verdade

A resposta inicial do cashout é pending. O status final só é definitivo quando chega cashout.confirmed ou cashout.failed no webhook. Não confirme nada para o usuário até receber.

Logue tudo (auditoria)

Cada chamada à BSPAY: request_id, external_id, code de erro. Isso acelera debugging com o suporte e atende a requisitos de compliance.

Nunca commite secrets

client_id, client_secret, signing_key, callback_secret: variáveis de ambiente ou secret manager (AWS Secrets Manager, Doppler, Infisical). Nunca em .env commitado.

AnteriorOutros Modelos

Esta página foi útil?