Como implementar limites de fundos no backend para Agentes de IA baseados na Stripe
TuBrief 편집팀
2026년 7월 24일
0
Internet Technology원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
Com a fundação x402 refinando a especificação HTTP 402 e a Stripe lançando o Agentic Commerce Suite, entramos na era em que os agentes de IA realizam pagamentos por conta própria. Do ponto de vista dos engenheiros, é uma dor de cabeça. Afinal, a pessoa que apertava o botão desapareceu.
Enquanto um humano hesitaria na tela de pagamento, um agente com um loop de LLM quebrado chama a API dezenas de vezes por segundo. Se não fizermos nada, o cartão corporativo é paralisado por excesso de limite num piscar de olhos. Se não conseguirmos controlar a carteira do agente em tempo real numa infraestrutura distribuída, um desastre de pagamentos é uma tragédia anunciada.
Quando a inferência do LLM fica presa num loop ou um serviço 402 externo inflaciona o valor da solicitação, devemos bloquear isso no nível do serviço. Se tentarmos validar mantendo a variável do valor de pagamento na memória da aplicação, ocorrerá uma condição de corrida (Race Condition) num ambiente de scale-out.
É mais seguro processar operações atômicas (Atomic) executando um script Lua no Redis. Em menos de 50ms, ele verifica simultaneamente, de forma isolada, o limite por transação e o limite diário acumulado.
`lua
-- Redis Lua Script for Dual Spending Limits (Per-Call & Daily Cumulative)
-- KEYS[1]: 일일 누적 사용량 키 (spending:agent:{agent_id}:{YYYYMMDD})
-- KEYS[2]: 속도 제한 키 (rate:agent:{agent_id})
-- ARGV[1]: 요청 결제 금액 (float)
-- ARGV[2]: 회당 최대 결제 한도 (float)
-- ARGV[3]: 일일 최대 결제 한도 (float)
-- ARGV[4]: 일일 키 만료 시간 (seconds, e.g., 86400)
local daily_key = KEYS[1]
local rate_key = KEYS[2]
local amount = tonumber(ARGV[1])
local max_per_call = tonumber(ARGV[2])
local max_daily = tonumber(ARGV[3])
local ttl_seconds = tonumber(ARGV[4])
-- 1. 회당 결제 한도 검증
if amount > max_per_call then
return {0, "EXCEEDS_PER_CALL_LIMIT"}
end
-- 2. 일일 누적 결제 한도 검증
local current_daily = tonumber(redis.call("GET", daily_key) or "0")
if (current_daily + amount) > max_daily then
return {0, "EXCEEDS_DAILY_LIMIT"}
end
-- 3. 누적 금액 업데이트 및 TTL 설정
redis.call("INCRBYFLOAT", daily_key, amount)
if redis.call("TTL", daily_key) == -1 then
redis.call("EXPIRE", daily_key, ttl_seconds)
end
return {1, "APPROVED"}
`
Aos pagamentos que passam pela validação, é anexado um número aleatório de uso único (Nonce). Nós o armazenamos no Redis com um TTL de 30 segundos e o apagamos logo após a conclusão da validação. Isso serve para bloquear ataques de repetição (Replay Attack), onde alguém intercepta os pacotes de rede para reenviar a mesma solicitação. Ao seguir o protocolo x402, agrupamos a URL, o valor, o Nonce e o timestamp, usando o cabeçalho PAYMENT-SIGNATURE assinado nos padrões HMAC-SHA256 ou EIP-712.
A resposta enviada pelo backend quando o limite é excedido também é importante. É preciso garantir que o LLM interprete a situação corretamente. Retornamos HTTP 429 se o limite de taxa for ultrapassado, e HTTP 402 se o orçamento for insuficiente ou o limite de gastos for excedido.
json { "x402Version": 2, "error": "DAILY_SPENDING_LIMIT_EXCEEDED", "message": "The requested transaction of $15.00 exceeds the remaining daily budget of $5.20.", "policy": { "max_per_call": 10.00, "daily_limit": 50.00, "current_daily_spend": 44.80, "currency": "USD" }, "action_required": "OPERATOR_APPROVAL_NEEDED" }
O procedimento de implementação é claro.
PAYMENT-SIGNATURE.Se ocorrer um timeout ao usar frameworks como LangChain ou CrewAI, o agente abrirá uma nova sessão sem hesitar e chamará a ferramenta novamente. Neste caso, se a Chave de Idempotência (Idempotency Key) for gerada como um UUID aleatório diferente a cada vez, ocorrerá um pagamento duplicado.
É necessário extrair uma chave de idempotência determinística (Deterministic) fazendo o hash dos próprios argumentos da ferramenta (Tool Arguments) passados pelo agente. Assim, a API da Stripe reutilizará automaticamente os resultados do pagamento anterior, evitando cobranças duplicadas injustas.
O que fazer se o pagamento for aprovado, mas o scraping ou a operação na sandbox que deveria ser executada logo em seguida falhar? Deve-se disparar uma transação de compensação (Saga Pattern) para restaurar os fundos. Como não podemos alterar diretamente o histórico de dedução de sistemas externos, a estrutura consiste em chamar a API de reembolso logo após a aprovação para realizar um rollback lógico.
Se houver um timeout de rede (HTTP 500, 502, 503, 504), a tempestade de tentativas deve ser controlada com recuo exponencial (Exponential Backoff with Full Jitter). Por outro lado, erros como cartão recusado (card_declined) ou expirado (expired_card) devem interromper o processo imediatamente, sem novas tentativas.
`python
import hashlib
import json
import uuid
import stripe
def generate_agent_idempotency_key(agent_id: str, tool_name: str, tool_args: dict) -> str:
canonical_args = json.dumps(tool_args, sort_keys=True)
raw_hash = hashlib.sha256(f"{agent_id}:{tool_name}:{canonical_args}".encode('utf-8')).hexdigest()
return str(uuid.UUID(raw_hash[:32]))
def execute_agent_payment_saga(
agent_id: str,
customer_id: str,
amount_cents: int,
tool_args: dict,
service_executor_func
) -> dict:
idem_key = generate_agent_idempotency_key(agent_id, "code_execution_tool", tool_args)
# 1단계: PaymentIntent 생성 및 승인 (Phase 1: Prepare & Commit)
try:
intent = stripe.PaymentIntent.create(
amount=amount_cents,
currency="usd",
customer=customer_id,
payment_method_types=["card", "link"],
confirm=True,
off_session=True,
idempotency_key=f"pi_{idem_key}"
)
except stripe.error.StripeError as e:
return {"success": False, "stage": "PAYMENT_FAILED", "error": e.user_message}
# 2단계: 실제 다운스트림 서비스 실행 (Phase 2: Execution)
try:
service_result = service_executor_func(tool_args)
return {"success": True, "data": service_result, "payment_id": intent.id}
except Exception as service_exception:
# 서비스 실행 실패 시 보상 트랜잭션 실행 (Saga Rollback: Automatic Refund)
try:
refund = stripe.Refund.create(
payment_intent=intent.id,
reason="requested_by_customer",
metadata={"rollback_reason": str(service_exception), "agent_id": agent_id},
idempotency_key=f"ref_{idem_key}"
)
return {
"success": False,
"stage": "SERVICE_FAILED_REFUNDED",
"refund_id": refund.id,
"error": str(service_exception)
}
except stripe.error.StripeError as refund_error:
return {
"success": False,
"stage": "CRITICAL_REFUND_FAILURE",
"payment_id": intent.id,
"error": str(refund_error)
}
`
A ordem para implantar este código no backend é a seguinte:
PaymentIntent na Stripe e execute a tarefa subsequente dentro de execute_agent_payment_saga().stripe.Refund no bloco except para devolver o dinheiro imediatamente.Isso poupa as 4 horas semanais perdidas tentando reconciliar pagamentos duplicados manualmente.
Quando o número de pagamentos do agente aumenta, a equipe de contabilidade entra em contato. Eles dizem que não conseguem emitir faturas porque os registros de chamadas dos prompts não batem com os dados dos recibos.
Na tabela do banco de dados de auditoria, você deve vincular o ID de inferência do prompt do LLM (prompt_id), o contexto do pagamento e o ID da transação da Stripe numa proporção de 1:1. Use BIGSERIAL e índices JSONB do PostgreSQL para montar a estrutura básica.
`sql
CREATE SCHEMA IF NOT EXISTS agent_audit;
CREATE TABLE agent_audit.transaction_logs (
id BIGSERIAL PRIMARY KEY,
prompt_id UUID NOT NULL,
agent_id VARCHAR(64) NOT NULL,
customer_id VARCHAR(255) NOT NULL,
payment_intent_id VARCHAR(255) UNIQUE,
mpp_tx_hash VARCHAR(255),
resource_url TEXT NOT NULL,
amount_subtotal NUMERIC(12, 4) NOT NULL,
amount_tax NUMERIC(12, 4) NOT NULL DEFAULT 0.0000,
amount_total NUMERIC(12, 4) NOT NULL,
currency VARCHAR(3) NOT NULL DEFAULT 'USD',
tax_code VARCHAR(32) NOT NULL DEFAULT 'txcd_10103000',
customer_tax_id VARCHAR(64),
tax_country VARCHAR(2) NOT NULL,
taxability_override VARCHAR(32),
payload_digest VARCHAR(64) NOT NULL,
signature TEXT NOT NULL,
prev_hash VARCHAR(64) NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_agent_audit_prompt ON agent_audit.transaction_logs(agent_id, prompt_id);
CREATE INDEX idx_agent_audit_created ON agent_audit.transaction_logs(created_at);
CREATE INDEX idx_agent_audit_gin_meta ON agent_audit.transaction_logs USING gin (metadata);
`
Com a Stripe Tax API, automatizamos a validação do identificador fiscal corporativo (Tax ID) e o cálculo da cobrança reversa (Reverse Charge). Para evitar contestações futuras de manipulação no banco de dados, é recomendável aplicar assinaturas Ed25519 e Hash-Chaining SHA-256. Isso faz com que cada log faça referência ao valor do hash do registro anterior (prev_hash). Assim, obtemos uma estrutura compatível com as especificações padrão do VOLT ou VeriLedger.
`python
from cryptography.hazmat.primitives.asymmetric import ed25519
import base64
import hashlib
private_key = ed25519.Ed25519PrivateKey.generate()
def create_tax_compliant_agent_invoice(
customer_id: str,
amount_cents: int,
currency: str,
tax_id_number: str = None,
country_code: str = "DE"
) -> dict:
if tax_id_number:
stripe.Customer.create_tax_id(customer_id, type="eu_vat", value=tax_id_number)
calculation = stripe.tax.Calculation.create(
currency=currency,
customer_details={
"address": {"country": country_code},
"address_source": "billing",
"tax_ids": [{"type": "eu_vat", "value": tax_id_number}] if tax_id_number else []
},
line_items=[{
"amount": amount_cents,
"reference": "AGENT_API_USAGE",
"tax_behavior": "exclusive",
"tax_code": "txcd_10103000"
}]
)
return {
"calculation_id": calculation.id,
"subtotal": amount_cents / 100.0,
"tax_amount": calculation.tax_amount_exclusive / 100.0,
"total": (amount_cents + calculation.tax_amount_exclusive) / 100.0,
"is_reverse_charge": calculation.customer_details.taxability_override == "reverse_charge"
}
def generate_non_repudiable_audit_proof(
prompt_id: str,
payment_intent_id: str,
amount: float,
prev_hash: str
) -> dict:
payload_raw = f"{prompt_id}:{payment_intent_id}:{amount:.4f}:{prev_hash}"
payload_hash = hashlib.sha256(payload_raw.encode('utf-8')).hexdigest()
signature_bytes = private_key.sign(payload_hash.encode('utf-8'))
signature_b64 = base64.b64encode(signature_bytes).decode('utf-8')
current_block_hash = hashlib.sha256(f"{payload_hash}:{signature_b64}".encode('utf-8')).hexdigest()
return {
"payload_hash": payload_hash,
"signature": signature_b64,
"block_hash": current_block_hash
}
`
A construção do pipeline prossegue da seguinte forma:
agent_audit e os índices no PostgreSQL de acordo com a DDL fornecida.Isso pode reduzir pela metade ou mais o esforço gasto no processamento das liquidações B2B trimestrais e na emissão de comprovantes fiscais.
| Área | Método Anterior (Usando SDK Comum) | Pós-aplicação das Travas de Segurança | Benefício Real |
|---|---|---|---|
| Controle de Fundos | Esgotamento da carteira em caso de erro de loop | Bloqueio atômico em tempo real baseado em Redis Lua | Prevenção de vazamento não autorizado do saldo |
| Erros de Pagamento | Pagamento duplicado em tentativas repetidas | Chave de idempotência baseada em argumentos + Reembolso Automático Saga | Redução de 4 horas/semana na correção de erros de pagamento |
| Liquidação B2B | Falta de mapeamento entre recibos e histórico de chamadas | Log com Stripe Tax + Hash-Chaining Ed25519 | Redução de mais de 50% no tempo de processamento contábil trimestral |
Ao entregar uma carteira a um agente, as funções de bloqueio e rollback devem ser projetadas antes da funcionalidade de pagamento. Ao adicionar o middleware de controle no Redis, o Padrão Saga e logs de hash-chaining, você cria um sistema que, mesmo em produção, permite que você durma tranquilo à noite.