Run and test a workflow

Lesson updated November 3, 2026 based on mistralai SDK 3.7.0: tracking an execution now uses client.workflows.executions.get_workflow_execution_async, and the result is read from result. Same-day addendum: previous examples still referenced execution.id, execution.output, and execution.error, which are absent from the SDK, and compared status to lowercase "completed" or "failed"; they now use execution_id, result, wait_for_result=True, and uppercase statuses. Source: client-python.

Trigger a workflow

You’ve written your activities, defined your workflow, and started your worker. Now you need to launch an execution. You have two options: the Python SDK or a cURL call to the REST API. The first fits your Python scripts and services; the second works for quick command-line tests or triggers from non-Python systems—a webhook, a scheduler, or a backend written in another language.

Three conditions must be met before you trigger anything. Your worker must be running (uv run python worker.py), your API key must be set in MISTRAL_API_KEY, and the workflow must be registered with that worker, meaning it appears in the list passed to run_worker().

Triggering via the Python SDK

Write the trigger in a script separate from the worker, since the worker process never returns control:

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,  # wait for completion and receive the result
    )

    print(f"Execution ID: {execution.execution_id}")
    print(f"Result: {execution.result}")

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

The workflow_identifier matches the name declared in @workflows.workflow.define() exactly, and the keys in the input dictionary correspond to the parameters of your run method. Run this script in a separate terminal, leaving the first one occupied by the worker:

uv run python trigger.py

With wait_for_result=True, the SDK waits for the execution to finish and returns the result directly, making testing very readable for short workflows. Without this parameter, the call returns immediately with the execution identifier, which you then track (see below).

Triggering via cURL

For quick tests or integration with other systems, the REST API exposes the same trigger:

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

With wait_for_result, the response contains the workflow name, the execution identifier, and its result; these are the fields of the SDK’s WorkflowExecutionSyncResponse model:

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

Keep this execution_id: it’s the key you’ll use later to query the execution or retrieve its traces. Without wait_for_result, the response describes the running execution, including its status (RUNNING initially).

End-to-end example

Let’s revisit the document analysis pipeline example, this time fully wired up. The first file combines the definitions and worker startup; note the truncation to 5000 characters, which protects both the activity size limit and your token budget.

File workflow.py (worker + definitions)

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]}  # Limit to 5000 characters

@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": "Summarize in 3 sentences."},
            {"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())

The second file triggers the workflow and displays the result, explicitly handling the failure case—a reflex to adopt from your first script:

File trigger.py (trigger)

import asyncio
from mistralai.client import Mistral

async def main():
    client = Mistral()

    print("Launching workflow...")
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="resume_page_web",
        input={"url": "https://example.com/article"},
    )
    print(f"Execution launched: {execution.execution_id}")

    # Track execution until completion
    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"Summary: {suivi.result['resume']}")
    else:
        print(f"Execution finished with status {suivi.status}")

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

Execution

Terminal 1 — Start the worker:

uv run python workflow.py

Terminal 2 — Trigger the execution:

uv run python trigger.py

Check results

The worker terminal is your first observation point: it displays executed activities in real time, along with their duration, showing you immediately which step consumes the most time.

[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

For long workflows, you obviously won’t stay glued to the terminal. Once you have the execution ID, you can query its status at any time from another process:

execution = await client.workflows.executions.get_workflow_execution_async(
    execution_id="exec_abc123"
)
print(f"Status: {execution.status}")
print(f"Result: {execution.result}")

The same operation is available via cURL, useful in a monitoring script:

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

Error handling

A failure doesn’t occur on the first exception: the platform first retries the activity according to its retry policy. Only after retries are exhausted does the execution move to FAILED status (SDK statuses are uppercase: RUNNING, COMPLETED, FAILED, CANCELED, TERMINATED, TIMED_OUT, etc.), and your calling code must handle this case instead of assuming a result is always present:

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

if execution.status == "FAILED":
    print(f"Execution {execution.execution_id} failed")

The response doesn’t contain an error message: the cause appears in the execution history and logs (get_workflow_execution_history_async, get_workflow_execution_logs_async). To pinpoint the faulty activity and the sequence that led to it, you’ll analyze OpenTelemetry traces, covered in Lesson 14.

Local testing

Shuttling between two terminals quickly becomes tedious during development. For development, a single script can start the worker in the background, trigger the workflow, then stop everything:

import asyncio
import mistralai.workflows as workflows

# ... activity and workflow definitions ...

async def main():
    # Start worker in background
    worker_task = asyncio.create_task(
        workflows.run_worker([MonWorkflow])
    )

    # Wait for worker to be ready
    await asyncio.sleep(2)

    # Trigger 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"Result: {execution.result}")

    # Stop worker
    worker_task.cancel()

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

The asyncio.sleep(2) gives the worker time to establish its connection before triggering, and the final worker_task.cancel() prevents the script from hanging indefinitely. Reserve this setup for development: in production, the worker and triggers remain separate processes with independent lifecycles.

Key takeaways

• Two trigger methods: Python SDK (execute_workflow_async) and REST API (cURL)
• The worker must be running for the workflow to execute
• With wait_for_result=True, the SDK waits for completion and returns the result (result); otherwise it returns the execution_id to track
• Use client.workflows.executions.get_workflow_execution_async() to check an execution’s status
• Statuses are uppercase (RUNNING, COMPLETED, FAILED, etc.); failure causes appear in the execution history, logs, and traces
• For development, combine worker and trigger in a single script using asyncio