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

# `createInngestAgent()`

`createInngestAgent()` 使用 [Inngest](https://www.inngest.com/docs) 驱动的持久执行封装现有 [`Agent`](https://mastra.zisheng.pro/reference/agents/agent)。与 [`createDurableAgent()`](https://mastra.zisheng.pro/reference/agents/durable-agent) 一样，它通过 [PubSub](https://mastra.zisheng.pro/docs/server/pubsub) 以 streaming 方式传输事件并支持可恢复 stream，但会在 Inngest 的执行引擎而非进程内运行 agentic loop。当运行必须经受进程重启或需要在分布式环境中执行时，请使用它。

对于进程内持久执行，请使用 [`createDurableAgent()`](https://mastra.zisheng.pro/reference/agents/durable-agent)。对于在内置 Workflow 引擎上触发后即无需等待的执行，请使用 [`createEventedAgent()`](https://mastra.zisheng.pro/reference/agents/durable-agent)。

## 使用示例

设置 Inngest client，封装 Agent，将其注册到 Mastra，并公开 Inngest serve endpoint：

```typescript
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'
import { createInngestAgent, serve as inngestServe } from '@mastra/inngest'
import { Inngest } from 'inngest'

const inngest = new Inngest({ id: 'my-app' })

const agent = new Agent({
  id: 'my-agent',
  name: 'My Agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
})

const durableAgent = createInngestAgent({ agent, inngest })

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
  server: {
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => inngestServe({ mastra, inngest }),
      },
    ],
  },
})
```

以 streaming 方式传输响应并读取结果：

```typescript
const { output, runId, cleanup } = await durableAgent.stream('Hello!')

const text = await output.text

cleanup()
```

## `createInngestAgent(options)`

使用 Inngest 驱动的持久执行和可恢复 stream 封装 `Agent`。

```typescript
import { createInngestAgent } from '@mastra/inngest'

const durableAgent = createInngestAgent({ agent, inngest })
```

返回：[`InngestAgent`](#inngestagent-interface)

### 参数

**agent** (`Agent`): 要使用 Inngest 持久执行封装的 Agent。InngestAgent 未实现的方法（例如 listTools() 和 getMemory()）会通过 Proxy 委托给此 Agent。

**inngest** (`Inngest`): Inngest client 实例。用于发送 Workflow 事件，并在 SDK v4 中发布 realtime stream 事件。

**id** (`string`): 覆盖 ID。 (Default: `agent.id`)

**name** (`string`): 覆盖名称。 (Default: `agent.name`)

**pubsub** (`PubSub`): 用于以 streaming 方式传输事件的 PubSub 实例。默认的 InngestPubSub 使用可跨进程工作的 Inngest Realtime。 (Default: `InngestPubSub`)

**cache** (`MastraServerCache`): 用于存储 stream 事件的 cache，从而支持可恢复 stream。提供后，PubSub 会自动由 CachingPubSub 封装。如果省略，Agent 会从 Mastra 实例继承 cache。

**mastra** (`Mastra`): 用于可观测性的 Mastra 实例。Agent 注册到 Mastra 时自动设置。

## `InngestAgent` interface

`createInngestAgent()` 返回的对象。它提供以下持久执行方法。任何未明确定义的属性或方法（例如 `listTools()` 和 `getMemory()`）都会通过 Proxy 转发给底层 Agent。

### 属性

**id** (`string`): Agent ID。

**name** (`string`): Agent 名称。

**agent** (`Agent`): 底层 Mastra Agent。

**inngest** (`Inngest`): Inngest client。

**cache** (`MastraServerCache | undefined`): 启用可恢复 stream 时解析出的 cache 实例。

**pubsub** (`PubSub`): 用于以 streaming 方式传输事件的 PubSub 实例。

## 方法

### 执行

#### `stream(messages, options?)`

使用 Inngest 的持久执行引擎以 streaming 方式传输响应。建立 PubSub 订阅后，通过 Inngest 事件触发 Workflow。

```typescript
const { output, runId, cleanup } = await durableAgent.stream('Hello!', {
  onChunk: chunk => console.log(chunk),
  onFinish: result => console.log('done', result),
})

const text = await output.text
cleanup()
```

返回：[`Promise<InngestAgentStreamResult>`](#inngestagentstreamresult)

#### `resume(runId, resumeData, options?)`

恢复已暂停的 Inngest 运行，例如在 Tool 审批后恢复。该方法从存储加载 Workflow 快照，找到已暂停的步骤，并向 Inngest 发送恢复事件。

```typescript
const { output, cleanup } = await durableAgent.resume(
  runId,
  {
    approved: true,
  },
  { threadId: 'thread-1', resourceId: 'user-1' },
)

await output.text
cleanup()
```

除了生命周期回调，第三个参数还接受 `threadId` 和 `resourceId`：

**threadId** (`string`): 与恢复运行关联的 thread ID。

**resourceId** (`string`): 与恢复运行关联的 resource ID。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): 每个 streaming chunk 到达时调用。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): agentic loop 中的步骤完成时调用。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): 运行完成时调用。

**onError** (`(error: Error) => void | Promise<void>`): 运行出错时调用。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): 运行暂停时调用。

返回：[`Promise<InngestAgentStreamResult>`](#inngestagentstreamresult)

#### `generate(messages, options?)`

在 Inngest 的持久执行引擎上运行响应，并解析为单个 `FullOutput`。如果运行暂停，`generate()` 会解析为 `finishReason: 'suspended'`。`runId` 选项是可选的。如果省略，`generate()` 会创建运行 ID，并通过 `result.runId` 返回。使用 [`resumeGenerate()`](#resumegeneraterunid-resumedata-options) 继续运行。当调用方需要暂停回调时，请使用带有 `onSuspended` 的 [`stream()`](#streammessages-options)。

```typescript
const result = await durableAgent.generate('Delete the old records', {
  requireToolApproval: true,
})

result.runId // Generated automatically
result.finishReason // 'suspended' when approval is required
```

返回：`Promise<FullOutput<TOutput>>`

#### `resumeGenerate(runId, resumeData, options?)`

恢复已暂停的 `generate()` 运行，并解析为单个 `FullOutput`。

```typescript
if (!result.runId) {
  throw new Error('Run ID is missing')
}

const resumedResult = await durableAgent.resumeGenerate(result.runId, { approved: true })
```

返回：`Promise<FullOutput<TOutput>>`

#### `observe(runId, options?)`

重新连接到现有运行，先重放缓存的事件，再传送实时事件。请在网络断开后使用此方法。传入 `offset` 可从已知位置开始重放。

```typescript
const { output, cleanup } = await durableAgent.observe(runId, {
  offset: 0,
  onChunk: chunk => console.log(chunk),
})

await output.text
```

`observe()` 的结果不包括 `threadId` 或 `resourceId`。

返回：`Promise<Omit<InngestAgentStreamResult, 'threadId' | 'resourceId'>>`

> **注意:** `observe()` 返回的 `cleanup()` 会销毁运行的注册表条目和缓存事件。仅在不再需要该运行时调用。如果运行已暂停且你打算稍后恢复，请勿调用 `cleanup()`。

#### `prepare(messages, options?)`

为持久执行准备运行，但不触发它。返回序列化后的 Workflow 输入，可用于手动触发 Inngest Workflow 事件。

```typescript
const { runId, messageId, workflowInput, threadId, resourceId } = await durableAgent.prepare(
  'Summarize the document',
  {
    memory: { threadId: 'thread-1', resourceId: 'user-1' },
  },
)
```

返回：

```typescript
interface PrepareResult {
  runId: string
  messageId: string
  workflowInput: any
  threadId?: string
  resourceId?: string
}
```

### 自省

#### `isInngestAgent(obj)`

检查对象是否为 `InngestAgent` 的类型守卫。

```typescript
import { isInngestAgent } from '@mastra/inngest'

if (isInngestAgent(agent)) {
  // agent is InngestAgent
}
```

返回：`boolean`

## Stream 选项

`stream()` 接受 `InngestAgentStreamOptions` 对象。它支持与 [`DurableAgent.stream()`](https://mastra.zisheng.pro/reference/agents/durable-agent) 相同的 Agent 执行选项，另加生命周期回调。

**runId** (`string`): 此次运行的唯一标识符。稍后可将其用于 resume() 或 observe()。

**instructions** (`AgentExecutionOptions['instructions']`): 覆盖 Agent 此次运行的默认指令。

**context** (`ModelMessage[]`): 提供给 Agent 的其他上下文消息。

**memory** (`object`): 用于对话持久化和检索的 memory 配置。

**requestContext** (`RequestContext`): 携带此次运行动态配置和状态的 request context。

**maxSteps** (`number`): 运行的最大步骤数。

**toolsets** (`object`): 此次运行可用的其他 Tool 集。

**clientTools** (`object`): 执行期间可用的客户端 Tool。

**toolChoice** (`'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }`): Tool 选择策略。

**modelSettings** (`object`): temperature 等 Model 特定设置。

**requireToolApproval** (`boolean`): 要求审批所有 Tool 调用，使运行暂停直至恢复。

**autoResumeSuspendedTools** (`boolean`): 自动恢复已暂停的 Tool，而不是等待外部 resume() 调用。

**toolCallConcurrency** (`number`): 并发执行的最大 Tool 调用数。

**includeRawChunks** (`boolean`): 在 stream 输出中包含原始 Provider chunk。

**maxProcessorRetries** (`number`): 每次生成中 processor 的最大重试次数。

**untilIdle** (`boolean | { maxIdleMs?: number }`): 设置后，在后台任务延续期间保持 stream 开启，直到 Agent 空闲。传入 true 使用默认的 5 分钟空闲超时，或传入 { maxIdleMs } 自定义。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): 每个 streaming chunk 到达时调用。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): agentic loop 中的步骤完成时调用。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): 运行完成时调用。

**onError** (`(error: Error) => void | Promise<void>`): 运行出错时调用。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): 运行暂停时调用，例如等待 Tool 审批时。

`observe()` 接受生命周期回调（`onChunk`、`onStepFinish`、`onFinish`、`onError`、`onSuspended`），并接受用于控制重放起始位置的 `offset`。

## `InngestAgentStreamResult`

`stream()` 和 `resume()` 返回的对象。`observe()` 方法返回相同的结构，但省略 `threadId` 和 `resourceId`。

```typescript
interface InngestAgentStreamResult<OUTPUT = undefined> {
  output: MastraModelOutput<OUTPUT>
  readonly fullStream: ReadableStream<any>
  runId: string
  threadId?: string
  resourceId?: string
  cleanup: () => void
}
```

**output** (`MastraModelOutput`): streaming 输出。等待 output.text 可获得完整文本，也可以消费 output.fullStream。

**fullStream** (`ReadableStream`): 完整事件 stream，委托给 output.fullStream。

**runId** (`string`): 唯一运行 ID。将其传给 resume() 或 observe() 以重新连接。

**threadId** (`string`): 使用 memory 时的 thread ID。

**resourceId** (`string`): 使用 memory 时的 resource ID。

**cleanup** (`() => void`): 取消 PubSub 订阅并清除运行的注册表条目。完成运行后调用。

## 提供 Inngest 函数

`@mastra/inngest` 包提供 `serve()` 和 `createServe()`，用于在 HTTP framework 中注册 Inngest Workflow 函数。

### `serve(options)`

使用 Hono（默认 framework）提供 Mastra Workflow。它从 Mastra 收集所有由 Inngest 支持的 Workflow，并将其注册为 Inngest 函数。

```typescript
import { serve } from '@mastra/inngest'

app.use('/inngest/api', async c => {
  return serve({ mastra, inngest })(c)
})
```

### `createServe(adapter)`

该工厂接受任意 Inngest serve adapter（`inngest/express`、`inngest/fastify`、`inngest/next` 等），并返回适用于该 framework 的 serve 函数。

```typescript
import { createServe } from '@mastra/inngest'
import { serve } from 'inngest/express'

const serveExpress = createServe(serve)
app.use('/inngest/api', serveExpress({ mastra, inngest }))
```

```typescript
import { createServe } from '@mastra/inngest'
import { serve } from 'inngest/next'

const serveNext = createServe(serve)
export const { GET, POST, PUT } = serveNext({ mastra, inngest })
```

### Serve 选项

**mastra** (`Mastra`): 包含已注册 Agent 和 Workflow 的 Mastra 实例。

**inngest** (`Inngest`): Inngest client 实例。

**functions** (`InngestFunction.Like[]`): 与 Mastra Workflow 一同提供的其他 Inngest 函数。

**registerOptions** (`RegisterOptions`): 传递给 Inngest 注册 handler 的选项。

## 相关内容

- [DurableAgent 参考](https://mastra.zisheng.pro/reference/agents/durable-agent)
- [Agent 类](https://mastra.zisheng.pro/reference/agents/agent)
- [Inngest 部署指南](https://mastra.zisheng.pro/guides/deployment/inngest)
- [PubSub](https://mastra.zisheng.pro/docs/server/pubsub)