Create a workflow — Mistral Workflows: Durable Orchestration

The role of the workflow

The workflow is the orchestration layer of your application. It defines what must happen, in what order, and under what conditions—without executing the work itself. Concretely, it coordinates activities, waits for external events, manages timers and conditional branches, but it never touches the network or the disk.

You can sum this up in one phrase to keep in mind every time you review code: activities do the work, the workflow directs. If you see httpx or open() inside a workflow class, something needs to be moved.

Workflow structure

A workflow is a Python class decorated with @workflows.workflow.define(), containing an entry method decorated with @workflows.workflow.entrypoint:

import mistralai.workflows as workflows

@workflows.workflow.define(name="mon_premier_workflow")
class MonPremierWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        """Workflow entry point."""
        resultat = await saluer(nom)
        return resultat

The @workflows.workflow.define(name="...") decorator registers the class as a workflow. The name it receives is the unique identifier you will use to trigger an execution—this string, not the Python class name, is what the caller sees. The @workflows.workflow.entrypoint decorator marks the method called at startup. The first parameter, self, remains the class instance, as in any Python object, and every activity call uses await.

Complete example: processing pipeline

Here is a realistic workflow that orchestrates multiple activities to process a document. Notice the division of labor: the three functions at the top contain all the code that touches the outside world, while the class at the bottom simply chains await calls and assembles a result.

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

# --- Activities ---

@workflows.activity()
async def extraire_texte(url_document: str) -> dict:
    """Download and extract text from a document."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url_document)
        texte = response.text
    return {"texte": texte, "longueur": len(texte)}

@workflows.activity()
async def analyser_avec_mistral(texte: str) -> dict:
    """Analyze text with Mistral to extract key points."""
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Extract the 5 key points from this text."},
            {"role": "user", "content": texte}
        ]
    )
    return {"analyse": response.choices[0].message.content}

@workflows.activity()
async def sauvegarder_resultat(analyse: dict) -> dict:
    """Save the analysis result."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.example.com/rapports",
            json=analyse
        )
        return {"id_rapport": response.json()["id"]}

# --- Workflow ---

@workflows.workflow.define(name="pipeline_analyse_document")
class PipelineAnalyseDocument:
    @workflows.workflow.entrypoint
    async def run(self, url_document: str) -> dict:
        """Orchestrate extraction, analysis, and saving."""
        # Step 1: Extract text
        extraction = await extraire_texte(url_document)

        # Step 2: Analyze with Mistral
        analyse = await analyser_avec_mistral(extraction["texte"])

        # Step 3: Save
        sauvegarde = await sauvegarder_resultat(analyse)

        return {
            "url_source": url_document,
            "longueur_texte": extraction["longueur"],
            "id_rapport": sauvegarde["id_rapport"]
        }

Every await in the workflow constitutes a persistence point. It is worth verifying this mentally with the example above: if the process fails between step 2 and step 3, the document will not be re-downloaded and the model will not be called again—so no extra token charges. The workflow will resume directly at the save step.

Workflow durability

Workflows survive restarts, infrastructure failures, and transient errors because the platform records every event and reconstructs the workflow state during replay. This reconstruction is reliable only under one condition, mentioned earlier: the workflow code must always produce the same sequence of operations for the same inputs. This is the determinism constraint. We will cover the details in lesson 8.

Workflow with conditional branching

A workflow is not limited to a linear sequence. It can contain conditional logic, such as in this order processing example where the path depends on stock status:

@workflows.workflow.define(name="traitement_commande")
class OrderProcessing:
    @workflows.workflow.entrypoint
    async def run(self, order: dict) -> dict:
        # Check stock
        stock = await check_stock(order["produit_id"])

        if stock["disponible"]:
            # Standard path
            payment = await process_payment(order["montant"])
            shipment = await prepare_shipment(order)
            await send_confirmation(order["email"], shipment)
            return {"status": "confirmée", "tracking": shipment["tracking"]}
        else:
            # Alternative path
            await notify_stockout(order["email"], order["produit_id"])
            return {"status": "rupture_stock"}

if/else branches are fully supported. The only constraint concerns the material used for the decision: the branch outcome must depend on the return value of an activity—a value recorded in the log and replayed identically—not on a direct call to an external service from the workflow. Here, verifier_stock queries inventory within an activity, and the if statement reads only a dictionary that has already been persisted.

Inputs with Pydantic

A workflow that takes six positional parameters quickly becomes unreadable. For structured inputs, use Pydantic models, which also provide type validation at the start of execution:

from pydantic import BaseModel

class ReportParams(BaseModel):
    report_type: str
    include_details: bool = False
    start_date: str
    end_date: str

@workflows.workflow.define(name="generation_rapport")
class ReportGeneration:
    @workflows.workflow.entrypoint
    async def run(self, params: ReportParams) -> dict:
        data = await collect_data(params.report_type, params.start_date, params.end_date)
        
        if params.include_details:
            data = await enrich_data(data)
        
        report = await generate_pdf_report(data)
        return {"url_rapport": report["url"]}

One point of caution when triggering this workflow: when you use a BaseModel, the fields become top-level keys in the JSON input sent to the workflow. You will send {"type_rapport": ..., "date_debut": ...}, not an object nested under a params key.

Best practices

Four habits make a difference over the long term. First, stick to one responsibility per workflow: mixing billing and sales follow-ups in the same class makes it impossible to replay one without the other. Next, pay attention to naming, since the workflow name is a public identifier called from other systems—"traitement_commande" is maintainable, "workflow_1" is not. Do not introduce any I/O in the workflow: every network call, file access, or database query belongs in an activity. Finally, keep the workflow code simple by offloading complex logic—parsing, transformations, heavy calculations—to activities, where it can be tested in isolation without determinism constraints.

Key takeaways

A workflow is a class decorated with @workflows.workflow.define() that orchestrates activities
The @workflows.workflow.entrypoint method is the entry point called at startup
Every activity await is a persistence point (automatic resumption in case of failure)
Conditional branches and loops are supported
Use Pydantic models for structured input parameters
The workflow must contain no direct I/O operations