Claude API × Inngest で耐障害性のある AI ワークフローを本番運用する — リトライ・冪等性・人間承認を組み込んだ実装ガイド

# 依存関係（Node 20+ 推奨）
npm install inngest @anthropic-ai/sdk
npm install -D @types/node
 
# 環境変数（.env.local）
ANTHROPIC_API_KEY=YOUR_ANTHROPIC_API_KEY
INNGEST_EVENT_KEY=YOUR_INNGEST_EVENT_KEY
INNGEST_SIGNING_KEY=YOUR_INNGEST_SIGNING_KEY

// src/inngest/client.ts
import { Inngest } from "inngest";
 
export const inngest = new Inngest({
  id: "claude-lab-agent",
  // 本番では eventKey を環境変数から自動読込
});

// src/app/api/inngest/route.ts
import { serve } from "inngest/next";
import { inngest } from "@/inngest/client";
import { runResearchAgent } from "@/inngest/functions/research-agent";
 
export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [runResearchAgent],
});

npx inngest-cli dev
# ブラウザで http://localhost:8288 を開く

// src/inngest/functions/research-agent.ts
import Anthropic from "@anthropic-ai/sdk";
import { inngest } from "../client";
 
const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
 
const tools: Anthropic.Tool[] = [
  {
    name: "web_search",
    description: "公開情報を検索して上位3件のスニペットを返します。",
    input_schema: {
      type: "object",
      properties: { query: { type: "string" } },
      required: ["query"],
    },
  },
  {
    name: "send_summary_email",
    description: "完成した要約を指定アドレスに送信します。",
    input_schema: {
      type: "object",
      properties: {
        to: { type: "string" },
        subject: { type: "string" },
        body: { type: "string" },
      },
      required: ["to", "subject", "body"],
    },
  },
];
 
export const runResearchAgent = inngest.createFunction(
  {
    id: "run-research-agent",
    // ループが暴走しないよう同時実行数を絞る
    concurrency: { limit: 5 },
    // Anthropic 側の一時障害に備えた既定リトライ
    retries: 4,
  },
  { event: "research/agent.requested" },
  async ({ event, step, logger }) => {
    const { topic, requestedBy, jobId } = event.data as {
      topic: string;
      requestedBy: string;
      jobId: string;
    };
 
    let messages: Anthropic.MessageParam[] = [
      { role: "user", content: `次のトピックを調査して要約してください: ${topic}` },
    ];
 
    // 最大 6 ターンで打ち切る（無限ループ防止）
    for (let turn = 0; turn < 6; turn++) {
      // ✅ Claude 呼び出しは step.run で囲む = 結果が永続化される
      const response = await step.run(`claude-turn-${turn}`, async () => {
        const res = await anthropic.messages.create({
          model: "claude-sonnet-4-6",
          max_tokens: 1024,
          tools,
          messages,
        });
        return res;
      });
 
      messages.push({ role: "assistant", content: response.content });
 
      if (response.stop_reason === "end_turn") {
        logger.info("agent finished", { turn, jobId });
        return { ok: true, finalMessages: messages };
      }
 
      if (response.stop_reason === "tool_use") {
        const toolUses = response.content.filter(
          (b): b is Anthropic.ToolUseBlock => b.type === "tool_use",
        );
        const toolResults: Anthropic.ToolResultBlockParam[] = [];
 
        for (const tu of toolUses) {
          // ツール実行も step に分けると、片方だけ失敗しても再開できる
          const result = await step.run(`tool-${turn}-${tu.id}`, async () => {
            return runTool(tu.name, tu.input, { jobId, requestedBy });
          });
          toolResults.push({
            type: "tool_result",
            tool_use_id: tu.id,
            content: JSON.stringify(result),
          });
        }
        messages.push({ role: "user", content: toolResults });
      }
    }
 
    throw new Error(`Agent did not converge in 6 turns (jobId=${jobId})`);
  },
);

// src/inngest/functions/tools.ts
import { Resend } from "resend";
import { db } from "@/db";
 
const resend = new Resend(process.env.RESEND_API_KEY!);
 
export async function runTool(
  name: string,
  input: unknown,
  ctx: { jobId: string; requestedBy: string },
): Promise<unknown> {
  if (name === "web_search") {
    return webSearch((input as { query: string }).query);
  }
 
  if (name === "send_summary_email") {
    const args = input as { to: string; subject: string; body: string };
    // ✅ 冪等性キー: jobId + tool_use_id 相当
    const idempotencyKey = `${ctx.jobId}:email:${hash(args)}`;
 
    // DB に「この key で既に送ったか」を先に記録（UPSERT）
    const inserted = await db.idempotency.upsert({
      where: { key: idempotencyKey },
      create: { key: idempotencyKey, status: "pending" },
      update: {},
    });
 
    if (inserted.status === "completed") {
      return { skipped: true, reason: "already sent" };
    }
 
    const result = await resend.emails.send({
      from: "agent@example.com",
      to: args.to,
      subject: args.subject,
      text: args.body,
    });
 
    await db.idempotency.update({
      where: { key: idempotencyKey },
      data: { status: "completed", externalId: result.id },
    });
    return { sent: true, id: result.id };
  }
 
  throw new Error(`unknown tool: ${name}`);
}
 
async function webSearch(query: string) {
  // 検索 API 呼び出し（省略）
  return [{ title: "...", snippet: "...", url: "..." }];
}
 
function hash(obj: unknown) {
  // 簡易: JSON 文字列のハッシュ。実運用は crypto.subtle で SHA-256
  return Buffer.from(JSON.stringify(obj)).toString("base64").slice(0, 24);
}

import { NonRetriableError } from "inngest";
 
const response = await step.run(`claude-turn-${turn}`, async () => {
  try {
    return await anthropic.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 1024,
      tools,
      messages,
    });
  } catch (e) {
    if (e instanceof Anthropic.APIError) {
      // 4xx 系（429 を除く）は即座に止める
      if (e.status && e.status >= 400 && e.status < 500 && e.status !== 429) {
        throw new NonRetriableError(
          `claude rejected request: ${e.status} ${e.message}`,
          { cause: e },
        );
      }
    }
    throw e; // それ以外は Inngest にリトライさせる
  }
});

// 高額決済前に人間承認を待つ例
if (toolName === "charge_customer" && (input as { amount: number }).amount > 50000) {
  // 承認リクエストを Slack や DB に書き出す処理
  await step.run("request-approval", async () => {
    return notifyApprovers({
      jobId,
      amount: (input as { amount: number }).amount,
      requestedBy,
    });
  });
 
  // 最大 24 時間、承認イベントを待つ
  const approval = await step.waitForEvent("wait-for-approval", {
    event: "agent/approval.received",
    timeout: "24h",
    if: `event.data.jobId == "${jobId}"`, // 同じ jobId のイベントだけ拾う
  });
 
  if (!approval) {
    throw new NonRetriableError("Approval timed out after 24h");
  }
  if (approval.data.decision !== "approved") {
    return { skipped: true, reason: "rejected by human" };
  }
  // 以降、本当の課金処理を実行
}

export const summarizeBatch = inngest.createFunction(
  { id: "summarize-batch", concurrency: { limit: 20 } },
  { event: "batch/summary.requested" },
  async ({ event, step }) => {
    const { docIds } = event.data as { docIds: string[] };
 
    // ✅ ファンアウト: 各ドキュメントごとに子関数を起動
    const summaries = await Promise.all(
      docIds.map((id) =>
        step.invoke(`summarize-${id}`, {
          function: summarizeOne,
          data: { docId: id },
        }),
      ),
    );
 
    // ✅ ファンイン: すべて揃ったら統合レポートを作る
    const report = await step.run("compose-report", async () => {
      const res = await anthropic.messages.create({
        model: "claude-sonnet-4-6",
        max_tokens: 4096,
        messages: [
          {
            role: "user",
            content: `以下の要約から統合レポートを作成してください:\n\n${summaries
              .map((s) => `- ${s.title}: ${s.summary}`)
              .join("\n")}`,
          },
        ],
      });
      return res.content[0].type === "text" ? res.content[0].text : "";
    });
 
    return { docCount: docIds.length, report };
  },
);
 
const summarizeOne = inngest.createFunction(
  { id: "summarize-one", concurrency: { limit: 10 } },
  { event: "internal/summarize.one" },
  async ({ event, step }) => {
    const { docId } = event.data as { docId: string };
    const doc = await step.run("fetch-doc", () => db.doc.findUniqueOrThrow({ where: { id: docId } }));
    const summary = await step.run("call-claude", async () => {
      const res = await anthropic.messages.create({
        model: "claude-haiku-4-5",
        max_tokens: 512,
        messages: [{ role: "user", content: `要約してください:\n\n${doc.body}` }],
      });
      return res.content[0].type === "text" ? res.content[0].text : "";
    });
    return { id: docId, title: doc.title, summary };
  },
);

const response = await step.run(`claude-turn-${turn}`, async ({ attempt }) => {
  const res = await anthropic.messages.create({ /* ... */ });
  logger.info("claude.usage", {
    jobId,
    turn,
    attempt,
    inputTokens: res.usage.input_tokens,
    outputTokens: res.usage.output_tokens,
    cacheCreation: res.usage.cache_creation_input_tokens,
    cacheRead: res.usage.cache_read_input_tokens,
  });
  return res;
});

// ❌ 危険: 再実行時に異なる値になる
const requestId = await step.run("save", async () => {
  const id = crypto.randomUUID(); // ← 再実行ごとに変わる
  await db.requests.create({ data: { id, ... } });
  return id;
});

// ✅ 正解
const requestId = crypto.randomUUID(); // step の外で確定
await step.run("save", async () => {
  await db.requests.create({ data: { id: requestId, ... } });
});

const compressed = await step.run(`compress-${tu.id}`, async () => {
  if (JSON.stringify(rawResult).length < 4000) return rawResult;
  const res = await anthropic.messages.create({
    model: "claude-haiku-4-5",
    max_tokens: 512,
    messages: [
      {
        role: "user",
        content: `以下の検索結果から、トピック「${topic}」に関係する情報のみ抽出してください:\n\n${JSON.stringify(rawResult)}`,
      },
    ],
  });
  return res.content[0].type === "text" ? res.content[0].text : "";
});