⬡ API & SDK/2026-04-25上級

Claude API × Convex でリアクティブな AI アプリを本番構築する — データフロー設計とエージェントパターン

Convex のリアクティブデータベースと Claude API を組み合わせて、本番運用に耐えるチャット・エージェント系アプリを構築する方法をまとめました。スキーマ設計、Action/Mutation/Query の境界、ストリーミング、ツール状態管理、冷え始めの落とし穴まで、開発現場の判断軸を共有します。

Claude API¹¹⁵ Convex Agent SDK⁴ TypeScript²⁴ Realtime Production²

✦ プレミアム記事

Convex と Claude API を組み合わせたチャットアプリや業務エージェントを作り始めた方から、ここ数週間で似たような相談が続けて届きました。「マルチプレイヤーのチャットは気持ち良く動くのに、Claude を呼ぶと突然レイテンシが倍になる」「Action のタイムアウトで長時間処理が落ちる」「リロードすると途中まで届いたツール実行結果が消える」。いずれも、Convex のリアクティブな世界観と、Claude の非同期・ストリーミング前提のリクエストを無理に1層で混ぜてしまったときに起きる症状です。

ここで扱うのは私が本番で運用している設計を元に、Convex と Claude API を噛み合わせるための層の切り分け方、スキーマ設計、ストリーミング同期、ツール状態管理、そして見落としがちな冷え始めレイテンシ対策までをまとめます。コードは TypeScript、@anthropic-ai/sdk と convex の最新安定版（2026 年4月時点）を前提にしていますが、どのバージョンでも通用する設計原則に重点を置いています。

Convex を AI アプリの基盤に選ぶときの判断軸

Convex は「リアクティブなクエリを中心に据えた、ドキュメントベースのサーバーレスプラットフォーム」です。単なるデータベースではなく、クエリが自動で subscribe され、書き込みによる影響範囲が差分で配信される点が特徴です。この性質は、AI との対話 UI と驚くほど相性が良く、新しいメッセージが追加されるたびに UI を更新する処理を自前で書く必要がありません。

ただし、Convex にはサーバーレス特有の制約もあります。mutation と query には CPU 時間の上限があり、外部 API を呼ぶのは原則として action 経由です。Claude API のように応答に 5〜60 秒かかる呼び出しは、常に action の層で扱い、結果を mutation で書き戻すパターンが基本になります。「Mutation から fetch したい」という欲求は早めに諦めてください。この境界を曖昧にすると、後から巨大なリファクタが必要になります。

私は Convex を選ぶかどうかを、以下の観点で判断しています。クライアント側でリアルタイム同期が必要で、かつユーザー同士の協調が発生する（共同編集・マルチプレイヤーなど）なら Convex が強力です。一方、単純な単一ユーザーのチャット UI で、既に Postgres と GraphQL が動いているなら、Convex を追加する必然性は高くありません。Claude API のストリーミング配信を SSE でダイレクトに送る方が、レイテンシは低く保てます。

スキーマ設計 — スレッド・メッセージ・ツールコールの3層モデル

私が本番で使っているスキーマは、チャット UI とエージェント実行を両方想定しています。最小構成でも、スレッド、メッセージ、ツール実行の3テーブルに分離するのがおすすめです。これは後述する「進行中の処理を別レコードで追う」パターンに直結します。

// convex/schema.ts
import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";
 
export default defineSchema({
  threads: defineTable({
    userId: v.string(),
    title: v.string(),
    model: v.string(), // "claude-sonnet-4-6" など
    systemPrompt: v.optional(v.string()),
    totalInputTokens: v.number(),
    totalOutputTokens: v.number(),
    archivedAt: v.optional(v.number()),
  }).index("byUser", ["userId", "archivedAt"]),
 
  messages: defineTable({
    threadId: v.id("threads"),
    role: v.union(v.literal("user"), v.literal("assistant"), v.literal("tool")),
    // ストリーミング中は contentDraft を更新し続け、確定後に content に移す
    content: v.optional(v.string()),
    contentDraft: v.optional(v.string()),
    status: v.union(
      v.literal("pending"),
      v.literal("streaming"),
      v.literal("complete"),
      v.literal("error"),
    ),
    errorMessage: v.optional(v.string()),
    inputTokens: v.optional(v.number()),
    outputTokens: v.optional(v.number()),
  }).index("byThread", ["threadId"]),
 
  toolCalls: defineTable({
    messageId: v.id("messages"),
    name: v.string(),
    input: v.any(),
    output: v.optional(v.any()),
    status: v.union(
      v.literal("pending"),
      v.literal("running"),
      v.literal("success"),
      v.literal("failure"),
    ),
    startedAt: v.number(),
    completedAt: v.optional(v.number()),
  }).index("byMessage", ["messageId"]),
});

contentDraft と content を分けるのは、「完了前の下書き」と「確定結果」を同じフィールドで管理しないためです。クライアントが content だけを購読するようにすれば、ストリーミング中の細かい再描画で UI がちらつく問題を回避できます。逆に「タイピング中のカーソル」を表示したい場合は、別コンポーネントで contentDraft を購読させれば良い、という分離が効きます。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦Convex + Claude API でチャット・エージェントを作り始めた人が、Action と Mutation の境界で詰まる問題を最初に解決できる

✦長時間実行するエージェントをサーバーレスで安定稼働させるスケジュール型ジョブと進行状況同期の実装を習得できる

✦コスト追跡・レート制限・スキーマ進化を含む本番運用の設計をそのままコピーして使える形で手に入る

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

Action と Mutation の境界設計

Convex の action は Node.js 環境で動き、外部 API を呼べます。一方 mutation は V8 isolate で動き、決定的・短時間であることが期待されます。Claude の呼び出しは必ず action で行い、結果を 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) => {
    // ユーザーメッセージを即座に書き込む（UIは反応的に更新される）
    await ctx.db.insert("messages", {
      threadId: args.threadId,
      role: "user",
      content: args.text,
      status: "complete",
    });
    // プレースホルダーとして assistant 行を pending で用意する
    const assistantId = await ctx.db.insert("messages", {
      threadId: args.threadId,
      role: "assistant",
      contentDraft: "",
      status: "pending",
    });
    // action にスケジュールする（ここでは await しない）
    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: "あなたは誠実で丁寧なアシスタントです。",
        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),
      });
    }
  },
});

このパターンの肝は3つあります。第一に、Mutation は「書き込みの意思決定」だけを担当し、実際の Claude 呼び出しは ctx.scheduler.runAfter(0, ...) でキューイングする点。これにより、ユーザーが「送信」を押した瞬間にクライアント側の UI は新しいメッセージを即座に表示できます。第二に、Action 内部では例外を必ず捕捉し、失敗を別 Mutation で書き戻すこと。これを怠ると、pending 状態のメッセージが延々と残り、再試行のトリガーが失われます。第三に、履歴の取得は ctx.runQuery(internal.messages.getHistory, ...) で内部 Query を叩くこと。Action 内で ctx.db を直接触ることは設計上できません。

ストリーミングレスポンスをどう同期するか

チャット UI では、Claude の応答が数秒単位で流れてくる様子を見せたいはずです。Convex のリアクティブ同期は強力ですが、一文字ごとに Mutation を叩くのは現実的ではありません。ワイヤーコストが増大し、決定論性の要件にも反します。

私が採用しているのは、「バッファリングして短い間隔で Mutation」パターンです。Action 内でストリーミングを受信しながら、100〜200ms 単位でまとめて書き戻します。

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),
      });
    }
  },
});

このコードの重要なポイントは、バッファを「64文字以上」または「150ms 以上経過」の条件で flush することです。短すぎると Convex の書き込み料金とレイテンシが跳ね上がり、長すぎると「人が書いている感覚」が失われます。また、最後に必ず flush() を呼んで、半端なバッファを書き漏らさないようにします。

クライアント側は、Convex の useQuery で messages テーブルを購読するだけで、自動的に contentDraft の更新が反映されます。特別な SSE ハンドラを書く必要がないのは、Convex を使う最大の恩恵のひとつです。

ツール実行の状態管理 — なぜ別テーブルが必要か

Claude の Tool Use を実装するとき、多くの人は最初「メッセージ本体にツール情報を埋め込めばいい」と考えます。しかし本番運用では、ツール実行だけが長時間かかる・失敗する・リトライしたくなる、という要件が頻発します。私は、ツール実行は toolCalls テーブルに独立させ、メッセージはそれを参照する設計を推しています。

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,
      });
      // ツール結果が揃ったら、再度 Claude を呼び出して続きを生成
      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),
      });
    }
  },
});

独立テーブルにする利点は3つあります。ひとつは、UI で「このツールは実行中」「このツールは失敗」とインラインで表示できること。ふたつめは、同じツール呼び出しを冪等に再試行できること。status が failure のものだけを再スケジュールする運用が可能です。みっつめは、コスト・レイテンシ分析が個別にできること。どのツールがボトルネックかを後から検証できます。

Claude API の高度な Tool Use パターンをすでに読まれている方には馴染みのある設計だと思いますが、Convex ではこの分離が特に活きます。

長時間実行エージェントをどう扱うか

Convex の action には実行時間の上限があります（2026年4月時点で無料/Pro プランは概ね10分、Enterprise で延長可能）。Claude Agent SDK で10ターン以上のループを回す場合、単発の Action では完結しません。

私の実装では、「エージェント1ターン = 1 Action」の粒度に分解し、ctx.scheduler.runAfter(0, ...) で自己再帰的にスケジュールします。各ターンで状態を DB に書き戻し、次のターンが必要なら新しい 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) {
      // ツール呼び出しを toolCalls に登録し、完了後に次のターンを呼ぶ
      await ctx.runMutation(internal.messages.registerToolCalls, {
        threadId: args.threadId,
        response,
      });
    } else {
      await ctx.runMutation(internal.messages.completeAgentResponse, {
        threadId: args.threadId,
        response,
      });
    }
  },
});

この設計の美点は、各ターンが独立しているので途中で Convex のデプロイがあっても、次のターンは新しいビルドで実行されることです。「デプロイ中に走っているエージェントが途中で止まる」問題を回避できます。

長時間実行の設計パターンについては、Claude Agent SDK × Temporal 製の耐障害性ワークフローと比較すると、Convex は「フルマネージドで楽だが、Workflow ロジックを厳格に管理する力は弱い」という違いがあります。私は、チャット由来のエージェントには Convex、バッチ的な大規模ワークフローには Temporal、と使い分けています。

コスト追跡とレート制限を運用視点で組み込む

Claude API の料金はモデルとトークン数で決まります。本番運用では、ユーザー単位・スレッド単位で入出力トークンを蓄積して可視化しておかないと、思わぬ高額請求に後から気付くことになります。先ほどのスキーマで threads に totalInputTokens と totalOutputTokens を持たせているのは、このためです。

Mutation でトークンを加算する際は、必ず差分更新するのが肝心です。

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,
    });
    // 閾値を超えたらユーザーにアラート送信など
    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: `このスレッドの API 利用料が $10 を超えました。`,
      });
      await ctx.db.patch(args.threadId, { notifiedOver10: true });
    }
  },
});

レート制限は、Convex の組み込み機能では対応できないため、外部ストア（Upstash Redis など）を使うか、自前のリーキーバケット実装をアクションに挟みます。私は、ユーザー単位で「直近1分間に5メッセージ」の制限を Upstash Redis のカウンタで実装し、超過時は mutation 側で拒否してユーザーに即座にフィードバックを返しています。これはアクション層で書くと、ユーザー体験が遅くなります。

本番でつまずく5つの落とし穴

1. 冷え始め（cold start）による初回レイテンシ

Convex の Action は Node.js 環境で走りますが、利用頻度が低いと冷え始めから起動します。Claude の初回呼び出しが「やけに遅い」と感じたら、ここが疑わしいです。対策は2つあります。ひとつは、アプリの入口（例えばスレッド作成直後）に「ウォームアップ用の空呼び出し」を挟むこと。もうひとつは、Action を共有する設計にして、生き続けている Action に処理を流し込むことです。前者は簡単ですが、根本解決にはなりません。

2. Action 内での await 忘れによる静かな失敗

ctx.scheduler.runAfter(0, ...) を await せずに書くと、親 Action が終了した瞬間にスケジュールが取り消されるケースがあります。Convex のドキュメントには明記されていますが、実装者が見落としがちです。私は CI に eslint のカスタムルールを入れて、ctx.scheduler と ctx.runMutation の呼び出しが必ず await されているかチェックしています。

3. スキーマ進化時の v.any() 罠

ツール入出力を v.any() で定義していると、スキーマバージョン間の互換性が崩れても気付けません。私は開発初期こそ v.any() を使いますが、本番投入前には v.object({...}) で明示的なスキーマに書き換えます。進化時は、移行用の Mutation を書いて既存データを新しい形に変換する必要があります。

4. ストリーミング中断時の孤立レコード

ネットワーク切断やブラウザリロードで Action が中断されたとき、status: "streaming" のメッセージが残り続けます。私は別途 cron を走らせて、「10分以上 streaming のまま」のメッセージを 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)",
      });
    }
  },
});

convex/crons.ts で cronJobs().interval("cleanup-stream", { minutes: 5 }, ...) として登録します。

5. Convex の料金体系を無視した設計

Convex は「関数呼び出し回数」と「DB 読み書きバイト数」で課金されます。ストリーミング中に過剰な appendDraft を走らせると、あっという間に閾値を超えます。前述のバッファリング戦略は、機能だけでなく料金対策でもあります。

Claude 連携部分のテスト戦略

「AI を組み込んだコードは書きにくい」と感じるのは、実は半分は思い込みだと私は思っています。Convex 層は十分テスト可能で、本当に曖昧なのはプロンプトとモデル応答の層だけです。両者を分けて考えると、書ける範囲が見えてきます。

Mutation と Query は純粋関数として設計されているので、通常のユニットテストに適しています。Convex には convexTest というヘルパーがあり、インメモリ DB に対して関数を実行できます。これを使って、sendUserMessage がユーザー行と assistant のプレースホルダーの両方を挿入すること、appendDraft が同じデルタを2回受けても冪等であること、failAssistant が pending から error へ正しく遷移することなどを検証します。ミリ秒単位で走るので、CI に載せてもストレスになりません。

Action はどうしても Claude を呼ぶので、SDK を薄いアダプター越しに使い、テストではフェイクに差し替える設計にします。フェイクは決め打ちの応答やストリームを返し、Action が履歴を正しくシリアライズしているか、ツール用ブロックを正しく扱っているか、トークンを正しい Mutation 経由で積み上げているかを検証します。ここが AI 機能の本体なので、一番エネルギーを割く価値があります。

E2E は Convex の Preview Deploy に対して、軽量モデル（Claude Haiku 4.5 で十分なことが多いです）を使った小さな Playwright スイートを走らせています。アサーションは「assistant メッセージが存在し、空ではなく、30 秒以内に status が complete になる」といった構造面にとどめます。応答の「中身」を直接アサートしようとすると、フレーキーテストの沼にはまることになります。

モニタリングと本番デプロイのリズム

本番投入前には、以下を最低限仕込んでおくことをおすすめします。Convex のダッシュボードで関数ごとの実行時間とエラー率を見られますが、これだけでは不十分です。Claude API 側のエラー（オーバーロード、レート制限、タイムアウト）を独立して記録するための errors テーブルを持ち、Slack / Discord に通知する action を繋げておきます。

デプロイは npx convex deploy で行いますが、重要なのは Preview Deploy を使えばす。convex.json の "preview" 設定で、PR ごとに独立したバックエンドが立ち上がります。私は、Claude のプロンプトやツール定義を変更した PR では、必ず preview 環境で数分間の対話テストをしてから main にマージしています。このひと手間が、「本番で想定外の応答が返ってくる」事故を大きく減らしてくれます。

ログは convex logs でリアルタイム確認できますが、長期保管には Datadog や Axiom へ転送する設定を入れておきます。私は Axiom を使っており、Convex の built-in HTTP logger と組み合わせて、Claude の応答時間分布をダッシュボード化しています。この可視化があると、「最近レイテンシが伸びている」といった微細な変化に早く気付けます。

結びに — 最初に触るべきひとつの実装

ここまで読んでいただきありがとうございました。もし Convex と Claude の組み合わせをこれから試されるなら、最初に手を動かすべきなのは、この記事の「Action と Mutation の境界設計」のサンプルをそのままコピーして動かしてみることだと思います。5分で動き始めますし、Mutation → Action → Mutation のリングが動くことが体感できれば、あとはストリーミングもエージェントループも、同じリズムの拡張として自然に書けるようになります。

この記事の実装を深めたい方には、Claude API を使ったマルチエージェント本番パターンが参考になります。Convex を選ばない場合の設計比較としても役立つはずです。