createInngestAgent()

createInngestAgent() 使用 Inngest 驱动的持久执行封装现有 Agent。与 createDurableAgent() 一样，它通过 PubSub 以 streaming 方式传输事件并支持可恢复 stream，但会在 Inngest 的执行引擎而非进程内运行 agentic loop。当运行必须经受进程重启或需要在分布式环境中执行时，请使用它。

对于进程内持久执行，请使用 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 }),
      },
    ],
  },
})

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

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 事件，并在 SDK v4 中发布 realtime stream 事件。

id?:

string

= agent.id

覆盖 ID。

name?:

string

= agent.name

覆盖名称。

pubsub?:

PubSub

= InngestPubSub

用于以 streaming 方式传输事件的 PubSub 实例。默认的 InngestPubSub 使用可跨进程工作的 Inngest Realtime。

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。

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

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>

generate(messages, options?)

在 Inngest 的持久执行引擎上运行响应，并解析为单个 FullOutput。如果运行暂停，generate() 会解析为 finishReason: 'suspended'。runId 选项是可选的。如果省略，generate() 会创建运行 ID，并通过 result.runId 返回。使用 resumeGenerate() 继续运行。当调用方需要暂停回调时，请使用带有 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() 运行，并解析为单个 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?)

重新连接到现有运行，先重放缓存的事件，再传送实时事件。请在网络断开后使用此方法。传入 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() 会销毁运行的注册表条目和缓存事件。仅在不再需要该运行时调用。如果运行已暂停且你打算稍后恢复，请勿调用 cleanup()。

prepare(messages, options?)

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

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 的类型守卫。

import { isInngestAgent } from '@mastra/inngest'

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

返回：boolean

Stream 选项

stream() 接受 InngestAgentStreamOptions 对象。它支持与 DurableAgent.stream() 相同的 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。

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

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

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