Step class

The Step class defines individual units of work within a workflow, encapsulating execution logic, data validation, and input/output handling. It can take a tool, agent, or configured classifier as a parameter to automatically create a step from it.

Usage example

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step-1',
  description: 'passes value from input to output',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    const { value } = inputData
    return {
      value,
    }
  },
})

Define schemas

You can define the step's inputSchema and outputSchema with any library that supports Standard JSON Schema. This includes libraries like Zod, Valibot, and ArkType.

Zod

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    formatted: z.string(),
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

Valibot

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import * as v from 'valibot'
import { toStandardJsonSchema } from '@valibot/to-json-schema'

const step1 = createStep({
  id: 'step-1',
  inputSchema: toStandardJsonSchema(
    v.object({
      message: v.string(),
    }),
  ),
  outputSchema: toStandardJsonSchema(
    v.object({
      formatted: v.string(),
    }),
  ),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

ArkType

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { type } from 'arktype'

const step1 = createStep({
  id: 'step-1',
  inputSchema: type({
    message: 'string',
  }),
  outputSchema: type({
    formatted: 'string',
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

Creating steps from agents

You can create a step directly from an agent. The step will use the agent's name as its ID.

Basic agent step

src/mastra/workflows/test-workflow.ts

import { testAgent } from '../agents/test-agent'

const agentStep = createStep(testAgent)
// inputSchema: { prompt: string }
// outputSchema: { text: string }

Agent step with structured output

Pass structuredOutput to have the agent return typed structured data:

src/mastra/workflows/test-workflow.ts

const articleSchema = z.object({
  title: z.string(),
  summary: z.string(),
  tags: z.array(z.string()),
})

const agentStep = createStep(testAgent, {
  structuredOutput: { schema: articleSchema },
})
// inputSchema: { prompt: string }
// outputSchema: { title: string, summary: string, tags: string[] }

Agent step options

structuredOutput:

{ schema: StandardJSONSchemaV1 }

When provided, the agent returns structured data matching this schema instead of plain text. The step's outputSchema is set to the provided schema.

onFinish:

(result: AgentResult) => void

Callback invoked when the agent completes generation.

Creating steps from classifiers

Pass a Classifier with constructor-configured questions to createStep(). By default, the classifier evaluates the complete inputData. Use state to select a JSON value from the step context.

src/mastra/workflows/ticket-triage.ts

import { Classifier } from '@mastra/core/classifier'
import { createStep } from '@mastra/core/workflows'

const router = new Classifier({
  id: 'ticket-router',
  model,
  questions: {
    route: {
      type: 'choice',
      criteria: {
        billing: 'Billing, invoices, and payments',
        support: 'Account access and product help',
        other: 'Anything else',
      },
    },
    urgent: { type: 'boolean' },
  },
})

const classifyTicket = createStep(router, {
  maxRetries: 2,
  retries: 1,
})

The typed step output is JSON-safe:

• answers.route.choice is the selected choice literal.
• answers.urgent.probability is the probability that the boolean answer is true, not a thresholded boolean.
• Choice and score answers may include distributions.
• usage contains normalized token usage.

maxRetries controls classifier model-call retries. retries controls workflow step retries. Classifier warnings, response metadata, and raw provider data aren't included in workflow output.

A classifier without constructor-configured questions can't be used as a workflow step. The step evaluates its complete input. Add a preceding .map() call when the workflow needs to select or reshape that input.

Constructor parameters

id:

string

Unique identifier for the step

description:

string

Optional description of what the step does

inputSchema:

StandardJSONSchemaV1

Standard JSON Schema defining the input structure

outputSchema:

StandardJSONSchemaV1

Standard JSON Schema defining the output structure

resumeSchema:

StandardJSONSchemaV1

Optional Standard JSON Schema for resuming the step

suspendSchema:

StandardJSONSchemaV1

Optional Standard JSON Schema for suspending the step

stateSchema:

StandardJSONSchemaV1

Optional Standard JSON Schema for the step state. Automatically injected when using Mastra's state system. The stateSchema must be a subset of the workflow's stateSchema. If not specified, type is 'any'.

requestContextSchema:

StandardJSONSchemaV1

Standard JSON Schema for validating request context values. When provided, the context is validated before the step's execute() runs, failing the step if validation fails.

execute:

(params: ExecuteParams) => Promise<any>

Async function containing step logic

ExecuteParams

inputData:

z.infer<TStepInput>

The input data matching the inputSchema

resumeData:

z.infer<TResumeSchema>

The resume data matching the resumeSchema, when resuming the step from a suspended state. Only exists if the step is being resumed.

suspendData:

z.infer<TSuspendSchema>

The suspend data that was originally passed to suspend() when the step was suspended. Only exists if the step is being resumed and was previously suspended with data.

mastra:

Mastra

Access to Mastra services (agents, tools, etc.)

getStepResult:

(step: Step | string) => any

Function to access results from other steps

getInitData:

() => any

Function to access the initial input data of the workflow in any step

suspend:

(suspendPayload: any, suspendOptions?: { resumeLabel?: string }) => Promise<void>

Function to pause workflow execution

state:

z.infer<TState>

The current workflow state. Contains shared values that persist across all steps and suspend/resume cycles. The structure is defined by the step's stateSchema.

setState:

(state: z.infer<TState>) => void

Function to set the state of the workflow. Inject via reducer-like pattern, such as 'setState({ ...state, ...newState })'

runId:

string

Current run id

requestContext?:

RequestContext

Request Context for dependency injection and contextual information.

retryCount?:

number

The retry count for this specific step, it automatically increases each time the step is retried

scorers:

MastraScorers | (({ requestContext }) => MastraScorers | Promise<MastraScorers>)

Scorers that run automatically after the step completes successfully. Each scorer evaluates the step's own input and output, and the results are stored and attached to the step's trace. Provide a map of { [name]: { scorer, sampling? } }, or a function that returns one. Scoring runs asynchronously and doesn't block the workflow. See Scoring step output.

retries:

number

Number of times to retry the step's execute function if it throws.

metadata:

Record<string, any>

Optional key-value pairs for storing additional step information. Values must be serializable (no functions, circular references, etc.).

Scoring step output

Attach scorers to a step to evaluate that step's output automatically, at the point it runs, instead of only scoring the workflow's final answer. This is useful for multi-step and RAG workflows, where you want to see which step degraded quality, for example whether a retrieval step returned relevant chunks before later steps reason over them.

Each scorer receives the step's own input and output. Scoring runs asynchronously after the step succeeds, and the result is stored against the step's trace. Use sampling to control how often a scorer runs.

The following example attaches a scorer to a retrieval step so every execution is scored:

src/mastra/workflows/rag-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'
import { retrievalRelevanceScorer } from '../scorers/retrieval-relevance'

const retrievalStep = createStep({
  id: 'retrieval',
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ query: z.string(), chunks: z.array(z.string()) }),
  scorers: {
    retrievalRelevance: {
      scorer: retrievalRelevanceScorer(),
      sampling: { type: 'ratio', rate: 1 },
    },
  },
  execute: async ({ inputData }) => {
    const chunks = await retrieve(inputData.query)
    return { query: inputData.query, chunks }
  },
})

Attach a scorer to each step you want to measure to build per-step scores across a multi-step workflow. Because scoring is scoped to a single step, you don't need a dedicated cross-step metric to see where quality changes.

Agent and tool steps added with Workflow.agent() and Workflow.tool() accept the same scorers option in their step options.

note

Visit the Scorers overview to learn how live evaluations run and where results are stored, and Custom scorers to build your own.

Related

• Workflow state
• Control flow
• Using agents and tools