Créer un worker

Le rôle du worker

Le worker est le processus Python qui exécute concrètement vos workflows et activités. Il se connecte à la plateforme Mistral, écoute les tâches en attente, et les exécute localement sur votre machine ou votre serveur. C’est un détail d’architecture qui surprend souvent au premier abord : votre code ne part jamais chez Mistral, seul l’état de l’orchestration y réside.

La conséquence est directe et explique la majorité des « rien ne se passe » rencontrés en début de projet : sans worker en cours d’exécution, vos workflows ne peuvent pas s’exécuter. Vous pouvez déclencher une exécution, elle restera simplement en attente qu’un worker capable de la traiter se présente. Le worker est le lien entre la plateforme d’orchestration, côté Mistral, et votre code, côté local.

Créer un worker minimal

Le code d’un worker fonctionnel tient en quelques lignes. L’exemple ci-dessous regroupe volontairement l’activité, le workflow et le démarrage dans un seul fichier pour que vous puissiez le lancer immédiatement :

import asyncio
import mistralai.workflows as workflows

# Importez vos workflows et activités
@workflows.activity()
async def dire_bonjour(nom: str) -> dict:
    return {"message": f"Bonjour, {nom} !"}

@workflows.workflow.define(name="salutation_workflow")
class SalutationWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        resultat = await dire_bonjour(nom)
        return resultat

# Démarrer le worker
async def main() -> None:
    await workflows.run_worker([SalutationWorkflow])

if __name__ == "__main__":
    asyncio.run(main())

La fonction clé est workflows.run_worker(). Elle prend en paramètre la liste des classes de workflows que ce worker peut exécuter — et uniquement des workflows : les activités sont découvertes à travers eux, vous n’avez pas à les énumérer.

Ce que fait `run_worker()`

Quand vous appelez workflows.run_worker([MonWorkflow]), cinq choses s’enchaînent. Le worker s’authentifie d’abord auprès de la plateforme Mistral avec votre clé API, puis il enregistre les workflows et activités déclarés — c’est cet enregistrement qui permet à la plateforme de savoir qu’un exécutant existe pour l’identifiant salutation_workflow. Il ouvre ensuite une connexion persistante et attend les tâches, en restant volontairement inactif tant qu’aucune exécution n’est déclenchée. Quand une exécution arrive, le worker exécute le code localement, et les résultats sont renvoyés à la plateforme pour persistance.

Le worker reste actif tant que le processus Python tourne, ce qui explique qu’il ne vous rende jamais la main dans le terminal. Utilisez Ctrl+C pour l’arrêter proprement.

Worker avec plusieurs workflows

Rien n’impose un worker par workflow. Le même processus peut en héberger plusieurs, y compris lorsqu’ils partagent des activités — ici extraire_texte et analyser_sentiment servent aux deux :

import asyncio
import mistralai.workflows as workflows

# --- Activités ---

@workflows.activity()
async def extraire_texte(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    return {"texte": response.text}

@workflows.activity()
async def analyser_sentiment(texte: str) -> dict:
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Analysez le sentiment : positif, négatif ou neutre."},
            {"role": "user", "content": texte}
        ]
    )
    return {"sentiment": response.choices[0].message.content}

@workflows.activity()
async def envoyer_alerte(message: str) -> dict:
    # Envoi vers Slack, email, etc.
    return {"status": "envoyé"}

# --- Workflows ---

@workflows.workflow.define(name="analyse_document")
class AnalyseDocument:
    @workflows.workflow.entrypoint
    async def run(self, url: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        return sentiment

@workflows.workflow.define(name="surveillance_mentions")
class SurveillanceMentions:
    @workflows.workflow.entrypoint
    async def run(self, url: str, seuil: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        if sentiment["sentiment"] == "négatif":
            await envoyer_alerte(f"Mention négative détectée : {url}")
        return sentiment

# --- Worker ---

async def main() -> None:
    await workflows.run_worker([
        AnalyseDocument,
        SurveillanceMentions,
    ])

if __name__ == "__main__":
    asyncio.run(main())

Structure de fichiers recommandée

Un fichier unique convient à l’apprentissage, beaucoup moins à un projet vivant. Pour la production, séparez le code en modules par nature :

mon-projet/
├── activities/
│   ├── __init__.py
│   ├── extraction.py      # extraire_texte()
│   ├── analyse.py          # analyser_sentiment()
│   └── notifications.py    # envoyer_alerte()
├── workflows/
│   ├── __init__.py
│   ├── analyse_document.py
│   └── surveillance.py
└── worker.py               # Point d'entrée unique

Le fichier worker.py ne contient alors plus que les imports et le démarrage, ce qui en fait un inventaire lisible de ce que ce processus sait exécuter :

import asyncio
import mistralai.workflows as workflows

from workflows.analyse_document import AnalyseDocument
from workflows.surveillance import SurveillanceMentions

async def main() -> None:
    await workflows.run_worker([
        AnalyseDocument,
        SurveillanceMentions,
    ])

if __name__ == "__main__":
    asyncio.run(main())

Lancer le worker

uv run python worker.py

Le worker affiche un message de connexion et reste actif en attente de tâches. Vous devez voir quelque chose comme :

Worker connected. Listening for tasks...

Tant que cette ligne n’apparaît pas, inutile de déclencher quoi que ce soit : vérifiez d’abord votre clé API et votre connexion réseau.

Worker en production

Un worker lancé à la main dans un terminal disparaît à la première déconnexion SSH. Confiez-le à un gestionnaire de processus — systemd, supervisor ou pm2 — qui le redémarrera automatiquement après un crash ou un reboot :

# Avec pm2 (si Node.js est disponible)
pm2 start "uv run python worker.py" --name workflow-worker

# Avec systemd (créer un fichier .service)

Ce redémarrage automatique se combine bien avec la durabilité : les workflows en cours reprennent d’eux-mêmes dès que le nouveau processus est connecté. Côté secrets, ne codez jamais la clé API en dur dans le fichier du worker, car elle finirait dans votre dépôt Git. Le SDK lit tout seul la variable d’environnement :

import os

# Le SDK lit automatiquement MISTRAL_API_KEY
# Pas besoin de le passer explicitement

Pensez enfin aux logs. Le worker émet ses messages via le module logging standard de Python, donc une simple configuration au démarrage vous donne la trace des activités exécutées, ce qui est précieux pour comprendre après coup ce qui tournait au moment d’un incident :

import logging
logging.basicConfig(level=logging.INFO)

Points clés à retenir

Le worker est le processus qui exécute vos workflows et activités localement
workflows.run_worker([...]) prend la liste des classes de workflows à exécuter
Un seul worker peut exécuter plusieurs workflows différents
Le worker reste actif et écoute les tâches tant que le processus tourne
En production, utilisez un gestionnaire de processus pour maintenir le worker actif
La clé API est lue automatiquement depuis MISTRAL_API_KEY