Create a worker

The role of the worker

The worker is the Python process that actually executes your workflows and activities. It connects to the Mistral platform, listens for pending tasks, and runs them locally on your machine or server. This architectural detail often surprises newcomers: your code never leaves your environment; only the orchestration state resides with Mistral.

The implication is direct and explains most “nothing is happening” issues encountered early in a project: without a running worker, your workflows cannot execute. You can trigger an execution, but it will simply wait until a capable worker appears. The worker is the link between the Mistral orchestration platform and your local code.

Create a minimal worker

A functional worker fits in just a few lines. The example below intentionally combines the activity, workflow, and startup logic into a single file so you can run it immediately:

import asyncio
import mistralai.workflows as workflows

# Import your workflows and activities
@workflows.activity()
async def dire_bonjour(nom: str) -> dict:
    return {"message": f"Bonjour, {nom} !"}

@workflows.workflow.define(name="salutation_workflow")
class SalutationWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        resultat = await dire_bonjour(nom)
        return resultat

# Start the worker
async def main() -> None:
    await workflows.run_worker([SalutationWorkflow])

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

The key function is workflows.run_worker(). It takes a list of workflow classes that this worker can execute—and only workflows. Activities are discovered through them, so you do not need to list activities separately.

What run_worker() does

When you call workflows.run_worker([MyWorkflow]), five things happen in sequence. First, the worker authenticates with the Mistral platform using your API key. Then, it registers the declared workflows and activities. This registration lets the platform know that an executor exists for the salutation_workflow identifier. Next, it opens a persistent connection and waits for tasks, remaining intentionally idle until an execution is triggered. When an execution arrives, the worker runs the code locally, and the results are sent back to the platform for persistence.

The worker stays active as long as the Python process runs, which is why it never returns control to your terminal. Use Ctrl+C to stop it cleanly.

Worker with multiple workflows

Nothing requires one worker per workflow. A single process can host multiple workflows, even when they share activities. Here, extraire_texte and analyser_sentiment serve both:

import asyncio
import mistralai.workflows as workflows

# --- Activities ---

@workflows.activity()
async def extraire_texte(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    return {"texte": response.text}

@workflows.activity()
async def analyser_sentiment(texte: str) -> dict:
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Analysez le sentiment : positif, négatif ou neutre."},
            {"role": "user", "content": texte}
        ]
    )
    return {"sentiment": response.choices[0].message.content}

@workflows.activity()
async def envoyer_alerte(message: str) -> dict:
    # Envoi vers Slack, email, etc.
    return {"status": "envoyé"}

# --- Workflows ---

@workflows.workflow.define(name="analyse_document")
class DocumentAnalysis:
    @workflows.workflow.entrypoint
    async def run(self, url: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        return sentiment

@workflows.workflow.define(name="surveillance_mentions")
class MentionMonitoring:
    @workflows.workflow.entrypoint
    async def run(self, url: str, seuil: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        if sentiment["sentiment"] == "négatif":
            await envoyer_alerte(f"Mention négative détectée : {url}")
        return sentiment

# --- Worker ---

async def main() -> None:
    await workflows.run_worker([
        DocumentAnalysis,
        MentionMonitoring,
    ])

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

Recommended file structure

A single file works for learning, but not for a live project. For production, separate your code into modules by type:

mon-projet/
├── activities/
│   ├── __init__.py
│   ├── extraction.py      # extraire_texte()
│   ├── analyse.py          # analyser_sentiment()
│   └── notifications.py    # envoyer_alerte()
├── workflows/
│   ├── __init__.py
│   ├── analyse_document.py
│   └── surveillance.py
└── worker.py               # Single entry point

The worker.py file then contains only imports and startup logic, making it a readable inventory of what this process can execute:

import asyncio
import mistralai.workflows as workflows

from workflows.document_analysis import DocumentAnalysis
from workflows.monitoring import MentionMonitoring

async def main() -> None:
    await workflows.run_worker([
        DocumentAnalysis,
        MentionMonitoring,
    ])

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

Start the worker

uv run python worker.py

The worker displays a connection message and stays active, waiting for tasks. You should see something like:

Worker connected. Listening for tasks...

Until this line appears, do not trigger anything. First, check your API key and network connection.

Worker in production

A worker launched manually in a terminal disappears after the first SSH disconnect. Hand it off to a process manager—systemd, supervisor, or pm2—which will automatically restart it after a crash or reboot:

# With pm2 (if Node.js is available)
pm2 start "uv run python worker.py" --name workflow-worker

# With systemd (create a .service file)

This automatic restart pairs well with durability: running workflows resume on their own as soon as the new process connects. Regarding secrets, never hard-code the API key in the worker file, or it will end up in your Git repository. The SDK reads the environment variable automatically:

import os

# The SDK automatically reads MISTRAL_API_KEY
# No need to pass it explicitly

Finally, consider logging. The worker emits messages via Python’s standard logging module. A simple configuration at startup gives you a trace of executed activities, which is valuable for understanding what was running during an incident:

import logging
logging.basicConfig(level=logging.INFO)

Key takeaways

• The worker is the process that executes your workflows and activities locally
• workflows.run_worker([...]) takes a list of workflow classes to execute
• A single worker can execute multiple different workflows
• The worker stays active and listens for tasks as long as the process runs
• In production, use a process manager to keep the worker running
• The API key is read automatically from MISTRAL_API_KEY