createInngestAgent()

beta

Breaking changes may occur without a major version bump until the API is stable.

createInngestAgent() wraps an existing Agent with Inngest-powered durable execution. Like createDurableAgent(), it streams events over PubSub and supports resumable streams, but runs the agentic loop on Inngest's execution engine instead of in-process. Use it when a run must survive process restarts or run in a distributed environment.

For in-process durable execution, use createDurableAgent(). For fire-and-forget execution on the built-in workflow engine, use createEventedAgent().

Usage example

Set up the Inngest client, wrap an agent, register it with Mastra, and expose the 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 }),
      },
    ],
  },
})

Stream a response and read the result:

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

const text = await output.text

cleanup()

createInngestAgent(options)

Wraps an Agent with Inngest-powered durable execution and resumable streams.

import { createInngestAgent } from '@mastra/inngest'

const durableAgent = createInngestAgent({ agent, inngest })

Returns: InngestAgent

Parameters

agent:

Agent

The Agent to wrap with Inngest durable execution. Methods not implemented by InngestAgent (e.g., listTools() and getMemory()) delegate to this agent via a Proxy.

inngest:

Inngest

The Inngest client instance. Used to send workflow events and, in SDK v4, publish realtime stream events.

id?:

string

= agent.id

ID override.

name?:

string

= agent.name

Name override.

pubsub?:

PubSub

= InngestPubSub

PubSub instance for streaming events. The default InngestPubSub uses Inngest Realtime, which works across processes.

cache?:

MastraServerCache

Cache for stored stream events, which enables resumable streams. When provided, the PubSub is automatically wrapped with CachingPubSub. If omitted, the agent inherits the cache from the Mastra instance.

mastra?:

Mastra

Mastra instance for observability. Set automatically when the agent is registered with Mastra.

shouldPersistSnapshot?:

(params: { stepResults, workflowStatus }) => boolean

Accepted for API symmetry with createDurableAgent, but ignored: Inngest's step memoization and replay own durability, so InngestAgent always persists suspended snapshots only (for human-in-the-loop resume). A warning is logged if set.

InngestAgent interface

The object returned by createInngestAgent(). It provides the durable execution methods below. Any property or method not explicitly defined (e.g., listTools() and getMemory()) is forwarded to the underlying agent via a Proxy. Thread APIs such as sendSignal(), sendStateSignal(), sendNotificationSignal(), and subscribeToThread() are forwarded too, but a signal that wakes an idle thread starts the run through the durable stream().

Properties

id:

string

The agent ID.

name:

string

The agent name.

agent:

Agent

The underlying Mastra Agent.

inngest:

Inngest

The Inngest client.

cache:

MastraServerCache | undefined

The resolved cache instance, if resumable streams are enabled.

pubsub:

PubSub

The PubSub instance used for streaming events.

Methods

Execution

stream(messages, options?)

Streams a response using Inngest's durable execution engine. The workflow is triggered via an Inngest event after the PubSub subscription is established.

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()

Returns: Promise<InngestAgentStreamResult>

resume(runId, resumeData, options?)

Resumes a suspended Inngest run, for example after a tool approval. Loads the workflow snapshot from storage and finds the suspended step before sending a resume event to Inngest.

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

await output.text
cleanup()

The third argument accepts threadId and resourceId in addition to the lifecycle callbacks:

threadId?:

string

Thread ID to associate with the resumed run.

resourceId?:

string

Resource ID to associate with the resumed run.

onChunk?:

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

Called for each streamed chunk.

onStepFinish?:

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

Called when a step in the agentic loop finishes.

onFinish?:

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

Called when the run finishes.

onError?:

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

Called when the run errors.

onSuspended?:

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

Called when the run suspends.

Returns: Promise<InngestAgentStreamResult>

generate(messages, options?)

Runs a response on Inngest's durable execution engine and resolves with a single FullOutput. If the run suspends, generate() resolves with finishReason: 'suspended'. The runId option is optional. If you omit it, generate() creates a run ID and returns it in result.runId. Continue the run with resumeGenerate(). Use stream() with onSuspended when the caller needs a suspension callback.

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

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

Returns: Promise<FullOutput<TOutput>>

resumeGenerate(resumeData, options)

Resumes a suspended generate() run and resolves with a single FullOutput. Pass the run ID as options.runId, the same as Agent.resumeGenerate(). The older resumeGenerate(runId, resumeData, options?) order still works but is deprecated.

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

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

Returns: Promise<FullOutput<TOutput>>

observe(runId, options?)

Reconnects to an existing run, replaying cached events before delivering live ones. Use this after a network disconnection. Pass offset to start replay from a known position.

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

await output.text

The result from observe() doesn't include threadId or resourceId.

Returns: Promise<Omit<InngestAgentStreamResult, 'threadId' | 'resourceId'>>

注意

The cleanup() returned by observe() destroys the run's registry entries and cached events. Only call it when you are done with the run. If the run is suspended and you intend to resume later, don't call cleanup().

prepare(messages, options?)

Prepares a run for durable execution without triggering it. Returns the serialized workflow input that can be used to manually trigger the Inngest workflow event.

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

Returns:

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

Introspection

isInngestAgent(obj)

Type guard that checks whether an object is an InngestAgent.

import { isInngestAgent } from '@mastra/inngest'

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

Returns: boolean

Stream options

stream() accepts an InngestAgentStreamOptions object. It supports the same agent execution options as DurableAgent.stream(), plus lifecycle callbacks.

runId?:

string

Unique identifier for this run. Use it later with resume() or observe().

instructions?:

AgentExecutionOptions['instructions']

Overrides the agent's default instructions for this run.

context?:

ModelMessage[]

Additional context messages to provide to the agent.

memory?:

object

Memory configuration for conversation persistence and retrieval.

requestContext?:

RequestContext

Request context carrying dynamic configuration and state for this run.

maxSteps?:

number

Maximum number of steps to run.

toolsets?:

object

Additional tool sets available for this run. Toolset tools contain server-side code that cannot be serialized, so they only work when the Inngest worker runs in the same process as the caller. A worker in a separate process throws an error naming the missing tools. Register these tools on the agent instead. The worker matches toolset tools by name, so if a toolset tool shares a name with an agent tool, the agent tool runs.

clientTools?:

object

Client-side tools available during execution.

toolChoice?:

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

Tool selection strategy.

modelSettings?:

object

Model-specific settings such as temperature.

requireToolApproval?:

boolean

Require approval for all tool calls, which suspends the run until resumed.

autoResumeSuspendedTools?:

boolean

Automatically resume tools that suspended, instead of waiting for an external resume() call.

closeOnSuspend?:

boolean

= false

When true, close the stream when the run suspends (for example for tool approval) so fullStream, text, and getFullOutput() resolve at the suspension boundary instead of staying open, matching non-durable Agent.stream(). Useful for callers such as AG-UI or A2A that need the stream to end. Defaults to false, which keeps the stream open across suspension so a later resume can continue streaming on the same reader. resume()/resumeStream() always return a fresh stream.

toolCallConcurrency?:

number

Maximum number of tool calls to execute concurrently.

includeRawChunks?:

boolean

Include raw provider chunks in the stream output.

maxProcessorRetries?:

number

Maximum number of processor retries per generation.

untilIdle?:

boolean | { maxIdleMs?: number }

When set, keeps the stream open across background-task continuations until the agent is idle. Pass true for the default 5-minute idle timeout, or { maxIdleMs } to customise.

onChunk?:

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

Called for each streamed chunk.

onStepFinish?:

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

Called when a step in the agentic loop finishes.

onFinish?:

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

Called when the run finishes.

onError?:

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

Called when the run errors.

onSuspended?:

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

Called when the run suspends, for example for tool approval.

observe() accepts the lifecycle callbacks (onChunk, onStepFinish, onFinish, onError, onSuspended) plus an offset to control where replay starts.

InngestAgentStreamResult

The object returned by stream() and resume(). The observe() method returns the same shape but omits threadId and resourceId.

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

output:

MastraModelOutput

The streaming output. Await output.text for the full text, or consume output.fullStream.

fullStream:

ReadableStream

The full event stream, delegating to output.fullStream.

runId:

string

The unique run ID. Pass it to resume() or observe() to reconnect.

threadId?:

string

Thread ID when using memory.

resourceId?:

string

Resource ID when using memory.

cleanup:

() => void

Unsubscribes from PubSub and clears registry entries for the run. Call it when done with the run.

Event size limit

Inngest Realtime caps each message at about 512 KB. The default InngestPubSub keeps every stream event under 480 KB:

• An oversized finish event is sent without output.steps. The final text, usage, and finish reason still arrive, but steps, toolCalls, and toolResults on the stream result are empty for that run.
• Any other oversized event, such as a tool-result chunk carrying a large tool result, is replaced with a data-oversized-event chunk. Its data holds the original eventType, chunkType, bytes, and limit. The chunk is transient, so it isn't saved to memory.

Both cases log a warning. Only the live stream is reduced: messages saved to memory and the run's workflow snapshot keep the full data.

To avoid the limit:

• Return small tool results, such as an ID or summary, and fetch large data separately.
• Read full results from memory or the workflow run after the stream finishes.
• Pass a pubsub without the limit, such as RedisStreamsPubSub:

src/mastra/index.ts

import { createInngestAgent } from '@mastra/inngest'
import { RedisStreamsPubSub } from '@mastra/redis-streams'

const durableAgent = createInngestAgent({
  agent,
  inngest,
  pubsub: new RedisStreamsPubSub({ url: 'redis://localhost:6379' }),
})

Serving Inngest functions

The @mastra/inngest package provides serve() and createServe() to register Inngest workflow functions with your HTTP framework.

serve(options)

Serves Mastra workflows with Hono (the default framework). Collects all Inngest-backed workflows from Mastra and registers them as Inngest functions.

import { serve } from '@mastra/inngest'

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

createServe(adapter)

Factory that accepts any Inngest serve adapter (inngest/express, inngest/fastify, inngest/next, etc.) and returns a serve function for that framework.

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 options

mastra:

Mastra

The Mastra instance containing registered agents and workflows.

inngest:

Inngest

The Inngest client instance.

functions?:

InngestFunction.Like[]

Additional Inngest functions to serve alongside Mastra workflows.

registerOptions?:

RegisterOptions

Options passed to the Inngest registration handler.

Related

• DurableAgent reference
• Agent class
• Inngest deployment guide
• PubSub