Inngest Workflow

Inngest 是一个用于构建和运行后台 Workflow 的开发者平台，无需管理基础设施。

包含高级流控制功能的完整示例，请参阅 Inngest Workflow 示例。

Inngest 与 Mastra 的工作原理

Inngest 与 Mastra 通过对齐 Workflow 模型进行集成：Inngest 将逻辑组织成由步骤组成的函数，而使用 createWorkflow() 和 createStep() 定义的 Mastra Workflow 会直接映射到该结构。每个 Mastra Workflow 都会成为具有唯一标识符的 Inngest 函数，Workflow 中的每个步骤则映射到一个 Inngest 步骤。

serve() 函数通过将 Mastra Workflow 注册为 Inngest 函数，并设置执行和监控所需的事件 handler，在两个系统之间建立桥梁。

当事件触发 Workflow 时，Inngest 会逐步执行并 memoize 每个结果。重试或恢复时，Inngest 会根据保存的结果跳过已完成的步骤。Mastra 控制流原语（例如循环、条件和嵌套 Workflow）会映射到同一个 Inngest 函数和步骤模型，同时保留组合、分支和暂停行为。

Inngest 的发布/订阅系统和仪表盘支持实时监控、暂停/恢复和步骤级 observability。每个步骤执行时，其状态和输出会使用 Mastra Storage 进行跟踪，并可按需恢复。

设置

安装所需包：

npm

npm install @mastra/inngest@latest inngest

pnpm

pnpm add @mastra/inngest@latest inngest

Yarn

yarn add @mastra/inngest@latest inngest

Bun

bun add @mastra/inngest@latest inngest

备注

需要 inngest@^4 和 Inngest Dev Server v1.18.0 或更高版本。v4 已在 SDK 中内置 Realtime，因此不再使用 @inngest/realtime 和 realtimeMiddleware。

构建 Inngest Workflow

本指南逐步介绍如何使用 Inngest 和 Mastra 创建 Workflow，并通过一个将数值递增到 10 的计数器应用进行演示。

初始化 Inngest

初始化 Inngest 集成，以获取与 Mastra 兼容的 Workflow 辅助函数。createWorkflow() 和 createStep() 函数用于创建兼容 Mastra 与 Inngest 的 Workflow 和步骤对象。

在开发环境中：

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

export const inngest = new Inngest({
  id: 'mastra',
  baseUrl: 'http://localhost:8288',
  isDev: true,
})

在生产环境中：

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

export const inngest = new Inngest({
  id: 'mastra',
})

使用 INNGEST_DEV 选择开发或云端模式

Inngest SDK 以开发或云端两种模式之一运行。从 v4 开始，SDK 默认为云端模式，需要 INNGEST_EVENT_KEY 和 INNGEST_SIGNING_KEY。上述示例在代码中使用 isDev: true 设置模式。也可以设置 INNGEST_DEV 环境变量，使同一份 Client 代码能在不同环境中工作：

• INNGEST_DEV=1：强制使用开发模式。SDK 与本地 Inngest Dev Server 通信并禁用签名验证。
• INNGEST_DEV=0：强制使用云端模式。SDK 与 Inngest Cloud 通信，需要事件 Key 和签名 Key。
• INNGEST_DEV=<url>：强制使用开发模式，并将 SDK 指向 <url> 处的 Dev Server，例如 http://localhost:8288。
• 未设置：默认为云端模式。

设置 INNGEST_DEV 后，可以从 Client 中移除 isDev 和 baseUrl：

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

export const inngest = new Inngest({
  id: 'mastra',
})

注意

请勿在生产环境中设置 INNGEST_DEV。该变量会禁用签名验证，而安全接收 Inngest Cloud 事件需要签名验证。

同时设置二者时，new Inngest() 上的 isDev 选项会覆盖 INNGEST_DEV。

创建步骤

定义组成 Workflow 的各个步骤：

src/mastra/workflows/index.ts

import { z } from 'zod'
import { inngest } from '../inngest'
import { init } from '@mastra/inngest'

// Initialize Inngest with Mastra, pointing to your local Inngest server
const { createWorkflow, createStep } = init(inngest)

// Step: Increment the counter value
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 }
  },
})

创建 Workflow

使用 dountil 循环模式将步骤组合成 Workflow。createWorkflow() 函数会在 Inngest Server 上创建一个可调用函数。

src/mastra/workflows/index.ts

// workflow that is registered as a function on inngest server
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 注册 Workflow，并配置 Inngest API 端点：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { serve } from '@mastra/inngest'
import { incrementWorkflow } from './workflows'
import { inngest } from './inngest'
import { PinoLogger } from '@mastra/loggers'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  server: {
    host: '0.0.0.0',
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => {
          return serve({ mastra, inngest })
        },
      },
    ],
  },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})

备注

路径是 /inngest/api，而不是 /api/inngest。Mastra 为内置路由（Agent、Workflow、Memory）保留 /api 前缀。以 Server 的 apiPrefix（默认为 /api）开头的自定义 apiRoutes 路径会在启动时抛出错误。相关背景请参阅 #15743；如果需要保留 /api/inngest，请直接前往使用自定义 apiPrefix。

运行 Workflow

在本地运行

1. 运行 npx mastra dev，在本地 4111 端口启动 Mastra Server

2. 启动 Inngest Dev Server。在新终端中运行：

   npx inngest-cli@latest dev -u http://localhost:4111/inngest/api

   备注

   -u 后的 URL 告知 Inngest Dev Server 在何处查找 Mastra /inngest/api 端点

3. 在 http://localhost:8288 打开 Inngest 仪表盘，前往边栏中的 Apps 部分，验证 Mastra Workflow 是否已注册

4. 在 Functions 中打开 Workflow。选择 Invoke，然后提供以下输入：

   {
     "data": {
       "inputData": {
         "value": 5
       }
     }
   }

5. 在 Runs 选项卡中监控 Workflow 执行，查看逐步执行进度

在生产环境中运行

开始之前，请确保具备：

• Vercel 账户且已安装 Vercel CLI（npm i -g vercel）
• Inngest 账户
• Vercel token

1. 在环境中设置 Vercel token：

   .env

   export VERCEL_TOKEN=your_vercel_token

2. 将 VercelDeployer 添加到 Mastra 实例

   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',
       // you can get your vercel token from the vercel dashboard by clicking on the user icon in the top right corner
       // and then clicking on "Account Settings" and then clicking on "Tokens" on the left sidebar.
       token: process.env.VERCEL_TOKEN,
     }),
   })

3. 构建 Mastra 实例

   npx mastra build

4. 部署到 Vercel

   cd .mastra/output
   vercel login
   vercel --prod

5. 选择 Sync new app with Vercel 并按照说明操作，与 Inngest 仪表盘同步

   注意

   Inngest 的自动发现约定假定路径为 /api/inngest。由于本指南使用 /inngest/api，请将 Inngest 应用的 URL 字段设置为部署来源加 /inngest/api（例如 https://your-app.vercel.app/inngest/api）。如果保留默认值，Inngest 仪表盘将找不到应用函数。

6. 在 Functions 中打开 workflow.increment-workflow。选择 All actions > Invoke，然后提供以下输入：

   {
     "data": {
       "inputData": {
         "value": 5
       }
     }
   }

7. 在 Runs 选项卡中监控执行，查看逐步进度

添加自定义 Inngest 函数

通过使用 serve() 的可选 functions 参数，可以在 Mastra Workflow 之外提供额外的 Inngest 函数。

创建自定义函数

首先，创建自定义 Inngest 函数：

src/inngest/custom-functions.ts

import { inngest } from '../inngest'

// Define custom Inngest functions
export const customEmailFunction = inngest.createFunction(
  { id: 'send-welcome-email' },
  { event: 'user/registered' },
  async ({ event }) => {
    // Custom email logic here
    console.log(`Sending welcome email to ${event.data.email}`)
    return { status: 'email_sent' }
  },
)

export const customWebhookFunction = inngest.createFunction(
  { id: 'process-webhook' },
  { event: 'webhook/received' },
  async ({ event }) => {
    // Custom webhook processing
    console.log(`Processing webhook: ${event.data.type}`)
    return { processed: true }
  },
)

与 Workflow 一同提供自定义函数

更新 Mastra 配置，导入并包含自定义函数。高亮行显示了新增内容：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { serve } from '@mastra/inngest'
import { incrementWorkflow } from './workflows'
import { inngest } from './inngest'
import { customEmailFunction, customWebhookFunction } from './inngest/custom-functions'
import { PinoLogger } from '@mastra/loggers'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  server: {
    host: '0.0.0.0',
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => {
          return serve({
            mastra,
            inngest,
            functions: [customEmailFunction, customWebhookFunction],
          })
        },
      },
    ],
  },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})

函数注册

包含自定义函数时：

1. Mastra Workflow 会自动转换为 ID 类似 workflow.${workflowId} 的 Inngest 函数
2. 自定义函数保留指定的 ID（例如 send-welcome-email、process-webhook）
3. 所有函数都通过同一个 /inngest/api 端点一同提供

这样便可以将 Mastra 的 Workflow 编排与现有 Inngest 函数结合使用。

与其他框架配合使用

默认 serve 函数在内部使用 Hono。如果使用 Express、Fastify 或 Koa 等其他 Web 框架，请将 createServe 工厂函数与相应的 Inngest 适配器配合使用。

Express

src/server.ts

import express from 'express'
import { createServe } from '@mastra/inngest'
import { serve as expressAdapter } from 'inngest/express'
import { mastra, inngest } from './mastra'

const app = express()

// Body parsing middleware required for Inngest
app.use(express.json())

const handler = createServe(expressAdapter)({ mastra, inngest })
app.use('/inngest/api', handler)

app.listen(3000)

Fastify

src/server.ts

import Fastify from 'fastify'
import { createServe } from '@mastra/inngest'
import { serve as fastifyAdapter } from 'inngest/fastify'
import { mastra, inngest } from './mastra'

const fastify = Fastify()

// JSON parsing is handled by Fastify's default content-type parser
const handler = createServe(fastifyAdapter)({ mastra, inngest })

fastify.route({
  method: ['GET', 'POST', 'PUT'],
  url: '/inngest/api',
  handler,
})

fastify.listen({ port: 3000 })

Koa

src/server.ts

import Koa from 'koa'
import Router from '@koa/router'
import bodyParser from 'koa-bodyparser'
import { createServe } from '@mastra/inngest'
import { serve as koaAdapter } from 'inngest/koa'
import { mastra, inngest } from './mastra'

const app = new Koa()
const router = new Router()

// Body parsing middleware required for Inngest
app.use(bodyParser())

const handler = createServe(koaAdapter)({ mastra, inngest })
router.all('/inngest/api', handler)

app.use(router.routes())
app.use(router.allowedMethods())

app.listen(3000)

Next.js

app/inngest/api/route.ts

import { createServe } from '@mastra/inngest'
import { serve as nextAdapter } from 'inngest/next'
import { mastra, inngest } from '@/mastra'

const handler = createServe(nextAdapter)({ mastra, inngest })

export { handler as GET, handler as POST, handler as PUT }

可用适配器

createServe 函数可与任意 Inngest 适配器配合使用。包括 AWS Lambda、Cloudflare Workers 等在内的完整适配器列表，请参阅 Inngest serve 文档。

作为 Connect Worker 运行

serve() 会暴露供 Inngest 调用的 HTTP 端点。另一种方式是使用 connect()，从 Worker 向 Inngest 建立长时出站连接，使 Worker 不需要公开可访问的端点。这适用于在 Kubernetes、Docker、ECS、Fly.io 或 Render 等 Runtime 上运行的长时 Worker 进程。

备注

Inngest Connect 处于公开 beta 阶段。Connect Worker 不支持 Vercel 和 AWS Lambda 等 Serverless Runtime。

何时使用 connect() 而不是 serve()

• Worker 进程运行在只允许出站连接的专用网络中。
• 你不希望繁重的 Workflow 执行与面向用户的 HTTP 流量由同一个进程提供服务。
• 你希望独立于 Mastra HTTP Server 扩缩 Worker，并为每个 Worker 设置并发限制。

serve() 和 connect() 使用相同的 Mastra Workflow 定义。标准 Workflow、嵌套 Workflow、cron Workflow 和额外 Inngest 函数的收集行为相同。

要求

• inngest@^4
• Node.js 22.13.0 或更高版本
• 一个用于 Worker 的长时运行进程

设置

将 Mastra Server 和 Connect Worker 作为两个进程运行。在此设置中，Mastra Server 无需暴露 /inngest/api：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { incrementWorkflow } from './workflows'
import { PinoLogger } from '@mastra/loggers'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})

src/worker.ts

import { connect } from '@mastra/inngest/connect'
import { mastra } from './mastra'
import { inngest } from './mastra/inngest'

await connect({
  mastra,
  inngest,
  instanceId: process.env.INNGEST_CONNECT_INSTANCE_ID,
  maxWorkerConcurrency: Number(process.env.INNGEST_CONNECT_MAX_WORKER_CONCURRENCY ?? 10),
})

本地开发期间，使用 INNGEST_DEV=1 node --import tsx src/worker.ts 启动 Worker。Inngest Dev Server 会通过出站连接发现 Worker，因此不需要 -u flag。

在生产环境中，请设置 INNGEST_EVENT_KEY 和 INNGEST_SIGNING_KEY。将 Inngest Client 上的 appVersion 设置为部署标识符，例如 commit SHA 或镜像 tag，以便 Inngest 管理滚动部署。

将现有生产应用从 serve() 迁移到 connect() 时，请先使用单独的 Inngest 应用测试 Worker，再迁移流量。

选项

connect() 接受与 Inngest connect 相同的选项，以及 Mastra 专用字段。最常用的选项包括：

• mastra：要暴露其 Workflow 的 Mastra 实例。
• inngest：Inngest Client。使用与传给 serve() 相同的 Client。
• functions：可选的额外 Inngest 函数数组，与 Mastra Workflow 一同注册。
• instanceId：Worker 的稳定标识符，显示在 Inngest 仪表盘中。默认值为机器 hostname。
• maxWorkerConcurrency：Worker 同时运行的最大步骤数。默认不受限制。
• registerOptions：在应用注册期间转发给 Inngest（例如 signingKey）。如果某字段同时在此处和顶层设置，则以 registerOptions 为准。这与 serve() 的行为一致。

connect() 返回 Inngest 的 WorkerConnection。Inngest SDK 默认处理 SIGINT 和 SIGTERM。只有 Worker 需要自定义关闭控制时，才应存储返回的连接并调用 .close()。

如果 Mastra 实例没有 InngestWorkflow，且未提供额外 functions，connect() 会记录警告，因为否则 Worker 将保持连接，却没有任何内容可执行。请在 Mastra 上注册至少一个 Workflow，或传入 functions: [...]。

使用自定义 apiPrefix

如果需要保留 /api/inngest（例如在不更改仪表盘 URL 的情况下匹配 Inngest 自动发现约定），请设置 server.apiPrefix 以重新定位 Mastra 内置路由：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { serve } from '@mastra/inngest'
import { inngest } from './inngest'

export const mastra = new Mastra({
  server: {
    apiPrefix: '/_mastra',
    apiRoutes: [
      {
        path: '/api/inngest',
        method: 'ALL',
        createHandler: async ({ mastra }) => serve({ mastra, inngest }),
      },
    ],
  },
})

Mastra 内置路由现在解析到 /_mastra/agents、/_mastra/workflows 等路径下，从而释放 /api/inngest 路径供自定义路由使用。

注意

默认身份验证配置会保护 /api/*，并将 /api、/api/auth/* 视为公共路径。更改 apiPrefix 后，这些默认值不再匹配，内置路由会落在受保护模式之外。请更新 server.auth.protected 和 server.auth.public 以引用新前缀，并更新所有访问 /api/* 的 Client 代码，包括 MastraClient 的 apiPrefix。

流控制

Inngest Workflow 支持并发限制、速率限制、节流、防抖和优先级队列等流控制功能。这些选项在 createWorkflow() 调用中配置，有助于大规模管理 Workflow 执行。

并发

控制可以同时运行的 Workflow 实例数量：

const workflow = createWorkflow({
  id: 'user-processing-workflow',
  inputSchema: z.object({ userId: z.string() }),
  outputSchema: z.object({ result: z.string() }),
  steps: [processUserStep],
  // Limit to 10 concurrent executions, scoped by user ID
  concurrency: {
    limit: 10,
    key: 'event.data.userId',
  },
})

速率限制

限制一段时间内的 Workflow 执行次数：

const workflow = createWorkflow({
  id: 'api-sync-workflow',
  inputSchema: z.object({ endpoint: z.string() }),
  outputSchema: z.object({ status: z.string() }),
  steps: [apiSyncStep],
  // Maximum 1000 executions per hour
  rateLimit: {
    period: '1h',
    limit: 1000,
  },
})

节流

确保两次 Workflow 执行之间的最短间隔：

const workflow = createWorkflow({
  id: 'email-notification-workflow',
  inputSchema: z.object({ organizationId: z.string(), message: z.string() }),
  outputSchema: z.object({ sent: z.boolean() }),
  steps: [sendEmailStep],
  // Only one execution per 10 seconds per organization
  throttle: {
    period: '10s',
    limit: 1,
    key: 'event.data.organizationId',
  },
})

防抖

延迟执行，直到一个时间窗口内没有新事件到达：

const workflow = createWorkflow({
  id: 'search-index-workflow',
  inputSchema: z.object({ documentId: z.string() }),
  outputSchema: z.object({ indexed: z.boolean() }),
  steps: [indexDocumentStep],
  // Wait 5 seconds of no updates before indexing
  debounce: {
    period: '5s',
    key: 'event.data.documentId',
  },
})

优先级

设置 Workflow 执行优先级：

const workflow = createWorkflow({
  id: 'order-processing-workflow',
  inputSchema: z.object({
    orderId: z.string(),
    priority: z.number().optional(),
  }),
  outputSchema: z.object({ processed: z.boolean() }),
  steps: [processOrderStep],
  // Higher priority orders execute first
  priority: {
    run: 'event.data.priority ?? 50',
  },
})

组合流控制选项

可以在单个 Workflow 中组合多个流控制选项：

const workflow = createWorkflow({
  id: 'comprehensive-workflow',
  inputSchema: z.object({
    userId: z.string(),
    organizationId: z.string(),
    priority: z.number().optional(),
  }),
  outputSchema: z.object({ result: z.string() }),
  steps: [comprehensiveStep],
  concurrency: {
    limit: 5,
    key: 'event.data.userId',
  },
  rateLimit: {
    period: '1m',
    limit: 100,
  },
  throttle: {
    period: '10s',
    limit: 1,
    key: 'event.data.organizationId',
  },
  priority: {
    run: 'event.data.priority ?? 0',
  },
})

所有流控制选项均为可选。如果未指定，Workflow 会按 Inngest 默认行为运行。有关更多信息，请参阅 Inngest 流控制文档。

Cron 调度

使用 cron 表达式按计划触发 Inngest Workflow。常见用途包括每日报告、每小时数据同步和维护任务。

基础 cron 调度

添加 cron 属性，配置 Workflow 按计划运行：

const workflow = createWorkflow({
  id: 'daily-report-workflow',
  inputSchema: z.object({ reportType: z.string() }),
  outputSchema: z.object({ generated: z.boolean() }),
  steps: [generateReportStep],
  // Run daily at midnight
  cron: '0 0 * * *',
})

Cron 计划格式

cron 属性接受标准 cron 表达式，格式为：minute hour day month dayOfWeek

• minute：0-59
• hour：0-23
• day：1-31
• month：1-12 或 JAN-DEC
• dayOfWeek：0-6（星期日 = 0）或 SUN-SAT

常见 cron 模式：

// Every 15 minutes
cron: '*/15 * * * *'

// Every hour at minute 0
cron: '0 * * * *'

// Every 6 hours
cron: '0 */6 * * *'

// Daily at midnight
cron: '0 0 * * *'

// Daily at 9 AM
cron: '0 9 * * *'

// Every weekday at 9 AM
cron: '0 9 * * 1-5'

// First day of every month at midnight
cron: '0 0 1 * *'

// Every Monday at 8 AM
cron: '0 8 * * 1'

为计划运行提供输入数据

可以提供每次计划执行时使用的静态输入数据：

const workflow = createWorkflow({
  id: 'scheduled-data-sync',
  inputSchema: z.object({
    source: z.string(),
    destination: z.string(),
  }),
  outputSchema: z.object({ synced: z.boolean() }),
  steps: [syncDataStep],
  cron: '0 */6 * * *', // Every 6 hours
  // Input data provided to each scheduled run
  inputData: {
    source: 'production-db',
    destination: 'analytics-warehouse',
  },
})

为计划运行提供初始状态

还可以为计划 Workflow 运行设置初始状态：

const workflow = createWorkflow({
  id: 'scheduled-aggregation',
  inputSchema: z.object({ date: z.string() }),
  outputSchema: z.object({ aggregated: z.boolean() }),
  stateSchema: z.object({
    processedCount: z.number(),
    lastProcessedDate: z.string(),
  }),
  steps: [aggregateDataStep],
  cron: '0 0 * * *', // Daily at midnight
  inputData: {
    date: new Date().toISOString().split('T')[0], // Today's date
  },
  initialState: {
    processedCount: 0,
    lastProcessedDate: '',
  },
})

将 cron 与流控制结合使用

Cron 调度可以与流控制选项结合使用：

const workflow = createWorkflow({
  id: 'scheduled-api-sync',
  inputSchema: z.object({ endpoint: z.string() }),
  outputSchema: z.object({ synced: z.boolean() }),
  steps: [syncApiStep],
  cron: '*/30 * * * *', // Every 30 minutes
  inputData: {
    endpoint: 'https://api.example.com/data',
  },
  // Limit concurrent executions even for scheduled runs
  concurrency: {
    limit: 5,
  },
  // Rate limit scheduled executions
  rateLimit: {
    period: '1h',
    limit: 100,
  },
})

Cron 函数的工作原理

使用 cron 属性配置 Workflow 时：

1. 系统会自动创建 ID 为 workflow.${workflowId}.cron 的独立 Inngest 函数
2. 该函数会向 Inngest 注册，并按指定计划触发
3. 每次计划执行都会使用提供的 inputData 和 initialState 创建新的 Workflow 运行
4. 调用 serve() 时，会同时提供 cron 函数和主 Workflow 函数

可以在 Inngest 仪表盘的 Functions 和 Runs 部分监控计划执行。cron 函数会作为独立函数显示在主 Workflow 函数旁。

有关 cron 调度的更多信息，请参阅 Inngest cron 文档。