Claude Agent SDK のツールを冪等にする — 二重実行を止める冪等性キーと Outbox の実装メモ

# outbox_tool.py — Outbox による冪等な副作用登録
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
import asyncpg
 
 
@dataclass
class OutboxEntry:
    idempotency_key: str
    operation: str
    payload: dict[str, Any]
    status: str  # 'pending' | 'completed' | 'failed'
 
 
async def enqueue_operation(
    conn: asyncpg.Connection,
    idempotency_key: str,
    operation: str,
    payload: dict[str, Any],
) -> tuple[OutboxEntry, bool]:
    """outbox に冪等に書き込む。返り値 (エントリ, 新規作成か)。"""
    # ON CONFLICT DO NOTHING で重複登録を一発で防ぐ
    row = await conn.fetchrow(
        """
        INSERT INTO outbox (idempotency_key, operation, payload, status, created_at)
        VALUES ($1, $2, $3, 'pending', $4)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING idempotency_key, operation, payload, status
        """,
        idempotency_key, operation, payload, datetime.now(timezone.utc),
    )
    if row is not None:
        return OutboxEntry(**dict(row)), True
 
    existing = await conn.fetchrow(
        "SELECT idempotency_key, operation, payload, status FROM outbox WHERE idempotency_key=$1",
        idempotency_key,
    )
    return OutboxEntry(**dict(existing)), False
 
 
async def charge_payment_tool(conn, idempotency_key, customer_id, amount_jpy, invoice_id):
    entry, created = await enqueue_operation(
        conn, idempotency_key, "stripe_charge",
        {"customer_id": customer_id, "amount_jpy": amount_jpy, "invoice_id": invoice_id},
    )
    if not created:
        # 既存 = リトライ。状態だけ返してツールを正常終了させる
        return {"status": "already_enqueued", "state": entry.status, "key": idempotency_key}
    return {"status": "enqueued", "key": idempotency_key}
 
 
# DDL（PostgreSQL）
# CREATE TABLE outbox (
#   idempotency_key TEXT PRIMARY KEY,
#   operation TEXT NOT NULL,
#   payload JSONB NOT NULL,
#   status TEXT NOT NULL,
#   attempts INT NOT NULL DEFAULT 0,
#   last_error TEXT,
#   created_at TIMESTAMPTZ NOT NULL,
#   completed_at TIMESTAMPTZ
# );

# idempotent_wrapper.py — 既存の async 関数を冪等にする
from __future__ import annotations
import functools
import json
from typing import Awaitable, Callable
import redis.asyncio as redis
 
 
def idempotent(store: redis.Redis, *, ttl_seconds: int = 86400, prefix: str = "idem"):
    """第1引数を冪等性キーとして扱い、成功結果をキャッシュする。"""
    def decorator(func: Callable[..., Awaitable[dict]]):
        @functools.wraps(func)
        async def wrapper(idempotency_key: str, *args, **kwargs) -> dict:
            cache_key = f"{prefix}:{idempotency_key}"
            inflight = f"{cache_key}:inflight"
 
            cached = await store.get(cache_key)
            if cached is not None:
                return {"cached": True, **json.loads(cached)}
 
            # 処理中ロック（SET NX EX）。並行リクエストの二重実行を防ぐ
            locked = await store.set(inflight, "1", nx=True, ex=300)
            if not locked:
                return {"status": "in_progress", "key": idempotency_key}
 
            try:
                result = await func(idempotency_key, *args, **kwargs)
                await store.set(cache_key, json.dumps(result), ex=ttl_seconds)
                return {"cached": False, **result}
            finally:
                # 失敗は保存せず（リトライ可能に）、ロックだけ必ず外す
                await store.delete(inflight)
        return wrapper
    return decorator

# payment_agent.py
from typing import Any
from claude_agent_sdk import (
    ClaudeAgentOptions, tool, create_sdk_mcp_server, query,
)
import asyncpg
from idempotency_key import stable_idempotency_key
from outbox_tool import charge_payment_tool
 
 
def build_payment_tools(session_id: str, conn: asyncpg.Connection):
    @tool(
        "charge_invoice",
        "顧客の請求書を決済する。リトライされても二重請求は起きない。",
        {"invoice_id": str, "customer_id": str, "amount_jpy": int},
    )
    async def charge_invoice(args: dict[str, Any]) -> dict[str, Any]:
        key = stable_idempotency_key(
            session_id=session_id,
            tool_name="charge_invoice",
            logical_args={
                "invoice_id": args["invoice_id"],
                "customer_id": args["customer_id"],
                "amount_jpy": args["amount_jpy"],
            },
        )
        result = await charge_payment_tool(
            conn, key, args["customer_id"], args["amount_jpy"], args["invoice_id"],
        )
        return {"content": [{"type": "text",
                "text": f"決済登録: {result['status']} (key={key[:16]}...)"}]}
 
    return [charge_invoice]
 
 
async def run_payment_agent(session_id: str, instruction: str):
    conn = await asyncpg.connect("postgresql://localhost/myapp")
    try:
        tools = build_payment_tools(session_id, conn)
        server = create_sdk_mcp_server(name="payments", version="1.0.0", tools=tools)
        options = ClaudeAgentOptions(
            mcp_servers={"payments": server},
            allowed_tools=["mcp__payments__charge_invoice"],
            system_prompt="あなたは決済処理の担当です。請求書の指示に従って処理してください。",
        )
        async for message in query(prompt=instruction, options=options):
            print(message)
    finally:
        await conn.close()

# metrics.py（Prometheus 例）
from prometheus_client import Counter, Gauge
 
tool_calls = Counter("agent_tool_calls_total", "tool calls", ["tool"])
duplicates = Counter("agent_duplicate_detected_total", "duplicates", ["tool"])
outbox_pending = Gauge("outbox_pending_entries", "pending outbox entries")
 
tool_calls.labels(tool="charge_invoice").inc()
if not created:
    duplicates.labels(tool="charge_invoice").inc()