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

# `createInngestAgent()`

`createInngestAgent()` 會以 [Inngest](https://www.inngest.com/docs) 支援的持久執行包裝現有 [`Agent`](https://mastra.zisheng.pro/zh-TW/reference/agents/agent)。和 [`createDurableAgent()`](https://mastra.zisheng.pro/zh-TW/reference/agents/durable-agent) 一樣，它會透過 [PubSub](https://mastra.zisheng.pro/zh-TW/docs/server/pubsub) 串流 event，並支援可恢復 stream；但會在 Inngest 執行引擎上執行 Agent 迴圈，而不是在處理程序內執行。若 run 必須在處理程序重新啟動後繼續，或在分散式環境中執行，請使用此函式。

若要進行處理程序內持久執行，請使用 [`createDurableAgent()`](https://mastra.zisheng.pro/zh-TW/reference/agents/durable-agent)。若要在內建 Workflow 引擎上進行傳送後不等待結果的執行，請使用 [`createEventedAgent()`](https://mastra.zisheng.pro/zh-TW/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 }),
      },
    ],
  },
})
```

串流回應並讀取結果：

```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 event；在 SDK v4 中也用於發布即時 stream event。

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

**name** (`string`): 名稱覆寫值。 (Default: `agent.name`)

**pubsub** (`PubSub`): 用於串流 event 的 PubSub 執行個體。預設的 InngestPubSub 使用 Inngest Realtime，可跨處理程序運作。 (Default: `InngestPubSub`)

**cache** (`MastraServerCache`): 儲存 stream event 的快取，可啟用可恢復 stream。提供此值時，系統會自動以 CachingPubSub 包裝 PubSub。若省略，Agent 會繼承 Mastra 執行個體的快取。

**mastra** (`Mastra`): 用於可觀測性的 Mastra 執行個體。將 Agent 註冊至 Mastra 時會自動設定。

## `InngestAgent` 介面

`createInngestAgent()` 回傳的物件。它提供下列持久執行方法。任何未明確定義的屬性或方法（例如 `listTools()` 與 `getMemory()`）都會透過 Proxy 轉送至底層 Agent。

### 屬性

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

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

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

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

**cache** (`MastraServerCache | undefined`): 若啟用可恢復 stream，則為已解析的快取執行個體。

**pubsub** (`PubSub`): 用於串流 event 的 PubSub 執行個體。

## 方法

### 執行

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

使用 Inngest 的持久執行引擎串流回應。建立 PubSub 訂閱後，會透過 Inngest event 觸發 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 run，例如 Tool 核准後。此方法會從儲存空間載入 Workflow 快照、找出暫停的步驟，並向 Inngest 傳送恢復 event。

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

await output.text
cleanup()
```

第三個引數除了生命週期 callback 外，也接受 `threadId` 與 `resourceId`：

**threadId** (`string`): 要與已恢復 run 關聯的 thread ID。

**resourceId** (`string`): 要與已恢復 run 關聯的 resource ID。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): 每個串流 chunk 都會呼叫。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): Agent 迴圈中的步驟完成時呼叫。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): run 完成時呼叫。

**onError** (`(error: Error) => void | Promise<void>`): run 發生錯誤時呼叫。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): run 暫停時呼叫。

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

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

在 Inngest 的持久執行引擎上執行回應，並解析為單一 `FullOutput`。若 run 暫停，`generate()` 會解析為 `finishReason: 'suspended'`。`runId` 選項為選用；若省略，`generate()` 會建立 run ID，並在 `result.runId` 中回傳。使用 [`resumeGenerate()`](#resumegeneraterunid-resumedata-options) 繼續 run。若呼叫端需要暫停 callback，請使用搭配 `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()` run，並解析為單一 `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?)`

重新連線至現有 run；先重播快取 event，再傳送即時 event。網路中斷後請使用此方法。傳入 `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()` 會刪除 run 的 registry 項目與快取 event。只有在 run 使用完畢後才能呼叫。若 run 已暫停且您打算稍後恢復，請勿呼叫 `cleanup()`。

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

準備 run 以供持久執行，但不會觸發。回傳序列化的 Workflow 輸入，可用來手動觸發 Inngest Workflow event。

```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` 的型別 guard。

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

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

回傳： `boolean`

## stream 選項

`stream()` 接受 `InngestAgentStreamOptions` 物件。它支援與 [`DurableAgent.stream()`](https://mastra.zisheng.pro/zh-TW/reference/agents/durable-agent) 相同的 Agent 執行選項，以及生命週期 callback。

**runId** (`string`): 此 run 的唯一識別碼。稍後可搭配 resume() 或 observe() 使用。

**instructions** (`AgentExecutionOptions['instructions']`): 覆寫此 run 的 Agent 預設指示。

**context** (`ModelMessage[]`): 要提供給 Agent 的其他 context 訊息。

**memory** (`object`): 用於保存及擷取對話的記憶體設定。

**requestContext** (`RequestContext`): 帶有此 run 動態設定與狀態的 request context。

**maxSteps** (`number`): 最多可執行的步驟數。

**toolsets** (`object`): 此 run 可用的其他 Tool set。

**clientTools** (`object`): 執行期間可用的 client 端 Tool。

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

**modelSettings** (`object`): 模型專屬設定，例如 temperature。

**requireToolApproval** (`boolean`): 要求核准所有 Tool 呼叫；run 會暫停直到恢復。

**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>`): 每個串流 chunk 都會呼叫。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): Agent 迴圈中的步驟完成時呼叫。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): run 完成時呼叫。

**onError** (`(error: Error) => void | Promise<void>`): run 發生錯誤時呼叫。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): run 暫停時呼叫，例如等待 Tool 核准。

`observe()` 接受生命週期 callback（`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`): 串流輸出。await output.text 可取得完整文字，或取用 output.fullStream。

**fullStream** (`ReadableStream`): 完整 event stream，會委派給 output.fullStream。

**runId** (`string`): 唯一的 run ID。將其傳給 resume() 或 observe() 可重新連線。

**threadId** (`string`): 使用記憶體時的 thread ID。

**resourceId** (`string`): 使用記憶體時的 resource ID。

**cleanup** (`() => void`): 取消 PubSub 訂閱，並清除 run 的 registry 項目。run 使用完畢後請呼叫此函式。

## 提供 Inngest 函式

`@mastra/inngest` 套件提供 `serve()` 與 `createServe()`，可將 Inngest Workflow 函式註冊至您的 HTTP framework。

### `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 函式的 factory。

```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/zh-TW/reference/agents/durable-agent)
- [Agent 類別](https://mastra.zisheng.pro/zh-TW/reference/agents/agent)
- [Inngest 部署指南](https://mastra.zisheng.pro/zh-TW/guides/deployment/inngest)
- [PubSub](https://mastra.zisheng.pro/zh-TW/docs/server/pubsub)