Instruction file imported from pehqge/sdr-agent-system (.cursor/rules/mastra/workflows/steps-patterns.mdc). Copyright stays with the author.

description: Patterns for creating and reusing steps in Mastra: step types, composition, reuse, and integration with agents/tools. Apply when creating steps. glob: "/workflows//*.ts"

Step Patterns in Mastra

Overview

Steps are self-contained work units within workflows. Each step receives input, processes, and produces output following well-defined schemas.

Fundamental principles:

• ✅ Reusable: Use the same step in multiple workflows
• ✅ Testable: Test each step in isolation
• ✅ Composable: Combine steps in different ways
• ✅ Reliable: Schemas catch errors early

Step Structure

Required Components

const myStep = createStep({
  id: "unique-step-name",              // 1. Unique identifier
  description: "What this step does",   // 2. Clear description
  inputSchema: z.object({              // 3. Input schema
    // Defines expected structure
  }),
  outputSchema: z.object({              // 4. Output schema
    // Defines returned structure
  }),
  execute: async ({ inputData }) => {   // 5. Execution function
    // Logic here
    return {
      // Data matching outputSchema
    };
  },
});

Optional Components

const myStep = createStep({
  // ... required components ...
  
  resumeSchema: z.object({              // For suspend/resume
    // Data structure when resuming
  }),
  suspendSchema: z.object({            // For suspend/resume
    // Data structure when suspending
  }),
  retries: 3,                          // Retry attempts on error
  stateSchema: z.object({              // For state management
    // Workflow state structure
  }),
});

Step Types

1. Simple Step (Direct Logic)

Steps that execute direct logic without external dependencies.

const validateStep = createStep({
  id: "validate-content",
  description: "Validates incoming text content",
  inputSchema: z.object({
    content: z.string().min(1),
  }),
  outputSchema: z.object({
    content: z.string(),
    wordCount: z.number(),
    isValid: z.boolean(),
  }),
  execute: async ({ inputData }) => {
    const { content } = inputData;
    const wordCount = content.trim().split(/\s+/).length;
    const isValid = wordCount >= 5;

    if (!isValid) {
      throw new Error(`Content too short: ${wordCount} words`);
    }

    return {
      content: content.trim(),
      wordCount,
      isValid,
    };
  },
});

Characteristics:

• Simple and direct logic
• No external calls
• Business rule validation
• Can throw errors to stop workflow

2. Step with Agent

Steps that use AI agents for intelligent processing.

const aiAnalysisStep = createStep({
  id: "ai-analysis",
  description: "AI-powered content analysis",
  inputSchema: z.object({
    content: z.string(),
    type: z.string(),
  }),
  outputSchema: z.object({
    content: z.string(),
    type: z.string(),
    aiAnalysis: z.object({
      score: z.number(),
      feedback: z.string(),
    }),
  }),
  execute: async ({ inputData, mastra }) => {
    const { content, type } = inputData;

    // Create prompt
    const prompt = `
Analyze this ${type} content: "${content}"
Provide quality score (1-10) and feedback as JSON.
Format: {"score": number, "feedback": "string"}
    `;

    // Get and call agent
    const agent = mastra.getAgent("contentAgent");
    const { text } = await agent.generate([
      { role: "user", content: prompt },
    ]);

    // Parse response
    let aiAnalysis;
    try {
      aiAnalysis = JSON.parse(text);
    } catch {
      aiAnalysis = {
        score: 7,
        feedback: "AI analysis completed. " + text,
      };
    }

    return {
      ...inputData,
      aiAnalysis,
    };
  },
});

Characteristics:

• Accesses mastra via execute function
• Uses mastra.getAgent(id) to get agents
• Needs to process agent response
• Must have fallback for response parsing

Alternative: Agent as Step directly:

import { contentAgent } from "../agents/content-agent";

const agentStep = createStep(contentAgent);

// Agent expects { prompt: string } and returns { text: string }

3. Step with Tool

Steps that execute tools for specific operations.

import { sendEmailTool } from "../tools/send-email";

const emailStep = createStep({
  id: "send-email",
  description: "Sends email notification",
  inputSchema: z.object({
    userId: z.string(),
    message: z.string(),
  }),
  outputSchema: z.object({
    userId: z.string(),
    message: z.string(),
    emailSent: z.boolean(),
    emailId: z.string().optional(),
  }),
  execute: async ({ inputData, runtimeContext }) => {
    const { userId, message } = inputData;

    // Execute tool
    const result = await sendEmailTool.execute({
      context: {
        to: userId,
        subject: "Notification",
        body: message,
      },
      runtimeContext,
    });

    return {
      ...inputData,
      emailSent: result.success,
      emailId: result.emailId,
    };
  },
});

Alternative: Tool as Step directly:

import { sendEmailTool } from "../tools/send-email";

const toolStep = createStep(sendEmailTool);

// Tool is used directly, inputSchema must match tool's context

4. Step with External API

Steps that make calls to external APIs.

const fetchDataStep = createStep({
  id: "fetch-data",
  description: "Fetches data from external API",
  inputSchema: z.object({
    apiKey: z.string(),
    endpoint: z.string(),
  }),
  outputSchema: z.object({
    apiKey: z.string(),
    endpoint: z.string(),
    data: z.any(),
    status: z.number(),
  }),
  execute: async ({ inputData }) => {
    const { apiKey, endpoint } = inputData;

    try {
      const response = await fetch(endpoint, {
        headers: {
          Authorization: `Bearer ${apiKey}`,
        },
      });

      if (!response.ok) {
        throw new Error(`API error: ${response.status}`);
      }

      const data = await response.json();

      return {
        ...inputData,
        data,
        status: response.status,
      };
    } catch (error) {
      throw new Error(`Failed to fetch data: ${(error as Error).message}`);
    }
  },
  retries: 3,  // Try 3 times on error
});

Characteristics:

• Robust error handling
• Configure retries for network requests
• Validate response before returning
• Preserve original data in return

Reuse Patterns

CRITICAL: Check for Existing Tools First

ALWAYS verify if an existing tool is available before creating a new step or tool for workflow functionality.

Workflow:

1. Before implementing a workflow step, search existing tools in the project
2. If a tool exists that matches the step's functionality → Use it in the step
3. Only create new tools when no existing tool provides the needed functionality

Why this matters:

• ✅ Avoids code duplication
• ✅ Maintains consistency across the codebase
• ✅ Reduces maintenance burden
• ✅ Leverages tested, working code

Example:

// ❌ WRONG: Creating new tool when one exists
const weatherStep = createStep({
  id: "get-weather",
  execute: async ({ inputData }) => {
    // Duplicating weather logic that exists in weatherTool
    const response = await fetch(`https://api.weather.com/...`);
    // ...
  },
});

// ✅ CORRECT: Using existing tool
import { weatherTool } from "../tools/weather-tool";

const weatherStep = createStep({
  id: "get-weather",
  inputSchema: z.object({
    location: z.string(),
  }),
  outputSchema: z.object({
    location: z.string(),
    temperature: z.number(),
    conditions: z.string(),
  }),
  execute: async ({ inputData, runtimeContext, mastra }) => {
    // Use existing tool
    const result = await weatherTool.execute({
      context: {
        location: inputData.location,
      },
      runtimeContext,
    });
    
    return {
      location: inputData.location,
      temperature: result.temperature,
      conditions: result.conditions,
    };
  },
});

// ✅ EVEN BETTER: Use tool directly as step (if schemas align)
const weatherStep = createStep(weatherTool);

Search strategy before creating:

1. Check src/mastra/tools/ directory for existing tools
2. Review tool descriptions in tools/index.ts
3. Search codebase for similar functionality
4. Check if MCP tools are available for the use case

Exception: Only create new tools when:

• Existing tool doesn't match requirements (different input/output)
• Tool needs significant modification that breaks existing usage
• Functionality is genuinely new and doesn't exist

Pattern 1: Reusable Step

Create steps that can be used in multiple workflows.

// src/mastra/workflows/steps/validation.ts
export const validateContentStep = createStep({
  id: "validate-content",
  description: "Validates text content",
  inputSchema: z.object({
    content: z.string().min(1),
  }),
  outputSchema: z.object({
    content: z.string(),
    wordCount: z.number(),
    isValid: z.boolean(),
  }),
  execute: async ({ inputData }) => {
    // Validation logic
  },
});

// Use in multiple workflows
import { validateContentStep } from "./steps/validation";

export const workflow1 = createWorkflow({...})
  .then(validateContentStep)  // ✅ Reused
  .commit();

export const workflow2 = createWorkflow({...})
  .then(validateContentStep)  // ✅ Reused
  .commit();

Benefits:

• ✅ DRY (Don't Repeat Yourself)
• ✅ Consistency across workflows
• ✅ Easy maintenance (change in one place)

Pattern 2: Parameterized Step

Create steps that accept configuration via input schema.

const processStep = createStep({
  id: "process-content",
  inputSchema: z.object({
    content: z.string(),
    options: z.object({
      trim: z.boolean().default(true),
      caseSensitive: z.boolean().default(false),
      maxLength: z.number().optional(),
    }),
  }),
  outputSchema: z.object({
    content: z.string(),
    processed: z.boolean(),
  }),
  execute: async ({ inputData }) => {
    const { content, options } = inputData;
    let processed = content;

    if (options.trim) {
      processed = processed.trim();
    }

    if (!options.caseSensitive) {
      processed = processed.toLowerCase();
    }

    if (options.maxLength && processed.length > options.maxLength) {
      processed = processed.substring(0, options.maxLength);
    }

    return {
      content: processed,
      processed: true,
    };
  },
});

Pattern 3: Step Composition

Combine smaller steps to create larger steps.

// Smaller steps
const validateStep = createStep({...});
const enhanceStep = createStep({...});
const summarizeStep = createStep({...});

// Composed workflow (can be used as step in another workflow)
export const contentProcessingWorkflow = createWorkflow({
  id: "content-processing",
  inputSchema: z.object({ content: z.string() }),
  outputSchema: z.object({ content: z.string(), summary: z.string() }),
})
  .then(validateStep)
  .then(enhanceStep)
  .then(summarizeStep)
  .commit();

// Use workflow as step in another workflow
export const mainWorkflow = createWorkflow({...})
  .then(contentProcessingWorkflow)  // ✅ Workflow as step!
  .commit();

Data Preservation

Pattern: Spread Previous Data

✅ GOOD: Preserve previous data and add new

execute: async ({ inputData }) => {
  const { content, wordCount } = inputData;
  
  // Process...
  const metadata = { readingTime: Math.ceil(wordCount / 200) };

  return {
    ...inputData,      // ✅ Preserves all previous data
    metadata,          // ✅ Adds new data
  };
}

❌ BAD: Lose previous data

execute: async ({ inputData }) => {
  // Process...
  const metadata = { readingTime: 5 };

  return {
    metadata,  // ❌ Lost content, wordCount, etc!
  };
}

When Not to Preserve

Sometimes it's intentional not to preserve:

// Step that completely transforms data
const transformStep = createStep({
  inputSchema: z.object({
    rawData: z.any(),
  }),
  outputSchema: z.object({
    processedData: z.object({
      // New structure
    }),
  }),
  execute: async ({ inputData }) => {
    // Complete transformation
    return {
      processedData: transform(inputData.rawData),
      // Doesn't preserve rawData intentionally
    };
  },
});

Accessing Mastra Resources

Access Agents

execute: async ({ mastra }) => {
  const agent = mastra.getAgent("agentId");
  const response = await agent.generate([...]);
}

Access Tools

execute: async ({ mastra }) => {
  const tool = mastra.getTool("toolId");
  const result = await tool.execute({ context: {...}, runtimeContext });
}

Access Other Workflows

execute: async ({ mastra }) => {
  const workflow = mastra.getWorkflow("workflowId");
  const run = await workflow.createRunAsync();
  const result = await run.start({ inputData: {...} });
}

Access Previous Step Results

execute: async ({ getStepResult }) => {
  const step1Result = getStepResult(step1);
  // Use result in current logic
}

Access Initial Workflow Input

execute: async ({ getInitData }) => {
  const initialInput = getInitData();
  // Use initial data even in later steps
}

Error Handling

Throwing Errors

Errors stop the workflow at that step:

execute: async ({ inputData }) => {
  if (!isValid) {
    throw new Error(`Validation failed: ${reason}`);
  }
}

Automatic Retry

Configure retries for steps that may fail temporarily:

const apiStep = createStep({
  // ...
  retries: 3,  // Try 3 times before failing
  execute: async ({ inputData }) => {
    // Logic that may fail (API calls, etc.)
  },
});

Try-Catch with Fallback

For recoverable errors:

execute: async ({ inputData }) => {
  try {
    const result = await riskyOperation();
    return { ...inputData, result };
  } catch (error) {
    // Fallback
    return {
      ...inputData,
      result: defaultValue,
      error: (error as Error).message,
    };
  }
}

Best Practices

✅ GOOD

1. Check for existing tools before creating steps:

   // ✅ Always verify existing tools first
   import { existingTool } from "../tools/existing-tool";
   
   const step = createStep({
     execute: async ({ inputData, runtimeContext }) => {
       return await existingTool.execute({
         context: inputData,
         runtimeContext,
       });
     },
   });

2. Unique and descriptive IDs:

   id: "validate-content"  // ✅

3. Clear descriptions:

   description: "Validates incoming text content and counts words"

4. Well-defined schemas:

   inputSchema: z.object({
     content: z.string().min(1, "Content cannot be empty"),
   })

5. Preserve data when appropriate:

   return {
     ...inputData,
     newField: value,
   };

6. Reuse common steps and tools:

   // Create once, use in many workflows

❌ AVOID

1. Creating new tools without checking existing ones:

   // ❌ Creating duplicate tool logic in step
   const step = createStep({
     execute: async ({ inputData }) => {
       // Duplicating tool that already exists
       return await fetch("...");
     },
   });

2. Generic IDs:

   id: "step1"  // ❌

3. Overly permissive schemas:

   inputSchema: z.any()  // ❌

4. Unnecessarily losing data:

   return { newField: value };  // ❌ Loses previous data

5. Duplicating logic:

   // Creating same step multiple times ❌

6. Generic errors:

   throw new Error("Error");  // ❌ No context

File Organization

Recommended Structure

src/mastra/workflows/
├── index.ts                    # Exports all workflows
├── content-workflow.ts         # Specific workflow
├── steps/                      # Reusable steps
│   ├── validation.ts
│   ├── enhancement.ts
│   └── analysis.ts
└── shared/                     # Steps shared across workflows
    ├── common-validation.ts
    └── utilities.ts

Exporting Steps

// src/mastra/workflows/steps/validation.ts
export const validateContentStep = createStep({...});
export const validateEmailStep = createStep({...});

Using Steps

// src/mastra/workflows/content-workflow.ts
import { validateContentStep } from "./steps/validation";

export const workflow = createWorkflow({...})
  .then(validateContentStep)
  .commit();

Complete Example: Reusable Step

// src/mastra/workflows/steps/content-analysis.ts
import { createStep } from "@mastra/core/workflows";
import { z } from "zod";

export const analyzeContentStep = createStep({
  id: "analyze-content",
  description: "Analyzes content for word count, reading time, and difficulty",
  inputSchema: z.object({
    content: z.string().min(1),
    type: z.enum(["article", "blog", "social"]).default("article"),
  }),
  outputSchema: z.object({
    content: z.string(),
    type: z.string(),
    wordCount: z.number(),
    readingTime: z.number(),
    difficulty: z.enum(["easy", "medium", "hard"]),
  }),
  execute: async ({ inputData }) => {
    const { content, type } = inputData;
    const words = content.trim().split(/\s+/);
    const wordCount = words.length;
    const readingTime = Math.ceil(wordCount / 200);

    let difficulty: "easy" | "medium" | "hard" = "easy";
    if (wordCount > 100) difficulty = "medium";
    if (wordCount > 300) difficulty = "hard";

    return {
      content: content.trim(),
      type,
      wordCount,
      readingTime,
      difficulty,
    };
  },
});

References

• Official docs: Step Class Reference
• Official docs: Agents and Tools in Workflows
• Examples: See .cursor/rules/helpers/examples-library.mdc