createInngestAgent()

createInngestAgent() は、既存の Agent を Inngest による耐久実行でラップします。createDurableAgent() と同様に、PubSub 経由でイベントをストリーミングし、再開可能な Stream をサポートします。ただし、Agentic Loop はプロセス内ではなく Inngest の実行 Engine 上で動作します。プロセスの再起動後も Run を継続する必要がある場合や、分散環境で実行する場合に使用します。

プロセス内で耐久実行するには、createDurableAgent() を使用します。組み込み Workflow Engine で Fire-and-forget 実行するには、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)

Agent を Inngest による耐久実行と再開可能な Stream でラップします。

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 ではリアルタイム Stream イベントの発行に使用します。

id?:

string

= agent.id

ID の上書き値。

name?:

string

= agent.name

名前の上書き値。

pubsub?:

PubSub

= InngestPubSub

イベントをストリーミングする PubSub インスタンス。デフォルトの InngestPubSub は Inngest Realtime を使用するため、プロセスをまたいで動作します。

cache?:

MastraServerCache

保存済み Stream イベントの Cache。Stream の再開を可能にします。指定すると、PubSub は自動的に CachingPubSub でラップされます。省略すると、Agent は Mastra インスタンスの Cache を継承します。

mastra?:

Mastra

Observability に使用する 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 が有効な場合に、解決された Cache インスタンス。

pubsub:

PubSub

イベントのストリーミングに使用する PubSub インスタンス。

メソッド

実行

stream(messages, options?)

Inngest の耐久実行 Engine を使用してレスポンスをストリーミングします。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?)

Tool の承認後などに、中断中の Inngest Run を再開します。Storage から Workflow Snapshot を読み込み、中断中の Step を見つけて Inngest に再開イベントを送信します。

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

await output.text
cleanup()

第3引数では、Lifecycle 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>

Agentic Loop の Step が完了したときに呼び出されます。

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 の耐久実行 Engine 上でレスポンスを実行し、単一の FullOutput に解決します。Run が中断した場合、generate() は finishReason: 'suspended' を返します。runId オプションは省略可能です。省略すると、generate() が Run ID を作成して result.runId で返します。Run を再開するには、resumeGenerate() を使用します。呼び出し元で中断時の 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 に再接続し、Cache 済みイベントを再生してから Live イベントを配信します。Network が切断された後に使用します。既知の位置から再生を開始するには 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 Entry と Cache 済みイベントを削除します。Run の使用を終えた場合にのみ呼び出してください。Run が中断中で、後から再開する場合は cleanup() を呼び出さないでください。

prepare(messages, options?)

トリガーせずに、耐久実行用の Run を準備します。Inngest Workflow イベントを手動でトリガーするために使用できる、Serialize 済みの Workflow Input を返します。

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 かどうかを確認する Type Guard です。

import { isInngestAgent } from '@mastra/inngest'

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

戻り値：boolean

Stream オプション

stream() は InngestAgentStreamOptions オブジェクトを受け取ります。DurableAgent.stream() と同じ Agent 実行オプションに加えて、Lifecycle Callback をサポートします。

runId?:

string

この Run の一意の識別子。後から resume() または observe() で使用します。

instructions?:

AgentExecutionOptions['instructions']

この Run で使用する Agent のデフォルト instructions を上書きします。

context?:

ModelMessage[]

Agent に渡す追加の Context Message。

memory?:

object

Conversation の永続化と取得に使用する Memory 設定。

requestContext?:

RequestContext

この Run の動的な設定と状態を保持する Request Context。

maxSteps?:

number

実行する最大 Step 数。

toolsets?:

object

この Run で利用できる追加の Tool Set。

clientTools?:

object

実行中に利用できる Client 側の Tool。

toolChoice?:

'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }

Tool の選択方法。

modelSettings?:

object

Temperature など、Model 固有の設定。

requireToolApproval?:

boolean

すべての Tool Call で承認を必須にし、再開されるまで Run を中断します。

autoResumeSuspendedTools?:

boolean

外部からの resume() 呼び出しを待たず、中断した Tool を自動的に再開します。

toolCallConcurrency?:

number

同時に実行する Tool Call の最大数。

includeRawChunks?:

boolean

Provider の Raw Chunk を Stream 出力に含めます。

maxProcessorRetries?:

number

Generation ごとの Processor の最大 Retry 回数。

untilIdle?:

boolean | { maxIdleMs?: number }

設定すると、Agent がアイドル状態になるまで、バックグラウンドタスクの継続処理をまたいで Stream を開いたままにします。デフォルトの5分間のアイドルタイムアウトを使用するには true を渡し、カスタマイズするには { maxIdleMs } を渡します。

onChunk?:

(chunk: ChunkType) => void | Promise<void>

ストリーミングされた Chunk ごとに呼び出されます。

onStepFinish?:

(result: AgentStepFinishEventData) => void | Promise<void>

Agentic Loop の Step が完了したときに呼び出されます。

onFinish?:

(result: AgentFinishEventData) => void | Promise<void>

Run が完了したときに呼び出されます。

onError?:

(error: Error) => void | Promise<void>

Run でエラーが発生したときに呼び出されます。

onSuspended?:

(data: AgentSuspendedEventData) => void | Promise<void>

Tool の承認時など、Run が中断したときに呼び出されます。

observe() は Lifecycle 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

Stream 出力。全文を取得するには output.text を await し、Stream を処理するには output.fullStream を使用します。

fullStream:

ReadableStream

output.fullStream に委譲される、完全なイベント Stream。

runId:

string

一意の Run ID。再接続するには resume() または observe() に渡します。

threadId?:

string

Memory を使用する場合の Thread ID。

resourceId?:

string

Memory を使用する場合の Resource ID。

cleanup:

() => void

PubSub の購読を解除し、Run の Registry Entry を削除します。Run の使用を終えたら呼び出してください。

Inngest Function の提供

@mastra/inngest Package は、HTTP Framework に Inngest Workflow Function を登録するための serve() と createServe() を提供します。

serve(options)

Mastra Workflow を Hono（デフォルトの Framework）で提供します。Mastra から Inngest ベースの Workflow をすべて収集し、Inngest Function として登録します。

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 Function を返す 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 Function。

registerOptions?:

RegisterOptions

Inngest の登録 Handler に渡すオプション。

関連項目

• DurableAgent リファレンス
• Agent クラス
• Inngest デプロイガイド
• PubSub