createInngestAgent()

createInngestAgent() 會以 Inngest 支援的持久執行包裝現有 Agent。和 createDurableAgent() 一樣，它會透過 PubSub 串流 event，並支援可恢復 stream；但會在 Inngest 執行引擎上執行 Agent 迴圈，而不是在處理程序內執行。若 run 必須在處理程序重新啟動後繼續，或在分散式環境中執行，請使用此函式。

若要進行處理程序內持久執行，請使用 createDurableAgent()。若要在內建 Workflow 引擎上進行傳送後不等待結果的執行，請使用 createEventedAgent()。

使用範例

設定 Inngest client、包裝 Agent、將其註冊至 Mastra，並公開 Inngest serve endpoint：

src/mastra/index.ts

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

串流回應並讀取結果：

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

const text = await output.text

cleanup()

createInngestAgent(options)

以 Inngest 支援的持久執行與可恢復 stream 包裝 Agent。

import { createInngestAgent } from '@mastra/inngest'

const durableAgent = createInngestAgent({ agent, inngest })

回傳： InngestAgent

參數

agent:

Agent

要以 Inngest 持久執行包裝的 Agent。InngestAgent 未實作的方法（例如 listTools() 與 getMemory()）會透過 Proxy 委派給此 Agent。

inngest:

Inngest

Inngest client 執行個體。用於傳送 Workflow event；在 SDK v4 中也用於發布即時 stream event。

id?:

string

= agent.id

ID 覆寫值。

name?:

string

= agent.name

名稱覆寫值。

pubsub?:

PubSub

= InngestPubSub

用於串流 event 的 PubSub 執行個體。預設的 InngestPubSub 使用 Inngest Realtime，可跨處理程序運作。

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。

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>

resume(runId, resumeData, options?)

恢復已暫停的 Inngest run，例如 Tool 核准後。此方法會從儲存空間載入 Workflow 快照、找出暫停的步驟，並向 Inngest 傳送恢復 event。

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>

generate(messages, options?)

在 Inngest 的持久執行引擎上執行回應，並解析為單一 FullOutput。若 run 暫停，generate() 會解析為 finishReason: 'suspended'。runId 選項為選用；若省略，generate() 會建立 run ID，並在 result.runId 中回傳。使用 resumeGenerate() 繼續 run。若呼叫端需要暫停 callback，請使用搭配 onSuspended 的 stream()。

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。

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 可從已知位置開始重播。

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。

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

回傳：

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

檢查

isInngestAgent(obj)

檢查物件是否為 InngestAgent 的型別 guard。

import { isInngestAgent } from '@mastra/inngest'

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

回傳： boolean

stream 選項

stream() 接受 InngestAgentStreamOptions 物件。它支援與 DurableAgent.stream() 相同的 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。

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 函式。

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。

Express

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

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

Next.js

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 參考文件
• Agent 類別
• Inngest 部署指南
• PubSub