Exécuter et tester un workflow

Déclencher un workflow

Une fois votre worker en cours d’exécution, vous pouvez déclencher un workflow de deux manières : via le SDK Python ou via un appel cURL à l’API REST.

Prérequis

Avant de déclencher une exécution :

1. Votre worker doit être en cours d’exécution (uv run python worker.py)
2. Votre clé API doit être configurée (MISTRAL_API_KEY)
3. Le workflow doit être enregistré auprès du worker

Déclenchement via le SDK Python

Créez un script séparé pour déclencher l’exécution :

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())

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.

Déclenchement via cURL

Pour les tests rapides ou l’intégration avec d’autres systèmes, utilisez l’API REST :

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

Exemple complet : de bout en bout

Reprenons l’exemple du pipeline d’analyse de document. Voici les trois fichiers nécessaires :

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())

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

Vérification dans les logs du worker

Le terminal du worker affiche les activités exécutées en temps réel :

[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

Récupérer le statut d’une exécution

Si vous avez l’ID d’exécution, vous pouvez interroger son statut :

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

Via cURL :

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

Gestion des erreurs

Si une activité échoue et que les retries sont épuisés, le workflow passe en statut "failed" :

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

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

Vous pouvez ensuite analyser les traces OpenTelemetry pour identifier l’activité qui a échoué (leçon 14).

Tester localement

Pour le développement, créez un script de test qui combine worker et déclenchement :

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())

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