Mastra Workflows

Mastra Workflows — DAG Orchestration

Build type-safe DAG workflows with @mastra/core/workflows.

Quick Start

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

const stepA = createStep({
  id: "step-a",
  inputSchema: z.object({ value: z.string() }),
  outputSchema: z.object({ result: z.string() }),
  execute: async ({ inputData }) => ({ result: inputData.value.toUpperCase() }),
});

const workflow = createWorkflow({ id: "my-workflow", inputSchema: z.object({ value: z.string() }) })
  .then(stepA)
  .commit();

const run = workflow.createRun();
const result = await run.start({ inputData: { value: "hello" } });
// result.results["step-a"].output → { result: "HELLO" }

Imports

Always import from @mastra/core/workflows, NOT from @mastra/core:

import { createStep, createWorkflow } from "@mastra/core/workflows";

Step Anatomy

const step = createStep({
  id: "unique-step-id",           // Required: unique identifier
  inputSchema: z.object({...}),   // Required: Zod schema for input
  outputSchema: z.object({...}),  // Required: Zod schema for output
  execute: async (params) => {    // Required: execution function
    const { inputData, getStepResult, getInitData, suspend, mapiData } = params;
    // inputData: typed input from previous step or workflow
    // getStepResult: access results from earlier named steps
    // getInitData: access original workflow input
    // suspend: pause execution (see references/suspend-resume.md)
    return { /* matches outputSchema */ };
  },
});

Control Flow

Chain steps with fluent API methods. Each returns the workflow for chaining.

Critical: Always call .commit() after the last control flow method.

For detailed patterns, examples, and advanced composition: See references/control-flow.md

Error Handling

const step = createStep({
  id: "retry-step",
  retryConfig: { attempts: 3, delay: 1000 },  // Auto-retry
  execute: async ({ inputData, bail }) => {
    if (fatalError) bail("Unrecoverable");     // Skip retries, fail immediately
    return { result: "ok" };
  },
});

Lifecycle callbacks on workflow run:

const result = await run.start({
  inputData: {...},
  onFinish: (result) => { /* always called */ },
  onError: (error) => { /* called on failure */ },
});

For retry strategies, bail patterns, conditional error branching: See references/error-handling.md

Suspend & Resume

Pause workflow execution for human input or external events:

const humanStep = createStep({
  id: "approval",
  inputSchema: z.object({ request: z.string() }),
  outputSchema: z.object({ approved: z.boolean() }),
  resumeSchema: z.object({ decision: z.enum(["approve", "reject"]) }),
  execute: async ({ inputData, suspend, resumeData }) => {
    if (!resumeData) {
      await suspend({ request: inputData.request });
      return undefined as never;
    }
    return { approved: resumeData.decision === "approve" };
  },
});

// Resume later:
await run.resume({ step: humanStep, resumeData: { decision: "approve" } });

For suspend patterns, multi-step suspend, nested workflows: See references/suspend-resume.md

State & Streaming

Workflow state — shared mutable state across steps:

const workflow = createWorkflow({
  id: "stateful",
  inputSchema: z.object({...}),
  stateSchema: z.object({ count: z.number() }),
});

// In step execute:
execute: async ({ state, setState }) => {
  setState({ ...state, count: state.count + 1 });
  return { result: state.count };
}

Streaming — real-time step events:

const run = workflow.createRun();
const stream = run.stream({ inputData: {...} });
for await (const chunk of stream) {
  // chunk contains step progress, transitions, outputs
}

For state patterns, streaming events, nested state: See references/state-streaming.md

Nested Workflows

Use a workflow as a step inside another workflow:

const subWorkflow = createWorkflow({ id: "sub", inputSchema, outputSchema })
  .then(stepX)
  .then(stepY)
  .commit();

const mainWorkflow = createWorkflow({ id: "main", inputSchema })
  .then(subWorkflow)
  .commit();

Agent Steps

Create steps from Mastra agents:

import { Agent } from "@mastra/core/agents";

const agent = new Agent({ name: "analyzer", instructions: "...", model });
const agentStep = createStep(agent);
// or with structured output:
const typedAgentStep = createStep(agent, { structuredOutput: { schema: z.object({...}) } });

Key Rules

1. Import path: @mastra/core/workflows — never @mastra/core
2. Always .commit() after the last chaining method
3. Zod schemas required on every step (input + output)
4. Parallel output is keyed by step ID — destructure accordingly
5. Branch output uses optional fields — next step must handle missing keys
6. getStepResult<typeof step>("step-id") for accessing earlier results (type-safe)
7. Foreach returns an array — next step receives { results: [...] }
8. Loop steps receive iterationCount in their context