Workflow

Type-safe workflow class for orchestrating multi-step processes with built-in error handling

Overview

The Workflow class provides type-safe, composable task execution with built-in error handling, state management, and observability. Workflows are composed of steps that can be chained, parallelized, or conditionally executed.

Type Parameters

TEngineType · any · default: "DefaultEngineType"

The execution engine type

TSteps · Step[] · default: "Step[]"

Array of step types in the workflow

TWorkflowId · string · default: "string"

The workflow identifier type

TState · any · default: "unknown"

The workflow state type

TInput · any · default: "unknown"

The workflow input type

TOutput · any · default: "unknown"

The workflow output type

TPrevSchema · any · default: "TInput"

The previous step’s output schema type

TRequestContext · Record<string, any> · default: "unknown"

The request context schema type

Creating Workflows

createWorkflow

createWorkflow<TWorkflowId, TState, TInput, TOutput, TSteps, TRequestContext>(
  params: WorkflowConfig<TWorkflowId, TState, TInput, TOutput, TSteps, TRequestContext>
): Workflow<DefaultEngineType, TSteps, TWorkflowId, TState, TInput, TOutput, TInput, TRequestContext>

Creates a new workflow with the specified configuration.

params · WorkflowConfig · required

Workflow configuration

Show properties

id · string · required

Unique identifier for the workflow

inputSchema · SchemaWithValidation<TInput> · required

Zod schema defining the input structure

outputSchema · SchemaWithValidation<TOutput> · required

Zod schema defining the output structure

stateSchema · SchemaWithValidation<TState>

Zod schema for workflow state

requestContextSchema · SchemaWithValidation<TRequestContext>

Zod schema for request context validation

description · string

Description of what the workflow does

mastra · Mastra

Reference to the Mastra runtime instance

steps · TSteps

Pre-defined steps for the workflow

executionEngine · ExecutionEngine

Custom execution engine

retryConfig · { attempts?: number; delay?: number }

Retry configuration for failed steps

type · 'default' | 'processor' · default: "'default'"

Type of workflow - ‘processor’ for processor workflows, ‘default’ otherwise

options · WorkflowOptions

Additional workflow options

Show properties

tracingPolicy · TracingPolicy

Policy for trace generation

validateInputs · boolean · default: "true"

Whether to validate inputs with schemas

shouldPersistSnapshot · (params: { stepResults: Record<string, StepResult>; workflowStatus: WorkflowRunStatus }) => boolean

Function to determine if snapshots should be persisted

onFinish · (result: WorkflowFinishCallbackResult) => Promise<void> | void

Called when workflow execution completes (success, failed, suspended, or tripwire)

onError · (errorInfo: WorkflowErrorCallbackInfo) => Promise<void> | void

Called only when workflow execution fails (failed or tripwire status)

workflow · Workflow

A new workflow builder instance

Properties

id · string

The unique identifier for the workflow

description · string | undefined

Description of the workflow

inputSchema · SchemaWithValidation<TInput>

Schema for validating input

outputSchema · SchemaWithValidation<TOutput>

Schema for validating output

stateSchema · SchemaWithValidation<TState> | undefined

Schema for workflow state

requestContextSchema · SchemaWithValidation<TRequestContext> | undefined

Schema for request context

steps · Record<string, StepWithComponent>

The steps in the workflow

engineType · WorkflowEngineType

The execution engine type (always ‘default’)

type · 'default' | 'processor'

Type of workflow

committed · boolean

Whether the workflow has been committed (finalized)

retryConfig · { attempts?: number; delay?: number }

Retry configuration

mastra · Mastra | undefined

The Mastra instance

options · WorkflowOptions

Workflow options including tracing and callbacks

Methods

Building Workflows

then

then<TStep extends Step>(
  step: TStep
): Workflow<...>

Adds a sequential step to the workflow.

step · Step · required

The step to add to the workflow chain

workflow · Workflow

The workflow builder for method chaining

parallel

parallel<TSteps extends Step[]>(
  ...steps: TSteps
): Workflow<...>

Adds steps that execute in parallel.

steps · Step[] · required

The steps to execute in parallel

workflow · Workflow

The workflow builder for method chaining

branch

branch<TSteps extends Step[]>(config: {
  conditions: ConditionFunction[];
  steps: TSteps;
}): Workflow<...>

Adds conditional branching to the workflow.

config · object · required

Branch configuration

Show properties

conditions · ConditionFunction[] · required

Array of condition functions that determine which step to execute

steps · Step[] · required

Steps to potentially execute (one per condition)

workflow · Workflow

The workflow builder for method chaining

loop

loop<TStep extends Step>(config: {
  step: TStep;
  condition: LoopConditionFunction;
  type: 'dowhile' | 'dountil';
}): Workflow<...>

Adds a loop to the workflow.

config · object · required

Loop configuration

Show properties

step · Step · required

The step to execute in the loop

condition · LoopConditionFunction · required

Condition function that determines if the loop should continue

type · 'dowhile' | 'dountil' · required

Loop type: ‘dowhile’ continues while condition is true, ‘dountil’ continues until condition is true

workflow · Workflow

The workflow builder for method chaining

foreach

foreach<TStep extends Step>(config: {
  step: TStep;
  concurrency?: number;
}): Workflow<...>

Adds a foreach loop that iterates over array items.

config · object · required

Foreach configuration

Show properties

step · Step · required

The step to execute for each item

concurrency · number · default: "1"

Number of concurrent executions

workflow · Workflow

The workflow builder for method chaining

sleep

sleep(config: {
  id: string;
  duration?: number;
  fn?: ExecuteFunction;
}): Workflow<...>

Adds a sleep/delay step to the workflow.

config · object · required

Sleep configuration

Show properties

id · string · required

Unique identifier for the sleep step

duration · number

Duration in milliseconds to sleep

fn · ExecuteFunction

Optional function to compute dynamic sleep duration

workflow · Workflow

The workflow builder for method chaining

sleepUntil

sleepUntil(config: {
  id: string;
  date?: Date;
  fn?: ExecuteFunction;
}): Workflow<...>

Adds a sleep until specific date/time step.

config · object · required

Sleep until configuration

Show properties

id · string · required

Unique identifier for the sleep step

date · Date

Date to sleep until

fn · ExecuteFunction

Optional function to compute dynamic target date

workflow · Workflow

The workflow builder for method chaining

commit

commit(): Workflow<...>

Finalizes the workflow and makes it executable.

workflow · Workflow

The finalized workflow ready for execution

Executing Workflows

createRun

createRun(options: {
  runId?: string;
  resourceId?: string;
  requestContext?: RequestContext;
}): Promise<Run>

Creates a new workflow run instance.

options · object

Run creation options

Show properties

runId · string

Custom run ID (auto-generated if not provided)

resourceId · string

Resource ID for multi-tenant scenarios

requestContext · RequestContext

Request context for the run

run · Run

A new workflow run instance

execute

execute(params: {
  inputData: TInput;
  initialState?: TState;
  requestContext?: RequestContext;
  resourceId?: string;
  runId?: string;
  tracingOptions?: TracingOptions;
}): Promise<WorkflowResult<TOutput>>

Executes the workflow with the given input and returns the result.

params · object · required

Execution parameters

Show properties

inputData · TInput · required

Input data for the workflow

initialState · TState

Initial workflow state

requestContext · RequestContext

Request context

resourceId · string

Resource ID for multi-tenant scenarios

runId · string

Custom run ID

tracingOptions · TracingOptions

Tracing configuration

result · WorkflowResult<TOutput>

The workflow execution result

stream

stream(params: {
  inputData: TInput;
  initialState?: TState;
  requestContext?: RequestContext;
  resourceId?: string;
  runId?: string;
  tracingOptions?: TracingOptions;
}): Promise<MastraWorkflowStream>

Executes the workflow and streams the results.

params · object · required

Same as execute parameters

stream · MastraWorkflowStream

A stream of workflow execution events

Workflow Management

listWorkflowRuns

listWorkflowRuns(params?: StorageListWorkflowRunsInput): Promise<WorkflowRuns>

Lists workflow runs with optional filtering.

params · StorageListWorkflowRunsInput

Filter parameters

Show properties

status · WorkflowRunStatus

Filter by status

fromDate · Date

Filter runs created after this date

toDate · Date

Filter runs created before this date

resourceId · string

Filter by resource ID

page · number

Page number for pagination

perPage · number | false

Items per page, or false to fetch all

runs · WorkflowRuns

Object containing runs array and total count

getWorkflowRunById

getWorkflowRunById(params: {
  runId: string;
  fields?: WorkflowStateField[];
}): Promise<WorkflowState | null>

Gets a specific workflow run by ID.

params · object · required

Query parameters

Show properties

runId · string · required

The run ID to retrieve

fields · WorkflowStateField[]

Optional fields to include: ‘result’, ‘error’, ‘payload’, ‘steps’, ‘activeStepsPath’, ‘serializedStepGraph’

state · WorkflowState | null

The workflow run state, or null if not found

deleteWorkflowRun

deleteWorkflowRun(runId: string): Promise<void>

Deletes a workflow run.

runId · string · required

The run ID to delete

Serialization

serialize

serialize(): WorkflowInfo

Serializes the workflow to a JSON-compatible format.

info · WorkflowInfo

Serialized workflow information including steps, schemas, and configuration

Example

import { createWorkflow, createStep } from '@mastra/core/workflows';
import { z } from 'zod';

// Define a workflow
const processDataWorkflow = createWorkflow({
  id: 'process-data',
  description: 'Processes user data through multiple steps',
  inputSchema: z.object({
    userId: z.string(),
    data: z.any()
  }),
  outputSchema: z.object({
    processed: z.boolean(),
    result: z.any()
  })
})
  .then(
    createStep({
      id: 'validate',
      description: 'Validate input data',
      inputSchema: z.object({ userId: z.string(), data: z.any() }),
      outputSchema: z.object({ valid: z.boolean() }),
      execute: async ({ inputData }) => {
        return { valid: true };
      }
    })
  )
  .then(
    createStep({
      id: 'transform',
      description: 'Transform data',
      inputSchema: z.object({ valid: z.boolean() }),
      outputSchema: z.object({ transformed: z.any() }),
      execute: async ({ inputData }) => {
        return { transformed: { ...inputData, timestamp: Date.now() } };
      }
    })
  )
  .commit();

// Execute the workflow
const result = await processDataWorkflow.execute({
  inputData: {
    userId: '123',
    data: { name: 'John' }
  }
});

console.log(result); // { processed: true, result: ... }