Your First Workflow

We'll walk through creating a simple workflow that executes a single activity.

Before running your workflow, you'll need to:

1. Connect to the Mistral Console with your Mistral account
2. Select a workspace and create a new API key
3. Keep this API key handy — you'll use it when running uvx mistralai-workflows-cli setup in the next step

The fastest way to get started is with the CLI:

uvx mistralai-workflows-cli setup

uvx mistralai-workflows-cli setup

This scaffolds a ready-to-run Python project with the Workflows SDK already configured, a minimal example workflow, and helper commands to run your worker and trigger executions.

Alternatively, if you're adding Workflows to an existing project, follow the installation instructions.

Create a file my_workflow.py

import mistralai.workflows as workflows


# Activities are async functions that can take any JSON-serializable types
@workflows.activity()
async def hello_world(name: str) -> dict:
    """Activities are the building blocks of workflows.
    1. It's the place where you do the actual work
       (e.g. call an API, process data, CPU intensive tasks, etc.)
    2. It must be async
    3. Parameters can be any JSON-serializable type (str, int, dict, list, etc.)
    """
    return {"message": f"Hello, {name}!"}


# Workflows orchestrate activities
@workflows.workflow.define(name="simple_example_workflow")
class SimpleExampleWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, name: str) -> dict:
        """Workflow entry point.
        1. It coordinate multiple activities
        2. It may need to wait for external events
        """
        return await hello_world(name)


# The Worker runs your workflows and activities
async def main() -> None:
    await workflows.run_worker([SimpleExampleWorkflow])

if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

import mistralai.workflows as workflows


# Activities are async functions that can take any JSON-serializable types
@workflows.activity()
async def hello_world(name: str) -> dict:
    """Activities are the building blocks of workflows.
    1. It's the place where you do the actual work
       (e.g. call an API, process data, CPU intensive tasks, etc.)
    2. It must be async
    3. Parameters can be any JSON-serializable type (str, int, dict, list, etc.)
    """
    return {"message": f"Hello, {name}!"}


# Workflows orchestrate activities
@workflows.workflow.define(name="simple_example_workflow")
class SimpleExampleWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, name: str) -> dict:
        """Workflow entry point.
        1. It coordinate multiple activities
        2. It may need to wait for external events
        """
        return await hello_world(name)


# The Worker runs your workflows and activities
async def main() -> None:
    await workflows.run_worker([SimpleExampleWorkflow])

if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

uv run python my_workflow.py

uv run python my_workflow.py

The worker will start, connect to the Mistral API, and register your workflow and wait for any task to do. You will need to trigger execution in order to see the workflow running.

1. Visit https://console.mistral.ai/ (make sure you are located in the same workspace than the one you created your API key in)
2. Click Workflows in the left sidebar
3. Select simple_example_workflow
4. Launch the workflow using the Start Workflow button and the input {"name": <your_name>}
5. Find your execution in the Executions tab
6. Make sure the workflow ran until the end by checking if the output is correct

For scaling patterns and running multiple workers, see Core Concepts - Scaling with Multiple Workers.