Cara Implementasi Backend untuk Membatasi Dana pada Agen AI berbasis Stripe
TuBrief 편집팀
2026년 7월 24일
0
Internet Technology원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
Dengan Yayasan x402 yang menyempurnakan spesifikasi HTTP 402 dan Stripe yang meluncurkan Agentic Commerce Suite, kita telah memasuki era di mana agen AI dapat melakukan pembayaran secara mandiri. Dari sudut pandang insinyur, ini sangat membingungkan. Karena orang yang menekan tombol pembayaran telah menghilang.
Manusia mungkin ragu-ragu di halaman pembayaran, tetapi agen AI yang terjebak dalam loop LLM akan memanggil API puluhan kali per detik secara beruntun. Jika dibiarkan, kartu kredit perusahaan akan lumpuh karena melebihi batas (limit) dalam sekejap. Tanpa kemampuan mengendalikan dompet agen secara real-time di infrastruktur terdistribusi, insiden pembayaran hanyalah masalah waktu.
Ketika inferensi LLM terjebak dalam loop atau layanan 402 eksternal menggelembungkan jumlah permintaan, kita harus menghentikannya di tingkat layanan. Jika Anda mencoba memvalidasi dengan menyimpan variabel jumlah pembayaran di memori aplikasi, race condition akan terjadi di lingkungan scale-out.
Lebih aman menggunakan skrip Lua di Redis untuk menangani operasi atomik (atomic operation). Dalam waktu kurang dari 50ms, skrip ini memvalidasi batas per transaksi dan batas kumulatif harian secara bersamaan dalam kondisi terisolasi.
`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"}
`
Pembayaran yang lolos validasi akan diberi angka acak sekali pakai (nonce). Simpan di Redis dengan TTL 30 detik, lalu segera hapus begitu validasi selesai. Ini bertujuan untuk mencegah serangan replay (replay attack), di mana paket jaringan dicuri di tengah jalan dan permintaan yang sama dikirim ulang. Saat menyesuaikan dengan protokol x402, kita menggunakan header PAYMENT-SIGNATURE yang ditandatangani dengan standar HMAC-SHA256 atau EIP-712 yang menggabungkan URL, jumlah, nonce, dan timestamp.
Respons yang dikirim backend saat batas terlampaui juga sangat penting. Anda harus membuat LLM menginterpretasikan situasi dengan benar. Jika melebihi batas kecepatan (rate limit), kembalikan HTTP 429; jika anggaran tidak mencukupi atau melebihi batas, kembalikan HTTP 402.
`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"
}
`
Langkah implementasinya sangat jelas.
PAYMENT-SIGNATURE.Ketika terjadi timeout saat menjalankan kerangka kerja seperti LangChain atau CrewAI, agen akan secara acak membuka sesi baru dan memanggil tool kembali. Pada saat ini, jika Anda membuat dan mengirim kunci idempotesitas (idempotency key) secara acak dengan UUID baru setiap kali, pembayaran ganda akan terjadi.
Anda harus mengekstrak kunci idempotensitas deterministik (deterministic idempotency key) dengan meng-hash argumen tool (tool arguments) yang disampaikan oleh agen itu sendiri. Dengan begitu, Stripe API akan secara otomatis menggunakan kembali hasil pembayaran sebelumnya sehingga mencegah penagihan ganda yang merugikan.
Bagaimana jika persetujuan pembayaran berhasil, tetapi operasi scraping atau sandbox yang harus dilakukan setelahnya gagal? Anda harus menjalankan transaksi kompensasi (Pola Saga) untuk mengembalikan dana. Karena kita tidak dapat mengubah catatan pengurangan di sistem eksternal secara langsung, strukturnya adalah melakukan rollback logis dengan mengirimkan API pengembalian dana (refund) segera setelah persetujuan.
Jika terjadi timeout jaringan (HTTP 500, 502, 503, 504), Anda harus mengendalikan lonjakan percobaan ulang (retry storm) dengan exponential backoff (exponential backoff with full jitter). Sebaliknya, kesalahan penolakan kartu (card_declined) atau kartu kadaluarsa (expired_card) harus segera dihentikan tanpa percobaan ulang.
`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)
}
`
Urutan untuk menerapkan kode ini pada backend adalah sebagai berikut.
PaymentIntent, dan jalankan tugas susulan di dalam execute_agent_payment_saga().stripe.Refund di dalam blok except untuk segera mengembalikan uang.Anda dapat menghemat waktu 4 jam setiap minggu yang sebelumnya terbuang untuk menyesuaikan pembayaran ganda secara manual.
Saat jumlah transaksi agen meningkat, tim akuntansi akan menghubungi Anda. Mereka tidak dapat menerbitkan faktur pajak karena riwayat pemanggilan prompt dan data tanda terima tidak cocok.
Di tabel DB audit, kita harus mengikat ID inferensi prompt LLM (prompt_id), konteks pembayaran, dan ID transaksi Stripe secara 1:1. Kita membuat kerangkanya menggunakan BIGSERIAL dan indeks JSONB pada PostgreSQL.
`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);
`
Otomatiskan verifikasi identifikasi pajak bisnis (Tax ID) dan kalkulasi pajak balik (reverse charge) menggunakan Stripe Tax API. Untuk mencegah perselisihan manipulasi DB di kemudian hari, ada baiknya Anda menerapkan tanda tangan Ed25519 dan hash-chaining SHA-256. Buat setiap log memegang nilai hash (prev_hash) dari rekam data sebelumnya. Ini menjadi struktur yang sesuai dengan spesifikasi standar VOLT atau 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
}
`
Pembangunan pipeline dilakukan sebagai berikut.
agent_audit dan indeks pada PostgreSQL sesuai DDL yang ditulis.Anda dapat memangkas tenaga yang terbuang untuk rekonsiliasi B2B triwulanan dan pemrosesan bukti pajak hingga lebih dari separuhnya.
| Area | Metode Lama (Menggunakan SDK Biasa) | Setelah Penerapan Pengaman | Manfaat Nyata |
|---|---|---|---|
| Kontrol Dana | Dompet terkuras saat loop malafungsi | Pemblokiran atomik real-time berbasis Redis Lua | Mencegah insiden kebocoran saldo tanpa izin |
| Kesalahan Pembayaran | Pembayaran ganda saat mencoba ulang | Kunci idempotensitas berbasis argumen + Pengembalian dana otomatis Saga | Pengurangan waktu koreksi kesalahan pembayaran sebesar 4 jam per minggu |
| Rekonsiliasi B2B | Tanda terima dan riwayat pemanggilan tidak terpetakan | Stripe Tax + Log hash chain Ed25519 | Waktu pemrosesan akuntansi triwulanan berkurang lebih dari 50% |
Saat memberikan dompet kepada agen, Anda harus merancang fitur pemblokiran dan rollback terlebih dahulu daripada fitur pembayaran. Saat Anda menambahkan middleware kontrol Redis, pola Saga, dan log hash chain, sistem yang aman akan tercipta sehingga Anda dapat tidur nyenyak di malam hari meskipun sistem telah berjalan di lingkungan produksi.