Exécuter et tester un workflow

Leçon corrigée le 3 novembre 2026 d’après le SDK mistralai 3.7.0 : le suivi d’une exécution passe par client.workflows.executions.get_workflow_execution_async, et le résultat se lit dans result. Complément du même jour : les autres exemples lisaient encore execution.id, execution.output et execution.error, absents du SDK, et comparaient le statut à "completed" ou "failed" en minuscules ; ils utilisent désormais execution_id, result, wait_for_result=True et les statuts en majuscules. Source : client-python.

Déclencher un workflow

Vos activités sont écrites, votre workflow est défini, votre worker tourne : il reste à lancer une exécution. Deux voies s’offrent à vous, le SDK Python et un appel cURL à l’API REST. La première convient à vos scripts et à vos services Python, la seconde à un test rapide en ligne de commande ou à un déclenchement depuis un système qui ne parle pas Python — un webhook, un ordonnanceur, un backend écrit dans un autre langage.

Trois conditions doivent être réunies avant de déclencher quoi que ce soit. Votre worker doit être en cours d’exécution (uv run python worker.py), votre clé API doit être configurée dans MISTRAL_API_KEY, et le workflow doit être enregistré auprès de ce worker, c’est-à-dire présent dans la liste passée à run_worker().

Déclenchement via le SDK Python

Écrivez le déclencheur dans un script séparé du worker, puisque le processus du worker ne rend jamais la main :

import asyncio
from mistralai.client import Mistral

async def main():
    client = Mistral()

    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="salutation_workflow",
        input={"nom": "Marie"},
        wait_for_result=True,  # attendre la fin et recevoir le résultat
    )

    print(f"ID d'exécution : {execution.execution_id}")
    print(f"Résultat : {execution.result}")

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

Le workflow_identifier reprend exactement le name déclaré dans @workflows.workflow.define(), et les clés du dictionnaire input correspondent aux paramètres de votre méthode run. Lancez ce script dans un autre terminal, le premier restant occupé par le worker :

uv run python trigger.py

Avec wait_for_result=True, le SDK attend la fin de l’exécution et retourne le résultat directement, ce qui rend le test très lisible pour un workflow court. Sans ce paramètre, l’appel rend la main tout de suite avec l’identifiant de l’exécution, que vous suivez ensuite (voir plus bas).

Déclenchement via cURL

Pour les tests rapides ou l’intégration avec d’autres systèmes, l’API REST expose le même déclenchement :

curl -X POST https://api.mistral.ai/v1/workflows/salutation_workflow/execute \
  -H "Authorization: Bearer $MISTRAL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"input": {"nom": "Marie"}, "wait_for_result": true}'

Avec wait_for_result, la réponse contient le nom du workflow, l’identifiant de l’exécution et son résultat ; ce sont les champs du modèle WorkflowExecutionSyncResponse du SDK :

{
  "workflow_name": "salutation_workflow",
  "execution_id": "…",
  "result": {
    "message": "Bonjour, Marie !"
  }
}

Conservez cet execution_id : c’est la clé qui vous permettra plus tard d’interroger l’exécution ou de retrouver ses traces. Sans wait_for_result, la réponse décrit l’exécution en cours, avec son status (RUNNING au départ).

Exemple complet : de bout en bout

Reprenons l’exemple du pipeline d’analyse de document, cette fois entièrement gréé. Le premier fichier réunit les définitions et le démarrage du worker ; notez la troncature à 5000 caractères, qui protège à la fois la limite de taille des activités et votre budget de tokens.

Fichier workflow.py (worker + définitions)

import asyncio
import httpx
import mistralai.workflows as workflows
from mistralai.client import Mistral

@workflows.activity()
async def extraire_texte(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    return {"texte": response.text[:5000]}  # Limiter à 5000 caractères

@workflows.activity()
async def resumer_texte(texte: str) -> dict:
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Résumez en 3 phrases."},
            {"role": "user", "content": texte}
        ]
    )
    return {"resume": response.choices[0].message.content}

@workflows.workflow.define(name="resume_page_web")
class ResumePageWeb:
    @workflows.workflow.entrypoint
    async def run(self, url: str) -> dict:
        extraction = await extraire_texte(url)
        resume = await resumer_texte(extraction["texte"])
        return {
            "url_source": url,
            "resume": resume["resume"]
        }

async def main() -> None:
    await workflows.run_worker([ResumePageWeb])

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

Le second fichier déclenche et affiche le résultat en distinguant explicitement le cas d’échec, réflexe à prendre dès le premier script :

Fichier trigger.py (déclencheur)

import asyncio
from mistralai.client import Mistral

async def main():
    client = Mistral()

    print("Lancement du workflow...")
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="resume_page_web",
        input={"url": "https://example.com/article"},
    )
    print(f"Exécution lancée : {execution.execution_id}")

    # Suivre l'exécution jusqu'à son terme
    while True:
        suivi = await client.workflows.executions.get_workflow_execution_async(
            execution_id=execution.execution_id
        )
        if suivi.status not in ("RUNNING", "RETRYING_AFTER_ERROR"):
            break
        await asyncio.sleep(2)

    if suivi.status == "COMPLETED":
        print(f"Résumé : {suivi.result['resume']}")
    else:
        print(f"Exécution terminée avec le statut {suivi.status}")

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

Exécution

Terminal 1 — Démarrez le worker :

uv run python workflow.py

Terminal 2 — Déclenchez l’exécution :

uv run python trigger.py

Vérifier les résultats

Le terminal du worker constitue votre premier poste d’observation : il affiche les activités exécutées en temps réel, avec leur durée, ce qui vous montre immédiatement quelle étape consomme le temps de l’exécution.

[INFO] Executing activity: extraire_texte
[INFO] Activity completed: extraire_texte (1.2s)
[INFO] Executing activity: resumer_texte
[INFO] Activity completed: resumer_texte (3.4s)
[INFO] Workflow completed: resume_page_web

Pour les workflows longs, vous ne resterez évidemment pas devant le terminal. Dès lors que vous disposez de l’ID d’exécution, vous pouvez interroger son statut à tout moment depuis un autre processus :

execution = await client.workflows.executions.get_workflow_execution_async(
    execution_id="exec_abc123"
)
print(f"Statut : {execution.status}")
print(f"Résultat : {execution.result}")

La même opération est disponible en cURL, pratique dans un script de supervision :

curl https://api.mistral.ai/v1/workflows/executions/exec_abc123 \
  -H "Authorization: Bearer $MISTRAL_API_KEY"

Gestion des erreurs

Un échec ne survient pas à la première exception venue : la plateforme retente d’abord l’activité selon sa politique de retry. C’est seulement une fois les retries épuisés que l’exécution passe en statut FAILED (les statuts du SDK sont en majuscules : RUNNING, COMPLETED, FAILED, CANCELED, TERMINATED, TIMED_OUT…), et votre code appelant doit traiter ce cas au lieu de supposer un result toujours présent :

execution = await client.workflows.executions.get_workflow_execution_async(
    execution_id="exec_abc123"
)

if execution.status == "FAILED":
    print(f"L'exécution {execution.execution_id} a échoué")

La réponse ne contient pas de message d’erreur : la cause se lit dans l’historique et les journaux de l’exécution (get_workflow_execution_history_async, get_workflow_execution_logs_async), et pour identifier précisément l’activité fautive et le déroulé qui y a mené, vous analyserez les traces OpenTelemetry, sujet de la leçon 14.

Tester localement

Faire la navette entre deux terminaux devient vite pénible en phase de mise au point. Pour le développement, un script unique peut lancer le worker en arrière-plan, déclencher le workflow, puis tout arrêter :

import asyncio
import mistralai.workflows as workflows

# ... définitions des activités et workflows ...

async def main():
    # Lancer le worker en arrière-plan
    worker_task = asyncio.create_task(
        workflows.run_worker([MonWorkflow])
    )

    # Attendre que le worker soit prêt
    await asyncio.sleep(2)

    # Déclencher le workflow
    from mistralai.client import Mistral
    client = Mistral()
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="mon_workflow",
        input={"test": True},
        wait_for_result=True,
    )
    print(f"Résultat : {execution.result}")

    # Arrêter le worker
    worker_task.cancel()

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

Le asyncio.sleep(2) laisse au worker le temps d’établir sa connexion avant le déclenchement, et le worker_task.cancel() final évite que le script reste bloqué indéfiniment. Réservez ce montage au développement : en production, le worker et les déclencheurs restent des processus distincts, avec des cycles de vie indépendants.

Points clés à retenir

• Deux méthodes de déclenchement : SDK Python (execute_workflow_async) et API REST (cURL)
• Le worker doit être en cours d’exécution pour que le workflow s’exécute
• Avec wait_for_result=True, le SDK attend la fin de l’exécution et retourne le résultat (result) ; sinon il rend l’execution_id à suivre
• Utilisez client.workflows.executions.get_workflow_execution_async() pour suivre le statut d’une exécution
• Les statuts sont en majuscules (RUNNING, COMPLETED, FAILED…) ; la cause d’un échec se lit dans l’historique, les journaux et les traces de l’exécution
• Pour le développement, combinez worker et trigger dans un même script avec asyncio