Aller au contenu principal

Exécuter et tester un workflow

Mis à jour le 29 juillet 2026

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 import Mistral

async def main():
    client = Mistral()

    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="salutation_workflow",
        input={"nom": "Marie"}
    )

    print(f"ID d'exécution : {execution.id}")
    print(f"Statut : {execution.status}")
    print(f"Résultat : {execution.output}")

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

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.

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"}}'

La réponse contient l’ID d’exécution, le statut et le résultat :

{
  "id": "exec_abc123def456",
  "status": "completed",
  "output": {
    "message": "Bonjour, Marie !"
  }
}

Conservez cet id : c’est la clé qui vous permettra plus tard d’interroger l’exécution ou de retrouver ses traces.

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 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 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"Statut : {execution.status}")
    if execution.status == "completed":
        print(f"Résumé : {execution.output['resume']}")
    else:
        print(f"Erreur : {execution.error}")

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.get_workflow_execution(execution_id="exec_abc123")
print(f"Statut : {execution.status}")
print(f"Résultat : {execution.output}")

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 le workflow passe en statut "failed", et votre code appelant doit traiter ce cas au lieu de supposer un output toujours présent :

execution = await client.workflows.execute_workflow_async(
    workflow_identifier="mon_workflow",
    input={"données": "test"}
)

if execution.status == "failed":
    print(f"Le workflow a échoué : {execution.error}")

Le message d’erreur vous donne la cause immédiate ; 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 import Mistral
    client = Mistral()
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="mon_workflow",
        input={"test": True}
    )
    print(f"Résultat : {execution.output}")

    # 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
  • Le SDK attend la fin de l’exécution et retourne le résultat directement
  • Utilisez get_workflow_execution() pour vérifier le statut d’une exécution en cours
  • Les workflows échoués passent en statut "failed" avec un message d’erreur
  • Pour le développement, combinez worker et trigger dans un même script avec asyncio
← Lecon precedenteCréer un workerLecon suivante →La règle de déterminisme