StripeベースのAIエージェントに資金制限をかけるバックエンド実装法
TuBrief 편집팀
2026년 7월 24일
0
Internet Technology원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
커뮤니티의 다른 글
댓글 (0)
Log in to leave a comment
아직 작성된 글이 없습니다
원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
x402財団がHTTP 402規格を整え、StripeがAgentic Commerce Suiteをリリースしたことで、AIエージェントが自律的に決済する時代が到来しました。エンジニアの立場からは頭の痛い話です。ボタンを押す人間がいなくなったのですから。
人間は決済画面で躊躇しますが、LLMループロジックがバグったエージェントは1秒間に数十回も連続してAPIを呼び出します。放置すれば、一瞬で法人カードが限度額オーバーで麻痺してしまいます。分散インフラ上でエージェントのウォレットをリアルタイム制御できなければ、決済事故は時間の問題です。
LLM推論がループに陥ったり、外部の402サービスが請求金額を水増ししたりする場合、サービス層でこれをブロックする必要があります。アプリケーションメモリ上に決済金額変数を置いて検証しようとすると、スケールアウト環境で競合状態（Race Condition）が発生します。
RedisでLuaスクリプトを実行し、原子（Atomic）演算として処理する方が安全です。50ms未満の時間で、1回あたりの上限と1日あたりの累積上限を同時隔離状態で検証します。
`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"}
`
検証を通過した決済リクエストには、使い捨ての乱数（Nonce）を付与します。Redisに30秒のTTLで保存し、検証が終わるとすぐに削除します。途中でネットワークパケットを盗聴され、同じリクエストを再送されるリプレイ攻撃（Replay Attack）を遮断するためです。x402プロトコルに準拠させる際は、URL、金額、Nonce、タイムスタンプをまとめ、HMAC-SHA256やEIP-712規格で署名した PAYMENT-SIGNATURE ヘッダーを使用します。
上限を超えた際にバックエンドが返すレスポンスも重要です。LLMが状況を正しく解釈できるように構築する必要があります。レート制限を超えた場合はHTTP 429、予算不足や上限オーバーの場合は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"
}
`
実装の手順は明確です。
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）エラーは、リトライせずに即座に停止させなければなりません。
`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)
}
`
このコードをバックエンドに組み込む順序は以下の通りです。
PaymentIntent を作成する際にこのキーを渡し、後続の処理を execute_agent_payment_saga() 内で実行します。except ブロックで stripe.Refund を実行して即座に資金を返金（回収）します。手作業で二重決済の照合を行い、毎週4時間も費やしていた時間を節約できます。
エージェントによる決済件数が増えてくると、経理チームから連絡が入ります。プロンプトの呼び出し履歴と領収書データが一致せず、請求書や領収書を発行できないと言うのです。
監査DBテーブルには、LLMプロンプトの推論ID（prompt_id）、決済コンテキスト、StripeトランザクションIDを1:1で紐付ける必要があります。PostgreSQLの BIGSERIAL と JSONB インデックスを使って骨組みを作ります。
`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);
`
Stripe Tax APIを利用して、事業者税金識別番号（Tax ID）の検証とリバースチャージ（Reverse Charge）計算を自動化します。後からのDB改ざん疑惑を防ぐには、Ed25519署名とSHA-256ハッシュチェーン（Hash-Chaining）を適用しておくのがベストです。各ログが前のレコードのハッシュ値（prev_hash）を保持するように設計します。これにより、VOLTや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
}
`
パイプラインの構築は以下のように進めます。
agent_audit スキーマとインデックスを作成します。四半期ごとのB2B精算や税務証明処理にかかっていた工数を半分以下に削減できます。
| 領域 | 従来方式（一般的なSDKの使用） | 安全装置の適用後 | 実質的なメリット |
|---|---|---|---|
| 資金制御 | ループ誤動作時にウォレットが枯渇 | Redis Luaベースのリアルタイム原子遮断 | 不正な残高流出事故の防止 |
| 決済エラー | リトライ時に二重決済が発生 | 引数ベースの冪等性キー + Saga自動返金 | 決済エラーの修正時間を週4時間削減 |
| B2B精算 | 領収書と呼び出し履歴が未マッピング | Stripe Tax + Ed25519ハッシュチェーンログ | 四半期ごとの会計処理時間を50%以上削減 |
エージェントにウォレットを持たせる際は、決済機能よりも遮断機能とロールバック機能を先に設計する必要があります。Redis制御ミドルウェア、Sagaパターン、そしてハッシュチェーンログを組み込んだ瞬間、本番環境にデプロイしても夜安心して眠れるシステムが完成します。