Vue d'ensemble des workflows

Les workflows vous permettent de définir des séquences de tâches complexes à l'aide d'étapes claires et structurées, plutôt que de vous reposer sur le raisonnement d'un seul agent. Ils vous donnent un contrôle total sur le découpage des tâches, la circulation des données entre elles et le moment où chaque élément est exécuté. Par défaut, les workflows utilisent le moteur d'exécution intégré, mais ils peuvent aussi être déployés sur des exécuteurs de workflows comme Inngest afin de bénéficier d'une infrastructure gérée.

Pour les agents IA : exécutez npx mastra api workflow run start test-workflow '{"inputData":{"message":"Hello, world!"}}' pour démarrer directement ce workflow sans écrire de script temporaire. Cette commande nécessite un serveur Mastra actif ; démarrez le serveur local avec npx mastra dev, ou fournissez l'URL de base du serveur accessible avec --url. Exécutez npx mastra api workflow run start --schema avant de construire une entrée différente. Installez la skill de Mastra avec npx skills add mastra-ai/skills --skill mastra pour obtenir des instructions complètes sur la découverte de l'API en ligne de commande, le ciblage, les schémas, l'authentification et la gestion des erreurs.

Quand utiliser les workflows

Utilisez les workflows pour les tâches clairement définies à l'avance, qui comportent plusieurs étapes à exécuter dans un ordre précis. Ils vous offrent un contrôle fin sur la circulation et la transformation des données entre les étapes, ainsi que sur les primitives appelées à chaque stade.

Principes fondamentaux

Les workflows Mastra reposent sur les principes suivants :

• Définir des étapes avec createStep, en précisant les schémas d'entrée et de sortie ainsi que la logique métier.
• Composer les étapes avec createWorkflow pour définir le flux d'exécution.
• Exécuter les workflows afin de lancer la séquence complète, avec une prise en charge intégrée de la suspension, de la reprise et de la diffusion des résultats en streaming.

Créer une étape de workflow

Les étapes sont les composants élémentaires des workflows. Créez une étape avec createStep(), puis définissez les données qu'elle accepte et renvoie à l'aide de inputSchema et outputSchema. Vous pouvez définir ces deux schémas avec Standard JSON Schema (Zod, Valibot, ArkType, etc.).

La fonction execute définit le comportement de l'étape. Utilisez-la pour appeler des fonctions de votre base de code, des API externes, des agents ou des outils.

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(),
    }
  },
})

Consultez Step pour obtenir la liste complète des options de configuration.

Utiliser des agents et des outils

Les étapes d'un workflow peuvent également appeler des agents enregistrés, ou importer et exécuter directement des outils. Consultez la page Utiliser des outils pour en savoir plus.

Créer un workflow

Créez un workflow avec createWorkflow(), puis définissez les données qu'il accepte et renvoie à l'aide de inputSchema et outputSchema. Vous pouvez définir ces deux schémas avec Standard JSON Schema (Zod, Valibot, ArkType, etc.). Ajoutez les étapes avec .then(), puis finalisez le workflow avec .commit().

Zod

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    output: z.string()
  })
})
  .then(step1)
  .commit();

Valibot

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import * as v from "valibot";
import { toStandardJsonSchema } from "@valibot/to-json-schema";

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: toStandardJsonSchema(v.object({
    message: v.string()
  })),
  outputSchema: toStandardJsonSchema(v.object({
    output: v.string()
  }))
})
  .then(step1)
  .commit();

ArkType

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: type({
    message: "string"
  }),
  outputSchema: type({
    output: "string"
  })
})
  .then(step1)
  .commit();

Consultez la classe Workflow pour obtenir la liste complète des options de configuration.

Comprendre le flux de contrôle

Les workflows peuvent être composés à l'aide de plusieurs méthodes. La méthode choisie détermine la structure du schéma de chaque étape. Consultez la page Flux de contrôle pour en savoir plus.

Studio

Ouvrez Studio, puis sélectionnez un workflow dans l'onglet Workflows.

• Vue graphique : le panneau central représente les étapes du workflow et son flux d'exécution.
• Formulaire d'entrée : la barre latérale droite génère un formulaire à partir de l'inputSchema du workflow. Remplissez-le, puis lancez l'exécution.
• État en direct : pendant l'exécution, le graphique actualise en temps réel l'état de chaque étape. La barre latérale affiche l'entrée, la sortie, l'état et les journaux du workflow.
• Voyage dans le temps : une fois l'exécution terminée, rejouez chaque étape afin de l'inspecter ou de la réessayer.

État du workflow

L'état du workflow vous permet de partager des valeurs entre les étapes sans les transmettre par les schémas inputSchema et outputSchema de chacune d'elles. Utilisez-le pour suivre la progression, accumuler des résultats ou partager une configuration dans l'ensemble du workflow.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  stateSchema: z.object({ counter: z.number() }),
  execute: async ({ inputData, state, setState }) => {
    // Read from state
    console.log(state.counter)

    // Update state for subsequent steps
    setState({ ...state, counter: state.counter + 1 })

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

Consultez État du workflow pour une documentation complète sur les schémas d'état, l'état initial, la persistance au fil des suspensions et reprises, ainsi que les workflows imbriqués.

Utiliser des workflows comme étapes

Utilisez un workflow comme étape afin de réemployer sa logique dans une composition plus vaste. Les entrées et sorties suivent les mêmes règles de schéma que celles décrites dans les principes fondamentaux.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});
const step2 = createStep({...});

const childWorkflow = createWorkflow({
  id: "child-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(step1)
  .then(step2)
  .commit();

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(childWorkflow)
  .commit();

Cloner un workflow

Clonez un workflow avec cloneWorkflow() lorsque vous souhaitez réutiliser sa logique tout en assurant un suivi distinct sous un nouvel ID. Chaque clone s'exécute indépendamment et apparaît comme un workflow distinct dans les journaux et les outils d'observabilité.

src/mastra/workflows/test-workflow.ts

import { cloneWorkflow } from "@mastra/core/workflows";

const step1 = createStep({...});

const parentWorkflow = createWorkflow({...})
const clonedWorkflow = cloneWorkflow(parentWorkflow, { id: "cloned-workflow" });

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .then(clonedWorkflow)
  .commit();

Enregistrer un workflow

Enregistrez votre workflow dans l'instance Mastra afin de le rendre disponible dans toute votre application. Une fois enregistré, il peut être appelé depuis des agents ou des outils et accède aux ressources partagées, notamment les fonctionnalités de journalisation et d'observabilité :

src/mastra/index.ts

import { Mastra } from '@mastra/core/mastra'
import { testWorkflow } from './workflows/test-workflow'

export const mastra = new Mastra({
  workflows: { testWorkflow },
})

Obtenir une référence à un workflow

Vous pouvez exécuter des workflows depuis des agents, des outils, le client Mastra ou la ligne de commande. Selon votre configuration, obtenez une référence en appelant .getWorkflow() sur votre instance mastra ou mastraClient :

const testWorkflow = mastra.getWorkflow('testWorkflow')

info

mastra.getWorkflow() est préférable à une importation directe pour deux raisons :

1. Cette méthode donne accès à la configuration de l'instance Mastra (journaliseur, télémétrie, stockage, agents enregistrés et bases de données vectorielles).
2. Elle offre une inférence complète des types TypeScript pour les schémas d'entrée et de sortie du workflow.

Utilisez getWorkflow() avec la clé d'enregistrement du workflow (celle employée lors de son ajout à Mastra). getWorkflowById() permet bien de récupérer les workflows à partir de leur propriété id, mais n'offre pas le même niveau d'inférence de types.

Exécuter des workflows

Les workflows peuvent être exécutés selon deux modes : start attend la fin de toutes les étapes avant de renvoyer une valeur, tandis que stream émet des événements pendant l'exécution. Choisissez l'approche adaptée à votre cas d'usage : start lorsque seul le résultat final vous intéresse, et stream lorsque vous souhaitez suivre la progression ou déclencher des actions à mesure que les étapes s'achèvent.

.start()

Créez une instance d'exécution du workflow avec createRun(), puis appelez .start() en fournissant un inputData conforme à l'inputSchema du workflow. Le workflow exécute toutes les étapes et renvoie le résultat final.

const run = await testWorkflow.createRun()

const result = await run.start({
  inputData: {
    message: 'Hello world',
  },
})

if (result.status === 'success') {
  console.log(result.result)
}

.stream()

Créez une instance d'exécution du workflow avec .createRun(), puis appelez .stream() en fournissant un inputData conforme à l'inputSchema du workflow. Parcourez fullStream pour suivre la progression, puis attendez result afin d'obtenir le résultat final du workflow.

const run = await testWorkflow.createRun()

const stream = run.stream({
  inputData: {
    message: 'Hello world',
  },
})

for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get the final result (same type as run.start())
const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

Type du résultat d'un workflow

run.start() et stream.result renvoient tous deux une union discriminée fondée sur la propriété status, qui peut valoir success, failed, suspended, tripwire ou paused. Quel que soit l'état, vous pouvez toujours accéder en toute sécurité à result.status, result.input, result.steps et, si elle est présente, à result.state.

En outre, les propriétés disponibles diffèrent selon l'état :

État	Propriétés spécifiques	Description
success	result	Données de sortie du workflow
failed	error	Erreur à l'origine de l'échec
tripwire	tripwire	Contient reason, retry?, metadata?, processorId?
suspended	suspendPayload, suspended	Données de suspension et tableau des chemins d'étapes suspendues
paused	(aucune)	Seules les propriétés communes sont disponibles

Pour accéder aux propriétés propres à un état, vérifiez d'abord status :

const result = await run.start({ inputData: { message: 'Hello world' } })

if (result.status === 'success') {
  console.log(result.result) // Only available when status is "success"
} else if (result.status === 'failed') {
  console.log(result.error.message)
} else if (result.status === 'suspended') {
  console.log(result.suspendPayload)
}

Sortie d'un workflow

Voici un exemple de résultat de workflow réussi, avec les propriétés input, steps et result :

{
  "status": "success",
  "steps": {
    "step-1": {
      "status": "success",
      "payload": {
        "message": "Hello world"
      },
      "output": {
        "formatted": "HELLO WORLD"
      }
    },
    "step-2": {
      "status": "success",
      "payload": {
        "formatted": "HELLO WORLD"
      },
      "output": {
        "emphasized": "HELLO WORLD!!!"
      }
    }
  },
  "input": {
    "message": "Hello world"
  },
  "result": {
    "emphasized": "HELLO WORLD!!!"
  }
}

Streaming

Pour une utilisation générique de l'API writer, consultez Streaming.

Inspecter les charges utiles du flux d'un workflow

Les événements écrits dans le flux sont inclus dans les chunks émis. Inspectez ces chunks pour accéder aux champs personnalisés, comme les types d'événements, les valeurs intermédiaires ou les données propres à une étape.

const testWorkflow = mastra.getWorkflow('testWorkflow')

const run = await testWorkflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream) {
  console.log(chunk)
}

if (result!.status === 'suspended') {
  // if the workflow is suspended, we can resume it with the resumeStream method
  const resumedStream = await run.resumeStream({
    resumeData: { value: 'resume data' },
  })

  for await (const chunk of resumedStream) {
    console.log(chunk)
  }
}

Reprendre le flux interrompu d'un workflow

Si le flux d'un workflow est fermé ou interrompu pour une raison quelconque, vous pouvez le reprendre avec la méthode resumeStream. Celle-ci renvoie un nouveau ReadableStream permettant d'observer les événements du workflow.

const newStream = await run.resumeStream()

for await (const chunk of newStream) {
  console.log(chunk)
}

Workflow utilisant un agent

Redirigez le textStream d'un agent vers le writer de l'étape du workflow. La sortie partielle est ainsi diffusée en streaming, et Mastra agrège automatiquement l'utilisation de l'agent dans l'exécution du workflow.

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

export const testStep = createStep({
  execute: async ({ inputData, mastra, writer }) => {
    const { city } = inputData

    const testAgent = mastra?.getAgent('testAgent')
    const stream = await testAgent?.stream(`What is the weather in ${city}?`)

    await stream!.textStream.pipeTo(writer!)

    return {
      value: await stream!.text,
    }
  },
})

Redémarrer les exécutions actives d'un workflow

Lorsqu'une exécution de workflow perd sa connexion au serveur, elle peut être redémarrée à partir de la dernière étape active. Cette fonctionnalité est utile pour les workflows de longue durée susceptibles de continuer à s'exécuter au moment où la connexion au serveur est perdue. Le redémarrage reprend l'exécution à la dernière étape active, puis le workflow poursuit son déroulement à partir de là.

Redémarrer toutes les exécutions actives d'un workflow avec restartAllActiveWorkflowRuns()

Utilisez restartAllActiveWorkflowRuns() pour redémarrer toutes les exécutions actives d'un workflow, sans avoir à parcourir et relancer manuellement chacune d'elles.

workflow.restartAllActiveWorkflowRuns()

Redémarrer une exécution active de workflow avec restart()

Utilisez restart() pour redémarrer une exécution active du workflow à partir de sa dernière étape active. L'exécution reprend à cette étape, puis le workflow se poursuit à partir de là.

const run = await workflow.createRun()

const result = await run.start({ inputData: { value: 'initial data' } })

const restartedResult = await run.restart()

Identifier les exécutions actives d'un workflow

Une exécution active de workflow possède un status égal à running ou waiting. Vous pouvez vérifier le status du workflow pour confirmer qu'il est actif et utiliser active afin d'identifier l'exécution active.

const activeRuns = await workflow.listActiveWorkflowRuns()
if (activeRuns.runs.length > 0) {
  console.log(activeRuns.runs)
}

remarque

Lorsque vous exécutez le serveur Mastra local, toutes les exécutions actives de workflows sont redémarrées automatiquement au lancement du serveur.

Utiliser RequestContext

Utilisez RequestContext pour accéder aux valeurs propres à une requête. Vous pouvez ainsi ajuster le comportement de manière conditionnelle en fonction du contexte de cette requête.

src/mastra/workflows/test-workflow.ts

export type UserTier = {
  'user-tier': 'enterprise' | 'pro'
}

const step1 = createStep({
  execute: async ({ requestContext }) => {
    const userTier = requestContext.get('user-tier') as UserTier['user-tier']

    const maxResults = userTier === 'enterprise' ? 1000 : 50

    return { maxResults }
  },
})

Consultez Contexte de requête pour en savoir plus.

astuce

Pour valider le schéma du contexte de requête avec la sûreté des types, consultez Validation du schéma.

Ressources associées

Pour découvrir les workflows plus en détail, consultez notre guide des workflows, qui présente les concepts fondamentaux à travers un exemple pratique.

• État du workflow
• Flux de contrôle
• Suspendre et reprendre
• Gestion des erreurs
• Workers : exécutez les workflows dans des processus d'arrière-plan dédiés
• 📹 Atelier sur les workflows agentiques avec Mastra