Stripe 기반 AI 에이전트에 자금 한도를 거는 백엔드 구현법
TuBrief 편집팀
2026년 7월 24일
0
AI/미래기술원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
x402 재단이 HTTP 402 규격을 가듬고 Stripe가 Agentic Commerce Suite를 내놓으면서, AI 에이전트가 알아서 결제하는 시대가 되었습니다. 엔지니어 입장에선 머리가 아픕니다. 버튼 누르는 사람이 사라졌으니까요.
사람은 결제창에서 주춤거리지만, LLM 루프가 꼬인 에이전트는 초당 수십 번씩 API를 연달아 호출합니다. 그냥 두면 순식간에 법인 카드가 한도 초과로 마비됩니다. 분산 인프라에서 에이전트 지갑을 실시간으로 제어하지 못하면 결제 사고는 예고된 수순입니다.
LLM 추론이 루프에 갇히거나 외부에 있는 402 서비스가 요청 금액을 뻥튀기할 때, 서비스 단에서 이를 막아야 합니다. 애플리케이션 메모리에 결제 금액 변수를 두고 검증하려 들면 스케일아웃 환경에서 경합 상태(Race Condition)가 터집니다.
Redis에서 Lua 스크립트를 돌려 원자적(Atomic) 연산을 처리하는 편이 안전합니다. 50ms도 안 되는 시간에 회당 한도와 일일 누적 한도를 동시 격리 상태로 검증합니다.
-- 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"}
검증을 통과한 결제건에는 일회성 난수(Nonce)를 붙여줍니다. Redis에 30초짜리 TTL로 저장했다가, 검증을 마치자마자 바로 지웁니다. 중간에서 네트워크 패킷을 훔쳐 똑같은 요청을 다시 보내는 재생 공격(Replay Attack)을 차단하기 위해서입니다. x402 프로토콜을 맞출 때는 URL과 금액, Nonce, 타임스탬프를 묶어서 HMAC-SHA256이나 EIP-712 규격으로 서명한 PAYMENT-SIGNATURE 헤더를 씁니다.
한도를 넘겼을 때 백엔드가 보내는 응답도 중요합니다. LLM이 상황을 제대로 해석하게 만들어야 합니다. 속도 제한을 넘겼다면 HTTP 429, 예산이 모자라거나 한도를 넘겼다면 HTTP 402를 돌려줍니다.
{
"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"
}
구현 절차는 명확합니다.
PAYMENT-SIGNATURE 헤더를 확인하게 합니다.LangChain이나 CrewAI 같은 프레임워크를 돌리다 타임아웃이 나면, 에이전트는 무심하게 세션을 다시 열어 툴을 호출합니다. 이때 멱등성 키(Idempotency Key)를 무작위 UUID로 매번 다르게 생성해 던지면 이중 결제가 터집니다.
에이전트가 전달하는 툴 인자(Tool Arguments) 자체를 해싱해서 결정론적(Deterministic) 멱등성 키를 뽑아내야 합니다. 그러면 Stripe API가 알아서 이전 결제 결과를 재활용하므로 억울한 중복 청구를 막아냅니다.
결제 승인은 떨어졌는데 정작 뒤따라 실행해야 할 스크래핑이나 샌드박스 연산이 터지면 어떻게 해야 할까요? 자금을 원상복구하는 보상 트랜잭션(Saga Pattern)을 태워야 합니다. 외부 시스템의 차감 기록을 직접 건드릴 수 없으니, 승인 직후 환불 API를 날려서 논리적 롤백을 수행하는 구조입니다.
네트워크 타임아웃(HTTP 500, 502, 503, 504)이 뜨면 지수 백오프(Exponential Backoff with Full Jitter)로 재시도 폭풍을 다스려야 합니다. 반면 카드 거절(card_declined)이나 만료(expired_card) 오류는 재시도 없이 바로 멈춰 세워야 합니다.
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)
}
이 코드를 백엔드에 올리는 순서는 이렇습니다.
PaymentIntent를 만들 때 이 키를 건네고, 후속 작업을 execute_agent_payment_saga() 안에서 실행합니다.except 블록에서 stripe.Refund를 때려 즉시 돈을 빼옵니다.수작업으로 이중 결제 맞추느라 매주 4시간씩 날리던 시간을 아낄 수 있습니다.
에이전트 결제 건수가 불어나면 회계팀에서 연락이 옵니다. 프롬프트 호출 내역과 영수증 데이터가 안 맞아서 세금 계산서를 못 끊겠다고요.
감사 DB 테이블에는 LLM 프롬프트 추론 ID(prompt_id), 결제 맥락, Stripe 트랜잭션 ID를 1:1로 묶어줘야 합니다. PostgreSQL의 BIGSERIAL과 JSONB 인덱스로 뼈대를 잡습니다.
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);
Stripe Tax API로 사업자 세금 식별자(Tax ID) 검증과 역거상 과세(Reverse Charge) 계산을 자동화합니다. 나중에 DB 조작 시비를 막으려면 Ed25519 서명과 SHA-256 해시 체이닝(Hash-Chaining)을 걸어두는 게 좋습니다. 각 로그가 이전 레코드의 해시값(prev_hash)을 물고 있게 만드는 겁니다. VOLT나 VeriLedger 표준 사양에 들어맞는 구조가 됩니다.
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
}
파이프라인 구축은 다음과 같이 진행합니다.
agent_audit 스키마와 인덱스를 생성합니다.분기별 B2B 정산과 세무 증빙 처리에 들던 공수를 반 이하로 덜어낼 수 있습니다.
| 영역 | 기존 방식 (일반 SDK 사용) | 안전장치 적용 후 | 실질적 이점 |
|---|---|---|---|
| 자금 제어 | 루프 오작동 시 지갑 방전 | Redis Lua 기반 실시간 원자적 차단 | 잔액 무단 유출 사고 방지 |
| 결제 오류 | 재시도 시 중복 결제 발생 | 인자 기반 멱등성 키 + Saga 자동 환불 | 결제 오차 수정 시간 주당 4시간 단축 |
| B2B 정산 | 영수증-호출 내역 미매핑 | Stripe Tax + Ed25519 해시체인 로그 | 분기별 회계 처리 시간 50% 이상 감소 |
에이전트에 지갑을 쥐여줄 때는 결제 기능보다 차단과 롤백 기능을 먼저 설계해야 합니다. Redis 제어 미들웨어와 Saga 패턴, 해시체인 로그를 얹는 순간, 프로덕션에 올려도 밤에 잠을 잘 수 있는 시스템이 만들어집니다.