Skip to Content
ドキュメントワークフローInngest Workflows | Workflows | Mastra Docs

Inngest ワークフロー

Inngest は、インフラストラクチャを管理することなく、バックグラウンドワークフローを構築・実行するための開発者プラットフォームです。

Mastraにおけるinngestの仕組み

InngestとMastraは、ワークフローモデルを整合させることで統合されています：Inngestはロジックをステップで構成される関数に整理し、createWorkflowとcreateStepを使用して定義されるMastraのワークフローはこのパラダイムに直接マッピングされます。各Mastraワークフローは一意の識別子を持つInngest関数となり、ワークフロー内の各ステップはInngestのステップにマッピングされます。

serve関数は、MastraワークフローをInngest関数として登録し、実行とモニタリングに必要なイベントハンドラーを設定することで、両システムを橋渡しします。

イベントがワークフローをトリガーすると、Inngestはステップごとに実行し、各ステップの結果をメモ化します。これにより、ワークフローが再試行または再開された場合、完了したステップはスキップされ、効率的で信頼性の高い実行が保証されます。Mastraのループ、条件分岐、ネストされたワークフローなどの制御フロープリミティブは、Inngestの関数/ステップモデルにシームレスに変換され、コンポジション、分岐、一時停止などの高度なワークフロー機能が保持されます。

リアルタイムモニタリング、一時停止/再開、ステップレベルの可観測性は、Inngestのパブリッシュ-サブスクライブシステムとダッシュボードを通じて実現されます。各ステップが実行されると、その状態と出力はMastraストレージを使用して追跡され、必要に応じて再開することができます。

セットアップ

npm install @mastra/inngest @mastra/core @mastra/deployer

Inngest Workflowの構築

このガイドでは、InngestとMastraを使用してワークフローを作成する方法を説明し、値を10に達するまでインクリメントするカウンターアプリケーションのデモンストレーションを行います。

Inngestの初期化

Inngest統合を初期化して、Mastra互換のワークフローヘルパーを取得します。createWorkflowとcreateStep関数は、MastraとInngestに互換性のあるワークフローとステップオブジェクトを作成するために使用されます。

開発環境では

src/mastra/inngest/index.ts
import { Inngest } from "inngest";
import { realtimeMiddleware } from "@inngest/realtime";
 
export const inngest = new Inngest({
  id: "mastra",
  baseUrl:"http://localhost:8288",
  isDev: true,
  middleware: [realtimeMiddleware()],
});

本番環境では

src/mastra/inngest/index.ts
import { Inngest } from "inngest";
import { realtimeMiddleware } from "@inngest/realtime";
 
export const inngest = new Inngest({
  id: "mastra",
  middleware: [realtimeMiddleware()],
});

ステップの作成

ワークフローを構成する個々のステップを定義します：

src/mastra/workflows/index.ts
import { z } from "zod";
import { inngest } from "../inngest";
import { init } from "@mastra/inngest";
 
// ローカルのInngestサーバーを指すように、MastraでInngestを初期化
const { createWorkflow, createStep } = init(inngest);
 
// ステップ: カウンター値をインクリメント
const incrementStep = createStep({
  id: "increment",
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    return { value: inputData.value + 1 };
  },
});

ワークフローの作成

dountilループパターンを使用してステップをワークフローに構成します。createWorkflow関数は、Inngestサーバー上で呼び出し可能な関数を作成します。

src/mastra/workflows/index.ts
// Inngestサーバー上で関数として登録されるワークフロー
const workflow = createWorkflow({
  id: "increment-workflow",
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
}).then(incrementStep);
 
workflow.commit();
 
export { workflow as incrementWorkflow };

Mastraインスタンスの設定とワークフローの実行

ワークフローをMastraに登録し、Inngest APIエンドポイントを設定します：

src/mastra/index.ts
import { Mastra } from "@mastra/core/mastra";
import { serve as inngestServe } from "@mastra/inngest";
import { incrementWorkflow } from "./workflows";
import { inngest } from "./inngest";
import { PinoLogger } from "@mastra/loggers";
 
// ワークフローとInngest APIエンドポイントでMastraを設定
export const mastra = new Mastra({
  workflows: {
    incrementWorkflow,
  },
  server: {
    // ローカルDockerコンテナがMastraサーバーに接続できるようにするためのサーバー設定が必要
    host: "0.0.0.0",
    apiRoutes: [
      // このAPIルートは、InngestサーバーにMastraワークフロー（Inngest関数）を登録するために使用される
      {
        path: "/api/inngest",
        method: "ALL",
        createHandler: async ({ mastra }) => inngestServe({ mastra, inngest }),
        // inngestServe関数は以下によってMastraワークフローをInngestと統合します：
        // 1. 一意のID（workflow.${workflowId}）を持つ各ワークフローのInngest関数を作成
        // 2. 以下を行うイベントハンドラーを設定：
        //    - 各ワークフロー実行に対して一意の実行IDを生成
        //    - ステップ実行を管理するInngestExecutionEngineを作成
        //    - ワークフロー状態の永続化とリアルタイム更新を処理
        // 3. workflow:${workflowId}:${runId}チャンネルを通じて
        //    リアルタイム監視のためのパブリッシュ・サブスクライブシステムを確立
      },
    ],
  },
  logger: new PinoLogger({
    name: "Mastra",
    level: "info",
  }),
});

ワークフローのローカル実行

前提条件:

  • Dockerがインストールされ、実行されている
  • Mastraプロジェクトがセットアップされている
  • 依存関係がインストールされている（npm install）
  1. npx mastra devを実行して、ポート4111でサーバーを提供するためにローカルでMastraサーバーを起動します。
  2. Inngest Dev Server（Docker経由）を開始 新しいターミナルで以下を実行：
docker run --rm -p 8288:8288 \
  inngest/inngest \
  inngest dev -u http://host.docker.internal:4111/inngest/api

注意: -u の後のURLは、Inngest dev serverにMastraの /api/inngest エンドポイントの場所を伝えます。

  1. Inngest Dashboardを開く
  • ブラウザで http://localhost:8288  にアクセスします。
  • サイドバーの Apps セクションに移動します。
  • Mastraワークフローが登録されているのが確認できるはずです。 Inngest Dashboard
  1. ワークフローを実行する
  • サイドバーの Functions セクションに移動します。
  • Mastraワークフローを選択します。
  • Invoke をクリックして、以下の入力を使用します：
{
  "data": {
    "inputData": {
      "value": 5
    }
  }
}

Inngest Function

  1. ワークフローの実行を監視する
  • サイドバーの Runs タブに移動します。
  • 最新の実行をクリックして、ステップバイステップの実行進捗を確認します。 Inngest Function Run

本番環境でのワークフロー実行

前提条件:

  • Vercelアカウントとインストール済みのVercel CLI (npm i -g vercel)
  • Inngestアカウント
  • Vercelトークン（推奨：環境変数として設定）
  1. MastraインスタンスにVercel Deployerを追加
src/mastra/index.ts
import { VercelDeployer } from "@mastra/deployer-vercel";
 
export const mastra = new Mastra({
  // ...その他の設定
  deployer: new VercelDeployer({
    teamSlug: "your_team_slug",
    projectName: "your_project_name",
    // vercelトークンは、右上のユーザーアイコンをクリックし、
    // 「Account Settings」をクリックして、左サイドバーの「Tokens」をクリックすることで
    // vercelダッシュボードから取得できます。
    token: "your_vercel_token",
  }),
});

注意: Vercelトークンを環境に設定してください：

export VERCEL_TOKEN=your_vercel_token
  1. mastraインスタンスをビルド
npx mastra build
  1. Vercelにデプロイ
cd .mastra/output
vercel --prod

ヒント: まだの場合は、vercel login でVercel CLIにログインしてください。

  1. Inngest Dashboardと同期
  • Inngest dashboard  に移動します。
  • Sync new app with Vercel をクリックして、指示に従います。
  • Mastraワークフローがアプリとして登録されているのが確認できるはずです。 Inngest Dashboard
  1. ワークフローを実行
  • Functions セクションで、workflow.increment-workflow を選択します。
  • All actions（右上）> Invoke をクリックします。
  • 以下の入力を提供します：
{
  "data": {
    "inputData": {
      "value": 5
    }
  }
}

Inngest Function Run

  1. 実行を監視
  • Runs タブに移動します。
  • 最新の実行をクリックして、ステップバイステップの実行進捗を確認します。 Inngest Function Run
ワークフローにおけるエラー処理 | Mastra ドキュメントワークフローを使った入力データマッピング | Mastra ドキュメント