Workflow Temporal

Temporal est une plateforme d’exécution durable permettant d’orchestrer des Workflows longs et tolérants aux pannes. Le package @mastra/temporal vous permet de créer des Workflows avec l’API Mastra standard et de les exécuter sur un cluster Temporal.

attention

@mastra/temporal est expérimental et n’est pas prêt pour une utilisation en production. Son API peut évoluer d’une version à l’autre. Consultez le README du package pour connaître son état actuel.

Fonctionnement de Temporal avec Mastra

Les Workflows Mastra créés avec createWorkflow() et createStep() sont associés au modèle de Workflows et d’activités de Temporal. Le MastraPlugin destiné au worker Temporal compile votre fichier d’entrée Mastra au moment du regroupement :

• Chaque gestionnaire createStep() est extrait dans une activité Temporal.
• Chaque createWorkflow() est réécrit en Workflow Temporal qui invoque ces activités.
• Le plugin enregistre automatiquement les activités et Workflows générés auprès du worker.

Lorsqu’une exécution démarre avec mastra.getWorkflow(...).createRun().start(...), le client Mastra cède le contrôle à Temporal. Temporal pilote alors l’exécution durable, les nouvelles tentatives et la persistance de l’état sur le worker.

Configuration

Installez les packages requis :

npm

npm install @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

pnpm

pnpm add @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

Yarn

yarn add @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

Bun

bun add @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

Vous avez également besoin d’accéder à un cluster Temporal. Pour le développement local, vous pouvez en exécuter un avec Docker ; consultez Exécuter localement.

Créer un Workflow basé sur Temporal

Ce guide détaille la création d’un Workflow avec Temporal et Mastra, à l’aide d’une application de compteur qui incrémente une valeur.

Initialisation de Temporal

Initialisez l’intégration Temporal pour obtenir des assistants de Workflow compatibles avec Mastra. Les fonctions createWorkflow() et createStep() sont liées à un client Temporal et à une file de tâches.

src/mastra/temporal/index.ts

import { init } from '@mastra/temporal'
import { Client, Connection } from '@temporalio/client'
import { loadClientConnectConfig } from '@temporalio/envconfig'

const config = loadClientConnectConfig()
const connection = await Connection.connect(config.connectionOptions)
const client = new Client({ connection })

export const { createWorkflow, createStep } = init({
  client,
  taskQueue: 'mastra',
})

loadClientConnectConfig() lit les variables d’environnement Temporal standard telles que TEMPORAL_ADDRESS, TEMPORAL_NAMESPACE et les paramètres mTLS. Consultez la documentation envconfig de Temporal pour obtenir la liste complète.

Créer des étapes

Définissez les étapes individuelles qui composent votre Workflow. Chaque étape devient une activité Temporal.

src/mastra/workflows/index.ts

import { z } from 'zod'
import { createWorkflow, createStep } from '../temporal'

const incrementStep = createStep({
  id: 'increment',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    return { value: inputData.value + 1 }
  },
})

Créer le Workflow

Assemblez les étapes dans un Workflow. L’id du Workflow doit être un littéral de chaîne statique afin que le transformateur au moment de la création puisse en déduire le nom d’export Temporal.

src/mastra/workflows/index.ts

const workflow = createWorkflow({
  id: 'increment-workflow',
  steps: [incrementStep],
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
}).then(incrementStep)

workflow.commit()

export { workflow as incrementWorkflow }

Configurer l’instance Mastra

Enregistrez le Workflow auprès de Mastra. L’exécution est pilotée par le worker Temporal.

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { PinoLogger } from '@mastra/loggers'
import { incrementWorkflow } from './workflows'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})

Exécuter le worker

Le worker est un processus Node.js de longue durée qui interroge une file de tâches Temporal. Installez MastraPlugin et faites pointer son option src vers le fichier d’entrée Mastra qui enregistre vos Workflows.

src/mastra/worker.ts

import { MastraPlugin } from '@mastra/temporal/worker'
import { NativeConnection, Worker } from '@temporalio/worker'

const connection = await NativeConnection.connect({
  address: 'localhost:7233',
})

const mastraPlugin = new MastraPlugin()

await mastraPlugin.prebuild({
  entryFile: import.meta.resolve('./index.ts'),
})

const worker = await Worker.create({
  connection,
  namespace: 'default',
  taskQueue: 'mastra',
  plugins: [mastraPlugin],
})

await worker.run()

MastraPlugin réécrit le fichier d’entrée en bundle ne contenant que le Workflow et câble les gestionnaires d’étapes comme activités Temporal. Vous n’avez pas besoin de transmettre manuellement activities ni workflowsPath à Worker.create().

Exécuter des Workflows

Exécuter localement

1. Démarrez un serveur Temporal local. L’option la plus simple est l’image Docker temporalio/auto-setup :

   docker run --rm -p 7233:7233 -p 8080:8080 temporalio/auto-setup:latest

2. Ouvrez l’interface Temporal à l’adresse http://localhost:8080 pour examiner les espaces de noms, les Workflows et les activités.

3. Dans un nouveau terminal, démarrez le worker en exécutant :

   npx tsx src/mastra/worker.ts

4. Déclenchez une exécution de Workflow depuis un script ou tout processus qui importe votre instance Mastra :

   scripts/run.ts

   import { mastra } from '../src/mastra'
   
   const run = await mastra.getWorkflow('incrementWorkflow').createRun()
   const result = await run.start({ inputData: { value: 5 } })
   
   console.log(result)

5. Surveillez l’exécution dans l’interface Temporal, sous Workflows, pour voir la progression des activités étape par étape et l’historique des nouvelles tentatives.

Exécuter en production

En production, utilisez Temporal Cloud ou un cluster Temporal auto-hébergé. Configurez les connexions du client et du worker avec les variables d’environnement lues par @temporalio/envconfig :

.env

TEMPORAL_ADDRESS=your-namespace.tmprl.cloud:7233
TEMPORAL_NAMESPACE=your-namespace
TEMPORAL_API_KEY=your-api-key

Consultez la documentation de connexion à Temporal Cloud pour les options mTLS et de clé API.

attention

Le worker Temporal doit s’exécuter comme processus de longue durée. Ne le déployez pas sur des plateformes serverless dont les limites d’exécution sont courtes, telles que les fonctions AWS Lambda ou Vercel. Utilisez un conteneur, une VM ou une plateforme adaptée aux workers, telle que Fly.io, Railway ou Kubernetes.

Options de configuration

taskQueue

Obligatoire. Identifie la file de tâches Temporal que votre worker interroge. La même valeur doit être transmise à init(), utilisé par le client pour démarrer les exécutions, et à Worker.create(), utilisé par le worker pour les recevoir.

startToCloseTimeout

Facultatif. Définit la durée maximale pendant laquelle une activité unique, c’est-à-dire une étape, est autorisée à s’exécuter avant que Temporal ne l’annule et applique la stratégie de nouvelle tentative. La valeur par défaut est 1 minute.

src/mastra/temporal/index.ts

export const { createWorkflow, createStep } = init({
  client,
  taskQueue: 'mastra',
  startToCloseTimeout: '5 minutes',
})

Contraintes et remarques

• Les identifiants de Workflow doivent être des littéraux de chaîne statiques. Le transformateur au moment de la création lit la valeur littérale pour en déduire les noms d’export de Workflows Temporal.
• Les activités sont générées automatiquement à partir des gestionnaires createStep(). Ne les transmettez pas à Worker.create({ activities }).

Ressources associées

• Exécuteurs de Workflows
• Présentation des Workflows
• Documentation Temporal