> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# Inngest Workflow

[Inngest](https://www.inngest.com/docs) 是一个用于构建和运行后台 Workflow 的开发者平台，无需管理基础设施。

包含高级流控制功能的完整示例，请参阅 [Inngest Workflow 示例](https://github.com/mastra-ai/mastra/tree/main/examples/inngest)。

## 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**:

```bash
npm install @mastra/inngest@latest inngest
```

**pnpm**:

```bash
pnpm add @mastra/inngest@latest inngest
```

**Yarn**:

```bash
yarn add @mastra/inngest@latest inngest
```

**Bun**:

```bash
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 和步骤对象。

在开发环境中：

```ts
import { Inngest } from 'inngest'

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

在生产环境中：

```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` 环境变量](https://www.inngest.com/docs/sdk/environment-variables#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`：

```ts
import { Inngest } from 'inngest'

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

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

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

### 创建步骤

定义组成 Workflow 的各个步骤：

```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 上创建一个可调用函数。

```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 端点：

```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](https://github.com/mastra-ai/mastra/pull/15743)；如果需要保留 `/api/inngest`，请直接前往[使用自定义 `apiPrefix`](#using-a-custom-apiprefix)。

## 运行 Workflow

### 在本地运行

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

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

   ```bash
   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**，然后提供以下输入：

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

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

### 在生产环境中运行

开始之前，请确保具备：

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

1. 在环境中设置 Vercel token：

   ```bash
   export VERCEL_TOKEN=your_vercel_token
   ```

2. 将 `VercelDeployer` 添加到 Mastra 实例

   ```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 实例

   ```bash
   npx mastra build
   ```

4. 部署到 Vercel

   ```bash
   cd .mastra/output
   vercel login
   vercel --prod
   ```

5. 选择 **Sync new app with Vercel** 并按照说明操作，与 [Inngest 仪表盘](https://app.inngest.com/env/production/apps)同步

   > **注意:** 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**，然后提供以下输入：

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

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

## 添加自定义 Inngest 函数

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

### 创建自定义函数

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

```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 配置，导入并包含自定义函数。高亮行显示了新增内容：

```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

```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

```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

```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

```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 文档](https://www.inngest.com/docs/reference/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`：

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

```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`](https://www.inngest.com/docs/setup/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 内置路由：

```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`](https://mastra.zisheng.pro/docs/server/mastra-client) 的 `apiPrefix`。

## 流控制

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

### 并发

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

```ts
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 执行次数：

```ts
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 执行之间的最短间隔：

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

### 防抖

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

```ts
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 执行优先级：

```ts
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 中组合多个流控制选项：

```ts
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 流控制文档](https://www.inngest.com/docs/guides/flow-control)。

## Cron 调度

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

### 基础 cron 调度

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

```ts
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 模式：

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

### 为计划运行提供输入数据

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

```ts
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 运行设置初始状态：

```ts
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 调度可以与流控制选项结合使用：

```ts
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 文档](https://www.inngest.com/docs/guides/scheduled-functions)。