Claude API × Convex: Reactive AI Apps — Data Flow, Streaming, and Agent Patterns

The Action/Mutation boundary you have to draw early

Convex Actions run in a Node.js environment and can make external calls. Mutations run in a V8 isolate and are expected to be fast and deterministic. Claude calls belong in Actions, full stop. Results come back to the database through ctx.runMutation().

// convex/messages.ts
import { action, mutation, query } from "./_generated/server";
import { v } from "convex/values";
import { api, internal } from "./_generated/api";
import Anthropic from "@anthropic-ai/sdk";
 
export const sendUserMessage = mutation({
  args: { threadId: v.id("threads"), text: v.string() },
  handler: async (ctx, args) => {
    // Insert the user message immediately so the UI reacts right away
    await ctx.db.insert("messages", {
      threadId: args.threadId,
      role: "user",
      content: args.text,
      status: "complete",
    });
    // Reserve a pending assistant row as a placeholder
    const assistantId = await ctx.db.insert("messages", {
      threadId: args.threadId,
      role: "assistant",
      contentDraft: "",
      status: "pending",
    });
    // Schedule the actual Claude call (do not await the work itself)
    await ctx.scheduler.runAfter(0, internal.messages.generateAssistantReply, {
      threadId: args.threadId,
      assistantId,
    });
    return { assistantId };
  },
});
 
export const generateAssistantReply = action({
  args: { threadId: v.id("threads"), assistantId: v.id("messages") },
  handler: async (ctx, args) => {
    const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY! });
    const history = await ctx.runQuery(internal.messages.getHistory, {
      threadId: args.threadId,
    });
    try {
      const response = await client.messages.create({
        model: "claude-sonnet-4-6",
        max_tokens: 4096,
        system: "You are a thoughtful, helpful assistant.",
        messages: history.map((m) => ({ role: m.role, content: m.content! })),
      });
      await ctx.runMutation(internal.messages.completeAssistant, {
        assistantId: args.assistantId,
        content: response.content[0].type === "text" ? response.content[0].text : "",
        inputTokens: response.usage.input_tokens,
        outputTokens: response.usage.output_tokens,
      });
    } catch (err) {
      await ctx.runMutation(internal.messages.failAssistant, {
        assistantId: args.assistantId,
        errorMessage: err instanceof Error ? err.message : String(err),
      });
    }
  },
});

Three things matter in this pattern. First, the Mutation only makes write decisions; the actual Claude work is enqueued with ctx.scheduler.runAfter(0, ...). That's what lets the UI render the user's message the instant they hit send. Second, the Action must catch every exception and write the failure back through another Mutation — otherwise pending messages accumulate forever and you lose any retry handle. Third, history lookup happens through ctx.runQuery(internal.messages.getHistory, ...). You cannot touch ctx.db directly inside an Action, and that's the intended design.

Streaming without drowning in writes

Chat UIs need to show text flowing in. Convex's reactive sync is great at this, but pushing a Mutation per character isn't viable — the wire cost explodes and you violate the determinism contract Mutations assume. The pattern I use is "buffer and flush on a short window." The Action receives the stream and writes back in batches every 100–200 ms.

export const generateStreamingReply = action({
  args: { threadId: v.id("threads"), assistantId: v.id("messages") },
  handler: async (ctx, args) => {
    const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY! });
    const history = await ctx.runQuery(internal.messages.getHistory, {
      threadId: args.threadId,
    });
    await ctx.runMutation(internal.messages.markStreaming, {
      assistantId: args.assistantId,
    });
 
    let buffer = "";
    let lastFlushAt = Date.now();
    let totalOutput = "";
 
    const flush = async () => {
      if (!buffer) return;
      totalOutput += buffer;
      await ctx.runMutation(internal.messages.appendDraft, {
        assistantId: args.assistantId,
        chunk: buffer,
      });
      buffer = "";
      lastFlushAt = Date.now();
    };
 
    try {
      const stream = await client.messages.stream({
        model: "claude-sonnet-4-6",
        max_tokens: 4096,
        messages: history.map((m) => ({ role: m.role, content: m.content! })),
      });
 
      for await (const event of stream) {
        if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
          buffer += event.delta.text;
          if (buffer.length > 64 || Date.now() - lastFlushAt > 150) {
            await flush();
          }
        }
      }
      await flush();
      const final = await stream.finalMessage();
      await ctx.runMutation(internal.messages.completeAssistant, {
        assistantId: args.assistantId,
        content: totalOutput,
        inputTokens: final.usage.input_tokens,
        outputTokens: final.usage.output_tokens,
      });
    } catch (err) {
      await ctx.runMutation(internal.messages.failAssistant, {
        assistantId: args.assistantId,
        errorMessage: err instanceof Error ? err.message : String(err),
      });
    }
  },
});

The heart of this is the flush condition: "at least 64 characters OR at least 150 ms since the last flush." Go smaller and Convex write cost plus perceived latency both creep up. Go larger and you lose the human feeling of watching someone type. Always call flush() one final time on exit so a dangling buffer doesn't get orphaned.

On the client side, your code is just useQuery over the messages table. The contentDraft updates arrive automatically. Not having to write SSE plumbing on the client is one of the nicest things about running AI on Convex.

Why tool calls deserve their own table

Most people reach for Tool Use by trying to embed tool info directly inside messages. In production that falls apart fast. Tool executions are the things that run long, fail, need retries, and deserve their own observability. I keep them in a separate toolCalls table that messages reference by ID.

export const runTool = internalAction({
  args: { toolCallId: v.id("toolCalls") },
  handler: async (ctx, args) => {
    const toolCall = await ctx.runQuery(internal.tools.getToolCall, {
      toolCallId: args.toolCallId,
    });
    await ctx.runMutation(internal.tools.markRunning, { toolCallId: args.toolCallId });
 
    try {
      let output: unknown;
      switch (toolCall.name) {
        case "web_search":
          output = await performWebSearch(toolCall.input);
          break;
        case "get_weather":
          output = await fetchWeather(toolCall.input);
          break;
        default:
          throw new Error(`Unknown tool: ${toolCall.name}`);
      }
      await ctx.runMutation(internal.tools.completeToolCall, {
        toolCallId: args.toolCallId,
        output,
      });
      // When all tools for this message have finished, continue generation
      await ctx.scheduler.runAfter(0, internal.messages.continueAfterTool, {
        messageId: toolCall.messageId,
      });
    } catch (err) {
      await ctx.runMutation(internal.tools.failToolCall, {
        toolCallId: args.toolCallId,
        errorMessage: err instanceof Error ? err.message : String(err),
      });
    }
  },
});

Three benefits come from this split. The UI can show inline progress per tool — "running," "succeeded," "failed" — without reparsing message contents. You can safely retry idempotent tool calls by rescheduling just the ones with status === "failure". And cost plus latency analysis becomes tractable: you learn which tool is actually the bottleneck, not the one you blamed.

If you've already read the advanced Tool Use patterns for the Claude API, this separation will feel familiar — Convex just makes it cleaner to express.

Agents that outlive a single Action

Convex Actions have a runtime limit (roughly 10 minutes on the free and Pro tiers as of April 2026, longer on Enterprise). A Claude Agent SDK loop with ten-plus turns won't fit in a single call. My implementation slices the work at "one agent turn = one Action" and uses ctx.scheduler.runAfter(0, ...) to self-recurse. Each turn persists state and, if another turn is needed, schedules a fresh Action.

export const agentStep = internalAction({
  args: { threadId: v.id("threads"), turn: v.number() },
  handler: async (ctx, args) => {
    if (args.turn > 20) {
      await ctx.runMutation(internal.threads.markAgentHalted, {
        threadId: args.threadId,
        reason: "max_turns_exceeded",
      });
      return;
    }
 
    const history = await ctx.runQuery(internal.messages.getHistory, {
      threadId: args.threadId,
    });
    const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY! });
    const response = await client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 4096,
      tools: agentTools,
      messages: history.map((m) => ({ role: m.role, content: m.content! })),
    });
 
    const hasToolUse = response.content.some((b) => b.type === "tool_use");
    if (hasToolUse) {
      await ctx.runMutation(internal.messages.registerToolCalls, {
        threadId: args.threadId,
        response,
      });
    } else {
      await ctx.runMutation(internal.messages.completeAgentResponse, {
        threadId: args.threadId,
        response,
      });
    }
  },
});

The quiet win here is deploy resilience. Because each turn is an independent Action, a Convex deploy that lands mid-loop doesn't kill the agent — the next turn simply runs on the fresh build. That property is how you avoid the "agent hangs whenever we deploy" stories that haunt monolithic worker setups.

If you want a point of comparison, my Claude Agent SDK × Temporal durable workflows guide describes a very different posture. My rule of thumb: Convex for chat-shaped agents where the user is in the loop, Temporal for big batch workflows where durability guarantees are the whole point.

Cost tracking and rate limiting, built in from day one

Claude pricing is a function of model and token count. If you don't accumulate per-user and per-thread token totals from the start, a surprise bill will teach you the hard way. That's why the schema above puts totalInputTokens and totalOutputTokens on the threads row.

When you accumulate tokens, always patch by delta — never overwrite.

export const accumulateTokens = internalMutation({
  args: {
    threadId: v.id("threads"),
    inputTokens: v.number(),
    outputTokens: v.number(),
  },
  handler: async (ctx, args) => {
    const thread = await ctx.db.get(args.threadId);
    if (!thread) return;
    await ctx.db.patch(args.threadId, {
      totalInputTokens: thread.totalInputTokens + args.inputTokens,
      totalOutputTokens: thread.totalOutputTokens + args.outputTokens,
    });
    // Notify the user once per thread if spend crosses a threshold
    const cost = estimateUsd(
      thread.totalInputTokens + args.inputTokens,
      thread.totalOutputTokens + args.outputTokens,
    );
    if (cost > 10 && thread.notifiedOver10 !== true) {
      await ctx.scheduler.runAfter(0, internal.notifications.notifyUser, {
        userId: thread.userId,
        message: `This thread has now used more than $10 in API calls.`,
      });
      await ctx.db.patch(args.threadId, { notifiedOver10: true });
    }
  },
});

Convex doesn't ship with primitives for rate limiting, so you reach for an external store (Upstash Redis is what I use) or implement a leaky bucket yourself. For "no more than 5 messages per minute per user," I enforce the check inside the Mutation — not the Action — so the user gets immediate feedback instead of waiting for an Action round-trip to fail.

Five pitfalls that bit me in production

1. Cold starts on the first call

Convex Actions run on Node.js and go cold when traffic is low. If the first Claude call of a session feels unreasonably slow, this is the usual suspect. Two practical mitigations: trigger a warm-up Action at the entry point of your app (for instance, right after thread creation), or share an Action across requests so a live one picks the work up. The warm-up is easier but doesn't fully solve the problem.

2. Missing awaits on ctx.scheduler and ctx.runMutation

When you forget to await ctx.scheduler.runAfter(0, ...), the schedule can be silently dropped if the parent Action exits first. Convex's docs are clear on this, but it's easy to miss under review pressure. I've added a custom ESLint rule to our CI that ensures every ctx.scheduler.* and ctx.runMutation(...) call site is awaited. It takes one afternoon to write and prevents a whole category of phantom bug.

3. v.any() leaking into production

Leaving tool input/output typed as v.any() feels convenient early on, but it hides schema drift. I use v.any() during prototyping and then tighten everything to explicit v.object({...}) before shipping. When you migrate, write a one-shot Mutation to backfill the existing rows into the new shape.

4. Orphaned streaming messages after disconnects

If a browser reload or network blip kills a stream, a message can sit at status: "streaming" forever. I run a scheduled cleanup to move anything stuck longer than 10 minutes to error.

export const cleanupStaleStreaming = internalMutation({
  handler: async (ctx) => {
    const threshold = Date.now() - 10 * 60 * 1000;
    const stale = await ctx.db
      .query("messages")
      .filter((q) => q.and(q.eq(q.field("status"), "streaming"), q.lt(q.field("_creationTime"), threshold)))
      .collect();
    for (const msg of stale) {
      await ctx.db.patch(msg._id, {
        status: "error",
        errorMessage: "Stream abandoned (timeout)",
      });
    }
  },
});

Register it in convex/crons.ts with cronJobs().interval("cleanup-stream", { minutes: 5 }, ...).

5. Ignoring Convex's pricing model when designing writes

Convex bills on function invocations and on bytes read/written. A naive streaming implementation that calls appendDraft on every delta will hit your soft cap alarmingly fast. The buffering strategy above isn't just a UX move — it's also a billing move.

Testing the pieces that talk to Claude

One reason people resist shipping AI code is that it feels untestable. I've come to see that framing as half-wrong — the Convex layer is easy to test; only the prompt-and-model layer is inherently fuzzy. Separating them helps.

The Mutations and Queries around your AI flow are pure by design, so they fit regular unit tests. Convex ships a test helper (convexTest) that runs your functions against an in-memory database. Use it to verify that sendUserMessage inserts both the user row and an assistant placeholder, that appendDraft is idempotent against repeated deltas, and that failAssistant transitions pending to error. These tests run in milliseconds and will catch 80% of the bugs you'd otherwise ship.

The first test I actually wrote was this one. Idempotency in appendDraft is what every interrupted stream will interrogate, so nailing it down early takes the fear out of everything built on top.

// convex/messages.test.ts
import { convexTest } from "convex-test";
import { expect, test } from "vitest";
import schema from "./schema";
import { api, internal } from "./_generated/api";
 
test("appendDraft does not duplicate content when the same delta arrives twice", async () => {
  const t = convexTest(schema);
  const threadId = await t.mutation(api.threads.create, { title: "t" });
  const { assistantId } = await t.mutation(api.messages.sendUserMessage, {
    threadId,
    text: "hello",
  });
 
  const delta = { seq: 1, text: "Hel" };
  await t.mutation(internal.messages.appendDraft, { assistantId, ...delta });
  await t.mutation(internal.messages.appendDraft, { assistantId, ...delta });
 
  const msg = await t.query(api.messages.get, { id: assistantId });
  expect(msg.content).toBe("Hel");
  expect(msg.status).toBe("streaming");
});

The mechanism is simply that appendDraft checks seq and silently drops any sequence number it has already absorbed. Carrying seq in the schema earns its keep for those two repeated lines in the test alone. Before I added it, a browser refresh mid-stream would double the assistant's text, and that shipped to production for several days before anyone told me. Working as an indie developer means the QA team is one person who already believes the code works, so the quiet failures are the ones that survive longest. I've stopped trying to catch them by looking harder and started putting them where a machine trips over them on every run.

Actions are harder because they call Claude. I wrap the SDK in a thin adapter that can be swapped for a fake in tests. The fake returns canned responses (or canned streams) and lets me assert that the Action serialized history correctly, handled tool-use blocks, and accumulated tokens through the right Mutation. That's the layer I put most of my energy into — it's where the shape of your AI feature actually lives.

For end-to-end tests, I run a small Playwright suite against a Convex preview deployment with a very cheap model (Claude Haiku 4.5 is usually enough) and assert on structural properties like "the assistant's response exists, has non-zero length, and status becomes complete within 30 seconds." Trying to assert on the content of a model's response will teach you a painful lesson about flakiness.

Monitoring and the release cadence that scales

Before going live, wire up a few things. The Convex dashboard gives you per-function execution time and error rates, which is a start but not enough. Add an errors table for Claude-side failures (overloaded, rate-limited, timed out), and hook an Action up to notify Slack or Discord so the team hears about them without polling. Those errors live in Claude's world, not Convex's, and you want them visible in their own lens.

Deploys happen via npx convex deploy, but the real gem is Preview Deploys. With "preview" configured in convex.json, every PR gets its own isolated backend. On any PR that touches Claude prompts or tool definitions, I do a few minutes of live chat on the preview environment before merging. That one habit has caught more "unexpected response" bugs than all my unit tests combined.

The point of the errors table is to carry only what the Convex dashboard cannot already tell you. I record which model was in play, what kind of failure it was, and which retry attempt it happened on. Function names and stack traces already live on the Convex side, so I don't duplicate them here.

// convex/errors.ts
import { internalMutation } from "./_generated/server";
import { v } from "convex/values";
 
export const record = internalMutation({
  args: {
    threadId: v.optional(v.id("threads")),
    model: v.string(),
    kind: v.union(
      v.literal("overloaded"),      // 529
      v.literal("rate_limited"),    // 429
      v.literal("timeout"),
      v.literal("invalid_request"), // 4xx from a bad payload
      v.literal("unknown"),
    ),
    attempt: v.number(),
    message: v.string(),
  },
  handler: async (ctx, args) => {
    await ctx.db.insert("errors", { ...args, at: Date.now() });
  },
});

Alert thresholds differ by kind, and that difference matters more than the table itself. overloaded reflects congestion upstream where retrying is the only lever I have, so it pages only past 20 events in a 15-minute window. invalid_request, on the other hand, usually means my own code or schema is wrong, so the very first one goes out. Before I drew that line I alerted on everything equally, Discord rang through every busy period, and within two weeks nobody was reading the channel — the exact outcome the alerting was supposed to prevent. An alert is worth having only if someone moves when it fires.

Logs stream in real time via convex logs, but for retention I ship them to Axiom. Paired with Convex's built-in HTTP logger, I build dashboards of Claude response-time distributions. That visibility is how you notice the slow drift of "things are getting a little slower" before it becomes "why is chat unusable today?"

Where to start

Thanks for reading. If you're about to try Convex with Claude, the single highest-leverage move is to paste the "Action and Mutation boundary" example above into a fresh Convex project and make it run. Five minutes in, you'll feel the Mutation → Action → Mutation loop click, and from that point on streaming, tool use, and agent loops all compose on the same rhythm.

For the mirror image of this design — the same experience built without Convex — building a scalable real-time AI chat server with the Claude API, WebSocket, and Redis Pub/Sub is the piece to read next. Seeing where the work moves when you own the reactive layer yourself is what makes both the case for Convex and the case against it concrete.