エージェントに「承認ゲート」を差し込む — Claude API で Human-in-the-loop を本番運用する設計

# agent_loop.py
# 目的: ツール呼び出しの直前で承認ゲートを挟み、承認が得られたら同じ状態から再開する
# 前提: anthropic>=0.40、Python 3.11+、永続化は Redis/Postgres どちらでも可
 
import json
import os
from dataclasses import dataclass
from typing import Any, Callable
 
from anthropic import Anthropic
 
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MODEL = "claude-sonnet-4-5"
 
# Red リスト: 必ず承認を要求するツール名
RED_TOOLS = {"send_money", "delete_user", "send_email_to_customer"}
 
@dataclass
class ApprovalRequired(Exception):
    session_id: str
    tool_use_id: str
    tool_name: str
    tool_input: dict
 
def needs_approval(tool_name: str, tool_input: dict) -> bool:
    """Red/Yellow 判定。Yellow は金額閾値で判定する例。"""
    if tool_name in RED_TOOLS:
        return True
    if tool_name == "refund_payment":
        amount = tool_input.get("amount_jpy", 0)
        return amount >= 10_000  # 1万円以上は承認
    return False
 
def run_tool(tool_name: str, tool_input: dict) -> str:
    """実ツール実行は呼び出し側で差し替え可能にする。"""
    # プロダクションでは Stripe SDK や DB クライアントを呼ぶ
    return json.dumps({"ok": True, "tool": tool_name, "input": tool_input})
 
def step_agent(
    session_id: str,
    messages: list,
    tools: list,
    store: "StateStore",
    on_approval_required: Callable[[ApprovalRequired], None],
) -> dict:
    """1ターン分だけ進める。承認が必要なら ApprovalRequired を投げる。"""
    resp = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        tools=tools,
        messages=messages,
    )
 
    # Claude の返答を履歴に追加
    messages.append({"role": "assistant", "content": resp.content})
 
    if resp.stop_reason == "end_turn":
        store.save(session_id, messages=messages, pending=None, status="completed")
        return {"status": "completed", "text": _final_text(resp)}
 
    # tool_use ブロックを処理
    tool_results = []
    for block in resp.content:
        if block.type \!= "tool_use":
            continue
 
        if needs_approval(block.name, block.input):
            # セッションを保存して承認待ちへ
            store.save(
                session_id,
                messages=messages,
                pending={
                    "tool_use_id": block.id,
                    "name": block.name,
                    "input": block.input,
                },
                status="awaiting_approval",
            )
            exc = ApprovalRequired(
                session_id=session_id,
                tool_use_id=block.id,
                tool_name=block.name,
                tool_input=block.input,
            )
            on_approval_required(exc)
            raise exc
 
        # 自動実行ツール
        result = run_tool(block.name, block.input)
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": result,
        })
 
    # tool_results をまとめて user メッセージに積む
    messages.append({"role": "user", "content": tool_results})
    store.save(session_id, messages=messages, pending=None, status="running")
    return {"status": "running"}
 
def _final_text(resp) -> str:
    return "".join(b.text for b in resp.content if b.type == "text")

# resume_agent.py
def resume_after_approval(
    session_id: str,
    decision: str,  # "approved" or "rejected"
    note: str,
    store: "StateStore",
    tools: list,
) -> dict:
    state = store.load(session_id)
    if state["status"] \!= "awaiting_approval":
        raise ValueError(f"session {session_id} is not waiting")
 
    pending = state["pending"]
    messages = state["messages"]
 
    if decision == "approved":
        result = run_tool(pending["name"], pending["input"])
        tool_result = {
            "type": "tool_result",
            "tool_use_id": pending["tool_use_id"],
            "content": result,
        }
    else:
        # 拒否理由を Claude に渡し、別の手段を考えさせる
        tool_result = {
            "type": "tool_result",
            "tool_use_id": pending["tool_use_id"],
            "content": f"Human reviewer rejected this tool call. Reason: {note}",
            "is_error": True,
        }
 
    messages.append({"role": "user", "content": [tool_result]})
    store.save(session_id, messages=messages, pending=None, status="running")
 
    return step_agent(session_id, messages, tools, store, on_approval_required=lambda e: None)

# notify_slack.py
# 目的: 承認待ちセッションを Slack に投げ、返信用の action_id にセッション ID を埋め込む
import os
import requests
 
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
 
def send_approval_request(session_id: str, tool_name: str, tool_input: dict) -> None:
    """Slack の承認依頼メッセージを投稿する。"""
    text_input = "\n".join(f"• *{k}*: `{v}`" for k, v in tool_input.items())
    payload = {
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*承認依頼* — エージェントが `{tool_name}` を実行しようとしています。",
                },
            },
            {"type": "section", "text": {"type": "mrkdwn", "text": text_input}},
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "承認"},
                        "style": "primary",
                        "action_id": f"approve::{session_id}",
                    },
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "拒否"},
                        "style": "danger",
                        "action_id": f"reject::{session_id}",
                    },
                ],
            },
        ]
    }
    r = requests.post(SLACK_WEBHOOK_URL, json=payload, timeout=5)
    r.raise_for_status()

# email_approval.py
# 目的: セッション ID と判定を署名付きトークンに詰め、メール URL を生成する
import hmac
import hashlib
import json
import base64
import os
import time
 
SIGNING_KEY = os.environ["APPROVAL_SIGNING_KEY"].encode()
 
def _sign(payload: dict) -> str:
    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    sig = hmac.new(SIGNING_KEY, raw, hashlib.sha256).digest()
    return base64.urlsafe_b64encode(raw + b"." + sig).decode()
 
def build_approval_url(session_id: str, decision: str, ttl_sec: int = 3600) -> str:
    token = _sign(
        {
            "session_id": session_id,
            "decision": decision,
            "exp": int(time.time()) + ttl_sec,
        }
    )
    return f"https://example.com/approve?token={token}"
 
def verify_token(token: str) -> dict:
    try:
        blob = base64.urlsafe_b64decode(token.encode())
        raw, sig = blob.rsplit(b".", 1)
        expected = hmac.new(SIGNING_KEY, raw, hashlib.sha256).digest()
        if not hmac.compare_digest(sig, expected):
            raise ValueError("bad signature")
        data = json.loads(raw)
        if data["exp"] < int(time.time()):
            raise ValueError("token expired")
        return data
    except Exception as e:
        raise ValueError(f"invalid token: {e}") from e

# timeout_worker.py
# 目的: 承認待ちセッションを定期的に走査し、リマインド / エスカレーション / タイムアウトを処理する
import time
 
def sweep(store, slack_sender, resumer):
    now = int(time.time())
    for session in store.list_awaiting_approval():
        age = now - session["created_at"]
        if age > 4 * 3600 and not session.get("expired"):
            # 自動拒否
            store.mark_expired(session["id"])
            resumer.resume(
                session["id"],
                decision="rejected",
                note="Approval request expired after 4 hours without response.",
            )
        elif age > 3600 and not session.get("escalated"):
            slack_sender.escalate(session)
            store.mark_escalated(session["id"])
        elif age > 900 and not session.get("reminded"):
            slack_sender.remind(session)
            store.mark_reminded(session["id"])