StreamErrorRetryProcessor

StreamErrorRetryProcessor is an error processor that retries transient errors emitted after an LLM stream starts. It includes built-in matching for OpenAI Responses stream errors and supports additional matchers for other provider-specific stream error shapes.

The processor isn't enabled by default in core. Add it to errorProcessors for agents that need stream-error retry handling.

Usage example

Add StreamErrorRetryProcessor to errorProcessors:

src/mastra/agents/openai-agent.ts

import { Agent } from '@mastra/core/agent'
import { StreamErrorRetryProcessor } from '@mastra/core/processors'

export const agent = new Agent({
  name: 'openai-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5',
  errorProcessors: [new StreamErrorRetryProcessor()],
})

How it works

The processor checks the error and its cause chain for:

• Provider retry metadata: isRetryable === true
• Built-in OpenAI Responses stream error matching
• Matcher results: Any configured matcher that returns true

When the error is retryable, the processor returns { retry: true }. It doesn't mutate messages.

When delayMs is set, the processor waits before signaling a retry. This is useful for transient network errors like ECONNRESET where immediately retrying is likely to fail again. The delay can be a fixed number of milliseconds or a function evaluated with the error args (for example, to implement exponential backoff).

Delaying retries

Use delayMs with a custom matcher to retry transient network resets with a wait:

src/mastra/agents/resilient-agent.ts

import { Agent } from '@mastra/core/agent'
import { StreamErrorRetryProcessor } from '@mastra/core/processors'

const isECONNRESET = (error: unknown) => {
  if (!error || typeof error !== 'object') return false
  const code = (error as { code?: unknown }).code
  if (typeof code === 'string' && code.toUpperCase() === 'ECONNRESET') return true
  const message = error instanceof Error ? error.message : undefined
  return typeof message === 'string' && /econnreset|socket hang up/i.test(message)
}

export const agent = new Agent({
  name: 'resilient-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5',
  errorProcessors: [
    new StreamErrorRetryProcessor({
      maxRetries: 2,
      delayMs: ({ retryCount }) => Math.min(1000 * 2 ** retryCount, 30000),
      matchers: [isECONNRESET],
    }),
  ],
})

Default OpenAI Responses matcher

isRetryableOpenAIResponsesStreamError matches OpenAI Responses stream error chunks with type: 'error' or type: 'response.failed'. It retries known transient OpenAI error codes and, as a fallback, errors with explicit retry guidance such as You can retry your request.

StreamErrorRetryProcessor includes this matcher by default. You can also import it and reuse it in custom retry logic.

Constructor parameters

options?:

StreamErrorRetryProcessorOptions

Configuration for retry handling.

number

StreamErrorRetryMatcher[]

number | ((args: ProcessAPIErrorArgs) => number | Promise<number>)

Properties

id:

'stream-error-retry-processor'

Processor identifier.

name:

'Stream Error Retry Processor'

Processor display name.

processAPIError:

(args: ProcessAPIErrorArgs) => ProcessAPIErrorResult | void

Retries stream errors up to the configured retry limit.

Related

• Processor interface
• Processors