Claude Agent SDK × Temporal.io で耐障害性のある長期実行AIワークフローを構築する — Durable Execution と補償トランザクションの本番設計パターン

# 推奨される構造
# ┌──────────── Workflow ─────────────┐
# │ agent_loop_workflow(input) {       │
# │   while not done:                   │
# │     response = await act.call_claude(messages)  ← Activity
# │     if response.is_tool_use:        │
# │       result = await act.run_tool(tool_name, args)  ← Activity
# │       messages.append(result)       │
# │     else:                           │
# │       done = True                   │
# │   return final_answer               │
# │ }                                   │
# └─────────────────────────────────────┘

# 依存関係のインストール
pip install temporalio anthropic python-dotenv
 
# プロジェクト構造
# workflows/
#   ├── __init__.py
#   ├── agent_workflow.py      # Workflow 定義
#   ├── activities.py          # Activity 定義（Claude API 呼び出し）
#   ├── worker.py              # Worker 起動スクリプト
#   └── starter.py             # Workflow 起動クライアント

# activities.py
import os
from anthropic import Anthropic, APIError, RateLimitError
from temporalio import activity
from dataclasses import dataclass
from typing import Any
 
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
 
@dataclass
class ClaudeCallInput:
    messages: list[dict[str, Any]]
    system: str
    tools: list[dict[str, Any]]
    max_tokens: int = 4096
    model: str = "claude-sonnet-4-5"
 
@dataclass
class ClaudeCallResult:
    stop_reason: str
    content: list[dict[str, Any]]
    usage_input_tokens: int
    usage_output_tokens: int
 
@activity.defn
async def call_claude(inp: ClaudeCallInput) -> ClaudeCallResult:
    """
    Claude API を叩く Activity。
    タイムアウト・レート制限は Temporal 側でリトライされるので、
    ここでは例外を握りつぶさずに上げるのが正解。
    """
    # Heartbeat: 長時間の呼び出し中もワーカーが生きていることを通知
    activity.heartbeat("calling claude")
 
    try:
        response = client.messages.create(
            model=inp.model,
            max_tokens=inp.max_tokens,
            system=inp.system,
            tools=inp.tools,
            messages=inp.messages,
        )
    except RateLimitError as e:
        # リトライしてほしい種類のエラーはそのまま上げる
        # Temporal の RetryPolicy が指数バックオフで再試行する
        raise
    except APIError as e:
        if e.status_code and e.status_code >= 500:
            # 5xx は一時的なので再試行対象
            raise
        # 4xx はリクエスト自体が間違っているので再試行しない
        raise activity.ApplicationError(
            f"Claude API client error: {e.status_code} {e}",
            non_retryable=True,
        )
 
    return ClaudeCallResult(
        stop_reason=response.stop_reason,
        content=[block.model_dump() for block in response.content],
        usage_input_tokens=response.usage.input_tokens,
        usage_output_tokens=response.usage.output_tokens,
    )

# activities.py (続き)
import hashlib
from typing import Any
 
@dataclass
class ToolRunInput:
    tool_name: str
    tool_input: dict[str, Any]
    idempotency_key: str  # Workflow 側で生成して渡す
 
@activity.defn
async def run_tool(inp: ToolRunInput) -> dict[str, Any]:
    """
    エージェントが呼ぶツールを実行する Activity。
    副作用を伴う操作は idempotency_key で重複実行を防ぐ。
    """
    if inp.tool_name == "send_invoice":
        return await _send_invoice_idempotent(inp.tool_input, inp.idempotency_key)
    elif inp.tool_name == "fetch_customer":
        # 冪等な読み取り操作はそのまま実行
        return await _fetch_customer(inp.tool_input)
    else:
        raise activity.ApplicationError(
            f"Unknown tool: {inp.tool_name}",
            non_retryable=True,
        )
 
async def _send_invoice_idempotent(args: dict, idem_key: str) -> dict:
    """
    請求書送付。Stripe API の Idempotency-Key を使って重複防止。
    同じキーで2回呼ばれても、Stripe 側が最初のレスポンスを返す。
    """
    # 実際は stripe.Invoice.create() に idempotency_key を渡す
    return {"invoice_id": f"inv_{idem_key[:16]}", "status": "sent"}

# agent_workflow.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy
 
# 型の循環インポートを防ぐ
with workflow.unsafe.imports_passed_through():
    from .activities import (
        call_claude, run_tool,
        ClaudeCallInput, ClaudeCallResult, ToolRunInput,
    )
 
@workflow.defn
class InvoiceProcessingWorkflow:
    def __init__(self) -> None:
        self._messages: list[dict] = []
        self._human_approval: bool | None = None
 
    @workflow.run
    async def run(self, customer_ids: list[str]) -> dict:
        """
        顧客リストに対して請求書処理を行う長期ワークフロー。
        数時間かかることを想定。人間の承認を途中で挟む。
        """
        system = "あなたは請求書処理エージェントです。指示された顧客に請求書を発行してください。"
        tools = [
            {
                "name": "fetch_customer",
                "description": "顧客情報を取得",
                "input_schema": {"type": "object", "properties": {"customer_id": {"type": "string"}}},
            },
            {
                "name": "send_invoice",
                "description": "請求書を送付",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "amount": {"type": "number"},
                    },
                },
            },
        ]
 
        self._messages = [
            {"role": "user", "content": f"顧客 {customer_ids} の請求書を発行してください。"}
        ]
 
        # リトライポリシー: 指数バックオフで最大5回まで再試行
        retry_policy = RetryPolicy(
            initial_interval=timedelta(seconds=1),
            maximum_interval=timedelta(minutes=5),
            maximum_attempts=5,
            non_retryable_error_types=["ApplicationError"],
        )
 
        results = []
        # エージェントループ
        for _ in range(50):  # 最大反復回数の安全装置
            claude_result = await workflow.execute_activity(
                call_claude,
                ClaudeCallInput(
                    messages=self._messages,
                    system=system,
                    tools=tools,
                ),
                start_to_close_timeout=timedelta(minutes=2),
                retry_policy=retry_policy,
            )
 
            self._messages.append({"role": "assistant", "content": claude_result.content})
 
            if claude_result.stop_reason == "end_turn":
                break
 
            if claude_result.stop_reason == "tool_use":
                tool_results = []
                for block in claude_result.content:
                    if block["type"] != "tool_use":
                        continue
 
                    # 冪等性キーは Workflow ID + ツール呼び出し情報から決定的に生成
                    idem_key = workflow.info().workflow_id + "_" + block["id"]
 
                    # send_invoice の前に人間の承認を待つ
                    if block["name"] == "send_invoice":
                        await workflow.wait_condition(
                            lambda: self._human_approval is not None,
                            timeout=timedelta(hours=24),
                        )
                        if not self._human_approval:
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block["id"],
                                "content": "承認されませんでした。処理をスキップします。",
                            })
                            self._human_approval = None
                            continue
                        self._human_approval = None  # 次の承認のためリセット
 
                    result = await workflow.execute_activity(
                        run_tool,
                        ToolRunInput(
                            tool_name=block["name"],
                            tool_input=block["input"],
                            idempotency_key=idem_key,
                        ),
                        start_to_close_timeout=timedelta(minutes=1),
                        retry_policy=retry_policy,
                    )
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block["id"],
                        "content": str(result),
                    })
                    results.append(result)
 
                self._messages.append({"role": "user", "content": tool_results})
 
        return {
            "customers_processed": customer_ids,
            "results": results,
            "total_messages": len(self._messages),
        }
 
    @workflow.signal
    def submit_approval(self, approved: bool) -> None:
        """人間の承認を外部から Signal で受け取る"""
        self._human_approval = approved
 
    @workflow.query
    def get_progress(self) -> dict:
        """現在の進捗を外部から Query で取得"""
        return {
            "messages_count": len(self._messages),
            "waiting_for_approval": self._human_approval is None,
        }

# agent_workflow.py に追加
from contextlib import suppress
 
@workflow.defn
class RefundableInvoiceWorkflow:
    @workflow.run
    async def run(self, customer_id: str, amount: float) -> dict:
        compensations = []  # 巻き戻し処理を積むスタック
 
        try:
            # Step 1: 顧客の信用照会
            credit_result = await workflow.execute_activity(
                run_tool,
                ToolRunInput(
                    tool_name="check_credit",
                    tool_input={"customer_id": customer_id},
                    idempotency_key=f"{workflow.info().workflow_id}_credit",
                ),
                start_to_close_timeout=timedelta(seconds=30),
            )
 
            # Step 2: 在庫予約
            reservation = await workflow.execute_activity(
                run_tool,
                ToolRunInput(
                    tool_name="reserve_inventory",
                    tool_input={"customer_id": customer_id, "amount": amount},
                    idempotency_key=f"{workflow.info().workflow_id}_reserve",
                ),
                start_to_close_timeout=timedelta(seconds=30),
            )
            compensations.append(("release_inventory", reservation))
 
            # Step 3: 請求書発行（ここで失敗したら Step 2 を巻き戻す）
            invoice = await workflow.execute_activity(
                run_tool,
                ToolRunInput(
                    tool_name="send_invoice",
                    tool_input={"customer_id": customer_id, "amount": amount},
                    idempotency_key=f"{workflow.info().workflow_id}_invoice",
                ),
                start_to_close_timeout=timedelta(minutes=1),
            )
            compensations.append(("void_invoice", invoice))
 
            return {"status": "success", "invoice": invoice}
 
        except Exception as e:
            # 逆順で補償処理を実行（Saga の原則）
            for tool_name, state in reversed(compensations):
                with suppress(Exception):
                    await workflow.execute_activity(
                        run_tool,
                        ToolRunInput(
                            tool_name=tool_name,
                            tool_input=state,
                            idempotency_key=f"{workflow.info().workflow_id}_comp_{tool_name}",
                        ),
                        start_to_close_timeout=timedelta(seconds=30),
                    )
            raise

# ❌ 間違い: 再実行時に別の値が出るため、Event History との整合性が崩れる
import datetime
timestamp = datetime.datetime.now()
 
# ✅ 正解: Temporal が提供する決定論的な API を使う
timestamp = workflow.now()
random_val = workflow.random().randint(0, 100)

# ❌ 間違い: 大きな文書をそのまま Activity から返す
@activity.defn
async def summarize_document(full_text: str) -> str:
    # full_text が 500KB の場合、次のターンで 500KB + 要約が Event History に入る
    result = await call_claude(...)
    return full_text + "\n\nSummary: " + result
 
# ✅ 正解: 大きなペイロードは外部ストレージに保存し、参照だけを渡す
@activity.defn
async def summarize_document(doc_id: str) -> str:
    full_text = await s3_get(doc_id)  # S3 から取得
    summary = await call_claude(...)
    summary_id = await s3_put(summary)  # 要約も S3 に
    return summary_id  # Event History には ID だけが残る

# 本番で使っている設定例
await workflow.execute_activity(
    call_claude,
    inp,
    start_to_close_timeout=timedelta(minutes=10),  # 極端な長文生成用に余裕
    heartbeat_timeout=timedelta(seconds=30),       # 30秒 heartbeat 来なければタイムアウト
    retry_policy=retry_policy,
)

@workflow.defn
class LongRunningAgent:
    @workflow.run
    async def run(self, state: dict) -> dict:
        iteration = state.get("iteration", 0)
 
        while iteration < 1000000:
            # ...エージェント処理...
            iteration += 1
 
            # 1000反復ごとに Continue-As-New で状態を引き継いで新しい Workflow として再スタート
            if iteration % 1000 == 0:
                workflow.continue_as_new({"iteration": iteration, "snapshot": self._state})

# ❌ 間違い: Signal が来なければ永遠に止まる
await workflow.wait_condition(lambda: self._human_approval is not None)
 
# ✅ 正解: タイムアウトを明示し、超過時のフォールバックを用意
try:
    await workflow.wait_condition(
        lambda: self._human_approval is not None,
        timeout=timedelta(hours=48),
    )
except asyncio.TimeoutError:
    # デフォルト承認、アラート、別ワークフローへのエスカレーション等
    self._human_approval = False
    await workflow.execute_activity(
        notify_timeout, inp, start_to_close_timeout=timedelta(seconds=30),
    )

@workflow.defn
class OrchestratorWorkflow:
    @workflow.run
    async def run(self, customer_batches: list[list[str]]) -> list[dict]:
        """
        顧客バッチを並列に処理。各バッチが独立した Child Workflow。
        """
        handles = []
        for batch_idx, batch in enumerate(customer_batches):
            handle = await workflow.start_child_workflow(
                InvoiceProcessingWorkflow.run,
                batch,
                id=f"{workflow.info().workflow_id}_batch_{batch_idx}",
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            handles.append(handle)
 
        # すべての Child の完了を待つ
        results = []
        for handle in handles:
            try:
                result = await handle
                results.append(result)
            except Exception as e:
                # 個別バッチの失敗は許容してログに残す
                results.append({"error": str(e)})
        return results

# activities.py にコスト集計 Activity を追加
@activity.defn
async def record_claude_cost(workflow_id: str, input_tokens: int, output_tokens: int) -> None:
    """各 Claude 呼び出し後にコストを記録"""
    # Sonnet 4.5 の料金例: $3/Mtok 入力, $15/Mtok 出力
    cost_usd = (input_tokens * 3.0 + output_tokens * 15.0) / 1_000_000
    await db_insert_cost(workflow_id, cost_usd, input_tokens, output_tokens)

# test_invoice_workflow.py
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from workflows.agent_workflow import InvoiceProcessingWorkflow
from workflows.activities import call_claude, run_tool
 
@pytest.mark.asyncio
async def test_invoice_workflow_with_mock_claude():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        # Activity をモックに差し替える
        async def mock_call_claude(inp):
            return ClaudeCallResult(
                stop_reason="end_turn",
                content=[{"type": "text", "text": "処理完了"}],
                usage_input_tokens=100,
                usage_output_tokens=50,
            )
 
        async with Worker(
            env.client,
            task_queue="test-queue",
            workflows=[InvoiceProcessingWorkflow],
            activities=[mock_call_claude, run_tool],
        ):
            result = await env.client.execute_workflow(
                InvoiceProcessingWorkflow.run,
                ["customer_1"],
                id="test-workflow",
                task_queue="test-queue",
            )
            assert result["customers_processed"] == ["customer_1"]

# worker.py
import asyncio
import os
from temporalio.client import Client
from temporalio.worker import Worker
from workflows.agent_workflow import InvoiceProcessingWorkflow, RefundableInvoiceWorkflow
from workflows.activities import call_claude, run_tool, record_claude_cost
 
async def main():
    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
    )
 
    worker = Worker(
        client,
        task_queue="claude-agent-queue",
        workflows=[InvoiceProcessingWorkflow, RefundableInvoiceWorkflow],
        activities=[call_claude, run_tool, record_claude_cost],
        max_concurrent_activities=20,   # 並列 Activity 数
        max_concurrent_workflow_tasks=50,  # 並列 Workflow タスク数
    )
    await worker.run()
 
if __name__ == "__main__":
    asyncio.run(main())