Claude API × Inngest — Durable AI Workflows with Retries, Idempotency, and Human Approval

# Node 20+ recommended
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",
  // The event key is read from env automatically in production.
});

// 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
# Open 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: "Search public sources and return the top 3 snippets.",
    input_schema: {
      type: "object",
      properties: { query: { type: "string" } },
      required: ["query"],
    },
  },
  {
    name: "send_summary_email",
    description: "Send the finished summary to a recipient.",
    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",
    // Cap parallelism so a runaway loop doesn't melt Anthropic's rate limit.
    concurrency: { limit: 5 },
    // Default retry budget for transient Anthropic errors.
    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: `Research and summarize this topic: ${topic}` },
    ];
 
    // Hard cap turns to prevent infinite loops.
    for (let turn = 0; turn < 6; turn++) {
      // Every Claude call is inside step.run, so reruns reuse the cached result.
      const response = await step.run(`claude-turn-${turn}`, async () => {
        return await anthropic.messages.create({
          model: "claude-sonnet-4-6",
          max_tokens: 1024,
          tools,
          messages,
        });
      });
 
      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) {
          // Splitting tool calls into their own steps lets one failed tool
          // resume independently of the others.
          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 };
    const idempotencyKey = `${ctx.jobId}:email:${hash(args)}`;
 
    // Pre-mark the operation as pending. If it already exists, we can short-circuit.
    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) {
  // Search API call (omitted)
  return [{ title: "...", snippet: "...", url: "..." }];
}
 
function hash(obj: unknown) {
  // Toy hash. Swap for SHA-256 via crypto.subtle in real code.
  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 (except 429) means the request itself is bad — don't waste retries.
      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; // Anything else, let Inngest's exponential backoff handle it.
  }
});

// Pause for human approval before charging large amounts.
if (toolName === "charge_customer" && (input as { amount: number }).amount > 50000) {
  await step.run("request-approval", async () => {
    return notifyApprovers({
      jobId,
      amount: (input as { amount: number }).amount,
      requestedBy,
    });
  });
 
  // Wait up to 24 hours for the approval event.
  const approval = await step.waitForEvent("wait-for-approval", {
    event: "agent/approval.received",
    timeout: "24h",
    if: `event.data.jobId == "${jobId}"`,
  });
 
  if (!approval) {
    throw new NonRetriableError("Approval timed out after 24h");
  }
  if (approval.data.decision !== "approved") {
    return { skipped: true, reason: "rejected by human" };
  }
  // Proceed with the actual charge.
}

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[] };
 
    // Fan out: one child function per document.
    const summaries = await Promise.all(
      docIds.map((id) =>
        step.invoke(`summarize-${id}`, {
          function: summarizeOne,
          data: { docId: id },
        }),
      ),
    );
 
    // Fan in: synthesize when every child has finished.
    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: `Compose a unified report from these summaries:\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: `Summarize this:\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;
});

// ❌ Dangerous: re-runs produce different IDs
const requestId = await step.run("save", async () => {
  const id = crypto.randomUUID(); // generated every retry
  await db.requests.create({ data: { id, /* ... */ } });
  return id;
});

// ✅ Safer
const requestId = crypto.randomUUID();
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: `Extract only information relevant to "${topic}" from these search results:\n\n${JSON.stringify(rawResult)}`,
      },
    ],
  });
  return res.content[0].type === "text" ? res.content[0].text : "";
});

// src/app/api/research/route.ts
import { NextResponse } from "next/server";
import { inngest } from "@/inngest/client";
import { auth } from "@/auth";
 
export async function POST(req: Request) {
  const session = await auth();
  if (!session?.user) return NextResponse.json({ error: "unauthenticated" }, { status: 401 });
 
  const { topic } = await req.json();
  if (typeof topic !== "string" || topic.length < 3) {
    return NextResponse.json({ error: "topic too short" }, { status: 400 });
  }
 
  const jobId = crypto.randomUUID();
  await inngest.send({
    name: "research/agent.requested",
    data: { topic, requestedBy: session.user.email!, jobId },
  });
 
  return NextResponse.json({ jobId, status: "queued" });
}

// inside the agent function
await step.run("checkpoint-start", () =>
  db.jobs.upsert({
    where: { id: jobId },
    create: { id: jobId, status: "running", topic, requestedBy },
    update: { status: "running" },
  }),
);
 
// at the end
await step.run("checkpoint-done", () =>
  db.jobs.update({
    where: { id: jobId },
    data: { status: "completed", completedAt: new Date() },
  }),
);

// test/agent.test.ts
import { describe, it, expect, vi } from "vitest";
import { runResearchAgent } from "@/inngest/functions/research-agent";
 
vi.mock("@/inngest/functions/tools", () => ({
  runTool: vi.fn(async (name: string) => {
    if (name === "web_search") return [{ title: "fixture", snippet: "test", url: "https://example.com" }];
    return { ok: true };
  }),
}));
 
describe("runResearchAgent", () => {
  it("converges within 6 turns and returns final messages", async () => {
    const ctx = makeFakeStepCtx();
    const result = await runResearchAgent.fn(ctx);
    expect(result.ok).toBe(true);
    expect(result.finalMessages.length).toBeGreaterThan(1);
    expect(ctx.steps.size).toBeLessThanOrEqual(15);
  });
});