How to Connect Parallel Execution, Human-in-the-Loop, and Multi-Agent with Mastra Workflow in a Single TypeScript File

import { createWorkflow } from '@mastra/core/workflows';
 
const contentWorkflow = createWorkflow({ id: 'content-pipeline' })
  .addStep(fetchTopicStep)                                  // Fetch the topic
  .parallel([researchStep, competitorAnalysisStep])         // Run concurrently
  .addStep(editorialStep)                                   // Merge results and edit
  .commit();

// Legacy approach: fan-out using .after()
myWorkflow
  .step(fetchTopicStep)
  .then(researchStep)
  .after(fetchTopicStep)
  .step(competitorAnalysisStep)
  .commit();

import { createStep, createWorkflow } from '@mastra/core/workflows';
import { z } from 'zod';
 
const approvalStep = createStep({
  id: 'approval-step',
  inputSchema: z.object({ content: z.string() }),
  resumeSchema: z.object({
    approved: z.boolean(),
    feedback: z.string().optional(),
  }),
  suspendSchema: z.object({ reason: z.string() }),
 
  async execute({ inputData, resumeData, suspend }) {
    // First run: no approval decision yet, so suspend
    if (resumeData?.approved === undefined) {
      await suspend({ reason: 'Reviewer approval required.' });
      return;
    }
 
    // After resume: resumeData contains the human's decision
    if (!resumeData.approved) {
      throw new Error(`Rejected: ${resumeData.feedback ?? 'No feedback provided'}`);
    }
 
    return { result: inputData.content };
  },
});

// Example: called from a POST /approve handler
await workflow.resume({
  runId: 'run-123',
  stepId: 'approval-step',
  resumeData: { approved: true, feedback: 'Looks good' },
});

import { Agent } from '@mastra/core/agent';
import { anthropic } from '@ai-sdk/anthropic';
 
// Specialized sub-agents
const researchAgent = new Agent({
  name: 'ResearchAgent',
  instructions: 'Deeply research a given topic and summarize the key findings.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const writerAgent = new Agent({
  name: 'WriterAgent',
  instructions: 'Write a reader-friendly blog post based on research findings.',
  model: anthropic('claude-sonnet-4-6'),
});
 
// Supervisor: register delegation targets via the agents array
const supervisorAgent = new Agent({
  name: 'SupervisorAgent',
  instructions: 'Analyze user requests and delegate to the appropriate specialist agent.',
  model: anthropic('claude-opus-4-7'), // Use a more powerful model for routing decisions
  agents: [researchAgent, writerAgent],
});

const result = await supervisorAgent.stream('Write a blog post on AI trends', {
  onDelegationStart({ agent, messages }) {
    console.log(`→ Delegating to ${agent.name}`);
    return messages; // Pass through unchanged; return a transformed version to modify
  },
});

import { Mastra } from '@mastra/core';
import { createStep } from '@mastra/core/workflows';
 
// Register the agent with the Mastra instance
const mastra = new Mastra({
  agents: { ResearchAgent: researchAgent },
});
 
const researchStep = createStep({
  id: 'research-step',
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ result: z.string() }),
 
  async execute({ inputData, mastra }) {
    const agent = mastra.getAgent('ResearchAgent'); // Reference a registered agent
    const response = await agent.generate(inputData.query);
    return { result: response.text };
  },
});

import { createWorkflow, createStep } from '@mastra/core/workflows';
import { z } from 'zod';
 
// fetchTopicStep: assumed to be injected externally in this example (scaffolding or separate file)
// Minimal stub example:
// const fetchTopicStep = createStep({
//   id: 'fetch-topic', inputSchema: z.object({ keyword: z.string() }),
//   outputSchema: z.object({ topic: z.string() }),
//   async execute({ inputData }) { return { topic: inputData.keyword }; },
// });
 
// 1. Two steps to run in parallel
const researchStep = createStep({
  id: 'research',
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
  async execute({ inputData, mastra }) {
    const agent = mastra.getAgent('ResearchAgent');
    const res = await agent.generate(`Research the topic: ${inputData.topic}`);
    return { summary: res.text };
  },
});
 
const competitorStep = createStep({
  id: 'competitor-analysis',
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ analysis: z.string() }),
  async execute({ inputData, mastra }) {
    const agent = mastra.getAgent('ResearchAgent');
    const res = await agent.generate(`Analyze the competitive landscape for: ${inputData.topic}`);
    return { analysis: res.text };
  },
});
 
// assessQuality: returns a quality score between 0 and 1 for a draft.
// The actual implementation varies by project — LLM-based evaluation, rule-based checks
// (length, keyword density, etc.), etc. Simple character-count dummy example below:
function assessQuality(text: string): number {
  if (text.length < 300) return 0.4;
  if (text.length < 800) return 0.6;
  return 0.85;
}
 
// 2. Editorial + quality review step (with HITL)
const editorialStep = createStep({
  id: 'editorial',
  inputSchema: z.object({ summary: z.string(), analysis: z.string() }),
  resumeSchema: z.object({ approved: z.boolean(), feedback: z.string().optional() }),
  suspendSchema: z.object({ draft: z.string(), reason: z.string() }),
  async execute({ inputData, resumeData, suspend, mastra }) {
    // Resume path: a human has made a decision
    if (resumeData !== undefined) {
      if (!resumeData.approved) {
        // This error puts the entire workflow into a failed state.
        // If retry logic is needed, handle it separately at the workflow level.
        throw new Error(`Editorial rejected: ${resumeData.feedback}`);
      }
      return { published: true };
    }
 
    // Initial run: generate draft
    const agent = mastra.getAgent('WriterAgent');
    const draft = await agent.generate(
      `Research: ${inputData.summary}\nCompetitor analysis: ${inputData.analysis}\n\nWrite a blog post draft based on this.`
    );
 
    const qualityScore = assessQuality(draft.text);
 
    // Request human review if quality is below threshold
    if (qualityScore < 0.7) {
      await suspend({ draft: draft.text, reason: `Quality score below threshold: ${qualityScore}` });
      return;
    }
 
    return { published: true, draft: draft.text };
  },
});
 
// 3. Assemble the workflow
const contentPipelineWorkflow = createWorkflow({ id: 'content-pipeline' })
  .addStep(fetchTopicStep)
  .parallel([researchStep, competitorStep])  // Parallel execution
  .addStep(editorialStep)                    // Merge results, edit, + HITL
  .commit();

import { Agent } from '@mastra/core/agent';
import { anthropic } from '@ai-sdk/anthropic';
 
const techSupportAgent = new Agent({
  name: 'TechSupportAgent',
  instructions: 'Diagnose technical issues and guide users to solutions.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const billingAgent = new Agent({
  name: 'BillingAgent',
  instructions: 'Handle billing inquiries. Work only with masked personal information.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const refundAgent = new Agent({
  name: 'RefundAgent',
  instructions: 'Review refund requests and guide users through the process.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const supportSupervisor = new Agent({
  name: 'SupportSupervisor',
  instructions: `Analyze customer inquiries and delegate to the appropriate specialist agent.
    - Technical issues → TechSupportAgent
    - Billing inquiries → BillingAgent
    - Refund requests → RefundAgent`,
  model: anthropic('claude-opus-4-7'), // A powerful model handles routing decisions
  agents: [techSupportAgent, billingAgent, refundAgent],
});
 
// The return value of onDelegationStart is the actual message delivered to the sub-agent.
// Card numbers are masked before being sent to billing/refund agents.
const result = await supportSupervisor.stream(customerMessage, {
  onDelegationStart({ agent, messages }) {
    if (['BillingAgent', 'RefundAgent'].includes(agent.name)) {
      return messages.map((msg) => ({
        ...msg,
        content:
          typeof msg.content === 'string'
            ? msg.content.replace(/\d{4}-\d{4}-\d{4}-\d{4}/g, '****-****-****-****')
            : msg.content,
      }));
    }
    return messages;
  },
});