Inngest

Inngest is a developer platform for building and running background workflows, without managing infrastructure.

For a complete example with advanced flow control features, see the Inngest workflow example.

How Inngest works with Mastra

Inngest and Mastra integrate by aligning their workflow models: Inngest organizes logic into functions composed of steps, and Mastra workflows defined using createWorkflow() and createStep() map directly onto this structure. Each Mastra workflow becomes an Inngest function with a unique identifier, and each step within the workflow maps to an Inngest step.

The serve() function bridges the two systems by registering Mastra workflows as Inngest functions and setting up the necessary event handlers for execution and monitoring.

When an event triggers a workflow, Inngest executes it step by step and memoizes each result. On retry or resume, Inngest skips completed steps based on those saved results. Mastra control flow primitives, such as loops, conditionals, and nested workflows, map to the same Inngest function and step model while preserving composition, branching, and suspension.

Real-time monitoring, suspend/resume, and step-level observability are enabled via Inngest's publish-subscribe system and dashboard. As each step executes, its state and output are tracked using Mastra storage and can be resumed as needed.

Setup

Install the required packages:

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

备注

Requires inngest@^4 and Inngest Dev Server v1.18.0 or later. Realtime is built into the SDK in v4, so @inngest/realtime and realtimeMiddleware are no longer used.

Building an Inngest workflow

This guide walks through creating a workflow with Inngest and Mastra, demonstrating a counter application that increments a value until it reaches 10.

Inngest initialization

Initialize the Inngest integration to obtain Mastra-compatible workflow helpers. The createWorkflow() and createStep() functions are used to create workflow and step objects that are compatible with Mastra and inngest.

In development:

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

In production:

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

Selecting dev or cloud mode with INNGEST_DEV

The Inngest SDK runs in one of two modes: dev or cloud. From v4 onward, the SDK defaults to cloud mode and requires INNGEST_EVENT_KEY and INNGEST_SIGNING_KEY. The examples above set the mode in code with isDev: true. As an alternative, set the INNGEST_DEV environment variable so the same client code works across environments:

• INNGEST_DEV=1: force dev mode. The SDK talks to a local Inngest Dev Server and disables signature verification.
• INNGEST_DEV=0: force cloud mode. The SDK talks to Inngest Cloud and requires event and signing keys.
• INNGEST_DEV=<url>: force dev mode and point the SDK at the Dev Server at <url>, for example http://localhost:8288.
• Unset: defaults to cloud mode.

When INNGEST_DEV is set, you can drop isDev and baseUrl from the client:

src/mastra/inngest/index.ts

import { Inngest } from 'inngest'

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

注意

Don't set INNGEST_DEV in production. It disables signature verification, which is required to securely receive events from Inngest Cloud.

The isDev option on new Inngest() overrides INNGEST_DEV when both are set.

Creating steps

Define the individual steps that will compose your 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 }
  },
})

Creating the workflow

Compose the steps into a workflow using the dountil loop pattern. The createWorkflow() function creates a function on Inngest server that's invocable.

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 }

Configuring the Mastra instance

Register the workflow with Mastra and configure the Inngest API endpoint:

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

备注

The path is /inngest/api, not /api/inngest. Mastra reserves the /api prefix for built-in routes (agents, workflows, memory). Custom apiRoutes paths that start with the server's apiPrefix (default /api) throw at startup. See #15743 for context, or skip to Using a custom apiPrefix if you need to keep /api/inngest.

Running workflows

Running locally

1. Run npx mastra dev to start the Mastra server locally on port 4111

2. Start the Inngest Dev Server. In a new terminal, run:

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

   备注

   The URL after -u tells the Inngest dev server where to find your Mastra /inngest/api endpoint

3. Open the Inngest Dashboard at http://localhost:8288 and go to the Apps section in the sidebar to verify your Mastra workflow is registered

4. In Functions, open your workflow. Select Invoke and provide the following input:

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

5. Monitor the workflow execution in the Runs tab to see step-by-step execution progress

Running in production

Before you begin, make sure you have:

• Vercel account and Vercel CLI installed (npm i -g vercel)
• Inngest account
• Vercel token

1. Set your Vercel token in your environment:

   .env

   export VERCEL_TOKEN=your_vercel_token

2. Add VercelDeployer to Mastra instance

   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. Build the mastra instance

   npx mastra build

4. Deploy to Vercel

   cd .mastra/output
   vercel login
   vercel --prod

5. Sync with the Inngest dashboard by selecting Sync new app with Vercel and following the instructions

   注意

   Inngest's auto-discover convention assumes /api/inngest. Because this guide uses /inngest/api, set the URL field on the Inngest app to your deployed origin plus /inngest/api (for example https://your-app.vercel.app/inngest/api). If you leave it on the default, the Inngest dashboard won't find your app's functions.

6. In Functions, open workflow.increment-workflow. Select All actions > Invoke and provide the following input:

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

7. Monitor execution in the Runs tab to see step-by-step progress

Adding custom Inngest functions

You can serve additional Inngest functions alongside your Mastra workflows by using the optional functions parameter in serve().

Creating custom functions

First, create your custom Inngest functions:

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

Serving custom functions with workflows

Update your Mastra configuration to import and include the custom functions. The highlighted lines show the additions:

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

Function registration

When you include custom functions:

1. Mastra workflows are automatically converted to Inngest functions with IDs like workflow.${workflowId}
2. Custom functions retain their specified IDs (e.g., send-welcome-email, process-webhook)
3. All functions are served together on the same /inngest/api endpoint

This allows you to combine Mastra's workflow orchestration with your existing Inngest functions.

Using with other frameworks

The default serve function uses Hono internally. If you're using a different web framework like Express, Fastify, or Koa, use the createServe factory function with the appropriate Inngest adapter.

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 }

Available adapters

The createServe function works with any Inngest adapter. See the Inngest serve documentation for a complete list of available adapters including AWS Lambda, Cloudflare Workers, and more.

Running as a Connect worker

serve() exposes an HTTP endpoint that Inngest calls into. As an alternative, connect() opens a long-lived outbound connection from your worker to Inngest, so the worker doesn't need a publicly reachable endpoint. Use this for long-running worker processes on runtimes such as Kubernetes, Docker, ECS, Fly.io, or Render.

备注

Inngest Connect is in public beta. Serverless runtimes such as Vercel and AWS Lambda aren't supported for Connect workers.

When to use connect() instead of serve()

• The worker process runs in a private network that only allows outbound connections.
• You want to keep heavy workflow execution out of the same process that serves user-facing HTTP traffic.
• You want to scale workers up and down independently of the Mastra HTTP server, with per-worker concurrency limits.

Use the same Mastra workflow definitions with either serve() or connect(). The collection behavior is the same for standard workflows, nested workflows, cron workflows, and additional Inngest functions.

Requirements

• inngest@^4
• Node.js 22.13.0 or later
• A long-running process for the worker

Setup

Run the Mastra server and the Connect worker as two processes. The Mastra server in this setup doesn't need to expose /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),
})

Start the worker with INNGEST_DEV=1 node --import tsx src/worker.ts during local development. The Inngest Dev Server discovers the worker through the outbound connection, so no -u flag is required.

In production, set INNGEST_EVENT_KEY and INNGEST_SIGNING_KEY. Set appVersion on the Inngest client to a deploy identifier, such as a commit SHA or image tag, so Inngest can manage rolling deploys.

When migrating an existing production app from serve() to connect(), test the worker with a separate Inngest app before moving traffic.

Options

connect() accepts the same options as Inngest's connect plus the Mastra-specific fields. The most common ones:

• mastra: The Mastra instance whose workflows should be exposed.
• inngest: The Inngest client. Use the same client you would pass to serve().
• functions: Optional array of additional Inngest functions to register alongside Mastra workflows.
• instanceId: Stable identifier for the worker, shown in the Inngest dashboard. Defaults to the machine hostname.
• maxWorkerConcurrency: Maximum number of steps the worker runs at a time. Defaults to unlimited.
• registerOptions: Forwarded to Inngest during app registration (for example signingKey). When a field is set both here and at the top level, registerOptions wins. This matches the behavior of serve().

connect() returns Inngest's WorkerConnection. The Inngest SDK handles SIGINT and SIGTERM by default. Store the returned connection and call .close() only when your worker needs custom shutdown control.

If the Mastra instance has no InngestWorkflow and no additional functions are provided, connect() logs a warning because the worker would otherwise stay connected with nothing to execute. Register at least one workflow on Mastra or pass functions: [...].

Using a custom apiPrefix

If you need to keep /api/inngest (for example to match Inngest's auto-discover convention without changing the dashboard URL), set server.apiPrefix to relocate Mastra's built-in routes:

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's built-in routes now resolve under /_mastra/agents, /_mastra/workflows, and so on, freeing the /api/inngest path for your custom route.

注意

The default auth configuration protects /api/* and treats /api, /api/auth/* as public. When you change apiPrefix, those defaults no longer match and built-in routes fall outside the protected pattern. Update server.auth.protected and server.auth.public to reference the new prefix, and update any client code (including MastraClient apiPrefix) that hits /api/*.

Flow control

Inngest workflows support flow control features including concurrency limits, rate limiting, throttling, debouncing, and priority queuing. These options are configured in the createWorkflow() call and help manage workflow execution at scale.

Concurrency

Control how many workflow instances can run simultaneously:

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

Rate limiting

Limit the number of workflow executions within a time period:

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

Throttling

Ensure minimum time between workflow executions:

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

Debouncing

Delay execution until no new events arrive within a time window:

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

Priority

Set execution priority for workflows:

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

Retries

Set retries to let a run survive a failed request to your app, such as a process restart, redeploy, or out-of-memory crash. Inngest calls the function again, skips the steps that already finished, and continues from the step that was interrupted:

const workflow = createWorkflow({
  id: 'long-running-workflow',
  inputSchema: z.object({ jobId: z.string() }),
  outputSchema: z.object({ done: z.boolean() }),
  steps: [longRunningStep],
  // Re-invoke up to 3 times after a failed request
  retries: 3,
})

retries defaults to 0. Errors thrown by your step code are retried per step with retryConfig or the step's retries option, not at the function level. createInngestAgent() accepts the same retries option and applies it to every Inngest function the durable agent creates. All durable agents share these functions, so the retries value of the first agent registered with Mastra applies to every durable agent. Set the same value on each agent to avoid confusion.

A nested workflow runs as its own Inngest function and uses its own retries setting. It doesn't inherit the parent's value, so set retries on each nested workflow that should recover from a failed request.

Combining flow control options

Multiple flow control options can be combined in a single 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',
  },
})

All flow control options are optional. If not specified, workflows run with Inngest's default behavior. For more information, see the Inngest flow control documentation.

Cron scheduling

Use cron expressions to trigger Inngest workflows on a schedule. Common uses include daily reports and hourly data syncs, as well as maintenance tasks.

Basic cron scheduling

Configure a workflow to run on a schedule by adding a cron property:

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 schedule format

The cron property accepts standard cron expressions in the format: minute hour day month dayOfWeek

• minute: 0-59
• hour: 0-23
• day: 1-31
• month: 1-12 or JAN-DEC
• dayOfWeek: 0-6 (Sunday = 0) or SUN-SAT

Common cron patterns:

// 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'

Providing input data for scheduled runs

You can provide static input data that will be used for each scheduled execution:

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

Providing initial state for scheduled runs

You can also set an initial state for scheduled workflow runs:

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: '',
  },
})

Combining cron with flow control

Cron scheduling can be combined with flow control options:

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

How cron functions work

When you configure a workflow with a cron property:

1. A separate Inngest function is automatically created with the ID workflow.${workflowId}.cron
2. This function is registered with Inngest and will trigger at the specified schedule
3. Each scheduled execution creates a new workflow run with the provided inputData and initialState
4. The cron function and the main workflow function are both served together when you call serve()

You can monitor scheduled executions in the Inngest dashboard under Functions and Runs sections. The cron function will appear as a separate function alongside your main workflow function.

For more information on cron scheduling, see the Inngest cron documentation.