API Workflows

L’API Workflows fournit des méthodes permettant d’interagir avec des Workflows automatisés et de les exécuter dans Mastra.

Obtenir tous les Workflows

Récupérez la liste de tous les Workflows disponibles :

const workflows = await mastraClient.listWorkflows()

Obtenir le nombre d’exécutions des Workflows

Récupérez en une seule requête, pour chaque Workflow, le nombre d’exécutions running et suspended. Les nombres sont calculés sur le serveur et indexés par la clé de registre du Workflow, c’est-à-dire la clé utilisée lors de son enregistrement dans la configuration Mastra, qui peut différer de son propre id :

const runCounts = await mastraClient.listWorkflowRunCounts()
// { "cityWorkflow": { running: 2, suspended: 1 }, ... }

Renvoie : Record<string, { running: number; suspended: number }>

Le serveur peut mettre en cache ces nombres pendant quelques secondes entre les requêtes. Les serveurs antérieurs à ce point de terminaison répondent avec 404 Not Found ; gérez cette erreur lorsque le client peut communiquer avec des déploiements plus anciens.

Utiliser un Workflow précis

Obtenez l’instance d’un Workflow précis à partir de son ID :

src/mastra/workflows/test-workflow.ts

export const testWorkflow = createWorkflow({
  id: 'city-workflow',
})

const workflow = mastraClient.getWorkflow('city-workflow')

Méthodes des Workflows

details()

Récupérez des informations détaillées sur un Workflow :

const details = await workflow.details()

createRun()

Créez une nouvelle instance d’exécution de Workflow :

const run = await workflow.createRun()

// Or with an existing runId
const run = await workflow.createRun({ runId: 'existing-run-id' })

// Or with a resourceId to associate the run with a specific resource
const run = await workflow.createRun({
  runId: 'my-run-id',
  resourceId: 'user-123',
})

Le paramètre resourceId associe l’exécution du Workflow à une ressource donnée (par exemple, un ID utilisateur ou un ID de locataire). Cette valeur est conservée avec l’exécution et peut servir ultérieurement à filtrer et à interroger les exécutions.

startAsync()

Démarrez l’exécution d’un Workflow et attendez qu’elle se termine ; le résultat complet est renvoyé comme sortie du Workflow.

const run = await workflow.createRun()

const result = await run.startAsync({
  inputData: {
    city: 'New York',
  },
})

Vous pouvez également transmettre initialState pour définir les valeurs initiales de l’état du Workflow :

const result = await run.startAsync({
  inputData: {
    city: 'New York',
  },
  initialState: {
    count: 0,
    items: [],
  },
})

L’objet initialState doit correspondre à la structure définie dans le stateSchema du Workflow. Pour en savoir plus, consultez l’état des Workflows.

Pour associer une exécution à une ressource donnée, transmettez resourceId à createRun() :

const run = await workflow.createRun({ resourceId: 'user-123' })

const result = await run.startAsync({
  inputData: {
    city: 'New York',
  },
})

start()

Démarrez l’exécution d’un Workflow sans attendre qu’elle se termine (lancement sans attente). Renvoie immédiatement un message de réussite. Utilisez runById() sur l’instance du Workflow pour vérifier les résultats ultérieurement :

const run = await workflow.createRun()

await run.start({
  inputData: {
    city: 'New York',
  },
})

// Poll for results later
const result = await workflow.runById(run.runId)

Cette méthode est utile pour les Workflows de longue durée que vous souhaitez démarrer, puis vérifier ultérieurement.

resumeAsync()

Reprenez une étape suspendue du Workflow et attendez le résultat complet :

const run = await workflow.createRun({ runId: prevRunId })

const result = await run.resumeAsync({
  step: 'step-id',
  resumeData: { key: 'value' },
})

resume()

Reprenez une étape suspendue du Workflow sans attendre qu’elle se termine :

const run = await workflow.createRun({ runId: prevRunId })

await run.resume({
  step: 'step-id',
  resumeData: { key: 'value' },
})

Lorsqu’une étape .foreach() se suspend au cours de plusieurs itérations, transmettez forEachIndex (indexé à partir de zéro ; 0 cible la première itération) pour reprendre une seule itération à la fois. Les itérations que vous ne ciblez pas restent suspendues.

await run.resume({
  step: 'approve',
  resumeData: { ok: true },
  forEachIndex: 1, // resumes the second iteration
})

forEachIndex est également pris en charge par resumeAsync() et resumeStream().

cancel()

Annulez un Workflow en cours d’exécution :

const run = await workflow.createRun({ runId: existingRunId })

const result = await run.cancel()
// Returns: { message: 'Workflow run canceled' }

Cette méthode arrête toutes les étapes en cours et empêche l’exécution des étapes suivantes. Les étapes qui vérifient le paramètre abortSignal peuvent réagir à l’annulation en libérant les ressources (délais d’expiration, requêtes réseau, etc.).

Consultez la référence de Run.cancel() pour en savoir plus sur le fonctionnement de l’annulation et la façon d’écrire des étapes qui y réagissent.

stream()

Diffusez l’exécution du Workflow afin de recevoir des mises à jour en temps réel :

const run = await workflow.createRun()

const stream = await run.stream({
  inputData: {
    city: 'New York',
  },
})

for await (const chunk of stream) {
  console.log(JSON.stringify(chunk, null, 2))
}

runById()

Obtenez le résultat d’une exécution de Workflow :

const result = await workflow.runById(runId)

// Or with options for performance optimization:
const result = await workflow.runById(runId, {
  fields: ['status', 'result'], // Only fetch specific fields
  withNestedWorkflows: false, // Skip expensive nested workflow data
  requestContext: { userId: 'user-123' }, // Optional request context
})

Format du résultat d’exécution

Le résultat d’une exécution de Workflow contient les éléments suivants :

runId:

string

Identifiant unique de cette instance d’exécution de Workflow

eventTimestamp:

Date

Horodatage de l’événement

payload:

object

Contient currentStep (id, status, output, payload) et workflowState (status, enregistrement des étapes)

Workflows dynamiques

beta

Les Workflows dynamiques sont en version bêta. Tant que l’API n’est pas stable, des changements incompatibles peuvent intervenir sans changement de version majeure.

Les Workflows dynamiques sont des définitions de Workflow exprimées en JSON. Le serveur conserve chaque définition et l’enregistre en tant que Workflow exécutable. Pour connaître le format des définitions, consultez les Workflows dynamiques.

listDynamicWorkflows()

Répertoriez les définitions de Workflows dynamiques, avec un filtrage facultatif par status ('active' | 'archived') et authorId :

const { definitions, total } = await mastraClient.listDynamicWorkflows({
  status: 'active',
})

upsertDynamicWorkflow()

Créez ou remplacez une définition de Workflow dynamique. Le serveur valide la définition, la conserve et l’enregistre à chaud pour l’exécution :

const stored = await mastraClient.upsertDynamicWorkflow({
  id: 'greeting-workflow',
  description: 'Returns a greeting for the supplied name',
  inputSchema: {
    type: 'object',
    properties: { name: { type: 'string' } },
    required: ['name'],
  },
  outputSchema: {
    type: 'object',
    properties: { message: { type: 'string' } },
    required: ['message'],
  },
  graph: [
    {
      type: 'mapping',
      id: 'create-greeting',
      mapConfig: JSON.stringify({
        message: { template: 'Hello, ${initData.name}!' },
      }),
    },
  ],
})

Lorsque la définition racine imbrique des Workflows auxiliaires qui n’existent pas encore, transmettez-les dans la même requête via dependencies. Le serveur valide et enregistre le bundle comme une unité, puis renvoie les ID des auxiliaires dans dependencyIds :

const stored = await mastraClient.upsertDynamicWorkflow({
  id: 'root-workflow',
  // ...schemas and graph referencing 'helper-workflow'...
  dependencies: [helperDefinition],
})

console.log(stored.dependencyIds) // ['helper-workflow']

getDynamicWorkflow()

Obtenez une instance de Workflow dynamique pour gérer sa définition. Pour exécuter un Workflow dynamique, utilisez getWorkflow(id).createRun() comme pour n’importe quel autre Workflow :

const dynamicWorkflow = mastraClient.getDynamicWorkflow('greeting-workflow')

dynamicWorkflow.details()

Récupérez la définition conservée, notamment les schémas, le graphe, le statut et les horodatages :

const definition = await dynamicWorkflow.details()

dynamicWorkflow.delete()

Supprimez la définition stockée et désenregistrez le Workflow actif :

await dynamicWorkflow.delete()

Exécuter un Workflow dynamique

Une fois enregistré, un Workflow dynamique s’exécute par l’intermédiaire de l’API ordinaire des Workflows :

const workflow = mastraClient.getWorkflow('greeting-workflow')
const run = await workflow.createRun()
const result = await run.startAsync({ inputData: { name: 'Ada' } })

Planifications

Les planifications sont déclarées dans le code au moyen du champ schedule de createWorkflow. Le SDK client expose des méthodes de lecture et d’exploitation permettant de gérer les planifications des Workflows pendant l’exécution. Consultez les Workflows planifiés.

createSchedule()

Créez une planification de Workflow en transmettant workflowId.

const schedule = await mastraClient.createSchedule({
  workflowId: 'daily-report',
  cron: '0 9 * * *',
  inputData: { reportType: 'summary' },
})

listSchedules()

Répertoriez les planifications de Workflows, avec un filtrage facultatif par ID de Workflow ou par statut.

const schedules = await mastraClient.listSchedules({
  workflowId: 'daily-report',
  status: 'active',
})

getSchedule()

Récupérez une planification de Workflow à partir de son ID.

const schedule = await mastraClient.getSchedule('daily-report')

updateSchedule()

Mettez à jour une planification de Workflow.

const updated = await mastraClient.updateSchedule('daily-report', {
  cron: '0 10 * * *',
  inputData: { reportType: 'summary' },
})

deleteSchedule()

Supprimez une planification de Workflow.

await mastraClient.deleteSchedule('daily-report')

runSchedule()

Déclenchez immédiatement une planification de Workflow une seule fois, sans modifier sa cadence cron.

const run = await mastraClient.runSchedule('daily-report')

pauseSchedule()

Mettez une planification en pause afin que le planificateur cesse de la déclencher. Renvoie la planification mise à jour.

await mastraClient.pauseSchedule('daily-report')

resumeSchedule()

Reprenez une planification en pause. La prochaine heure de déclenchement est recalculée à partir de l’instant présent ; une planification longtemps interrompue ne déclenche donc pas les occurrences en attente. Renvoie la planification mise à jour.

await mastraClient.resumeSchedule('daily-report')

listScheduleTriggers()

Répertoriez l’historique des déclenchements d’une planification de Workflow, y compris le résumé de l’exécution associée à chaque déclenchement.

const { triggers } = await mastraClient.listScheduleTriggers('daily-report', {
  limit: 50,
})