Mistral Workflows

https://docs.mistral.ai/studio-api/workflows/
Admin
Mistral Workflows is a platform for building production-grade AI workflows that combine multi-step...
Tokens:58,601
Snippets:631
Trust Score:10
Update:2 months ago
Show doc for...
Tokens:
Raw
### Start cookbook examples

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/cookbook_examples

Navigate to the project directory and register workflows with the workspace.

```bash
cd my-workflow # or the name of the folder you created
make start-examples
```

--------------------------------

### Cargo release input JSON

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/cookbook_examples

Example input structure for starting the cargo-release-compliance workflow in Studio.

```json
{
  "document_uri": "src/examples/cargo_release/sample_data/shipping_doc_anomaly.png",
  "shipment_id": "BL-2024-HAM-009371"
}
```

--------------------------------

### Install Workflows SDK with storage support

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/payload_offloading

Install the necessary dependencies for your specific cloud storage provider.

```bash
# AWS S3 support
uv add "mistralai-workflows[s3]"

# Azure Blob Storage support
uv add "mistralai-workflows[azure]"

# Google Cloud Storage support
uv add "mistralai-workflows[gcs]"

# All storage providers
uv add "mistralai-workflows[storage]"
```

--------------------------------

### Install Workflows SDK

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/installation

Installs the core Workflows package using the uv package manager.

```bash
uv add mistralai-workflows
```

--------------------------------

### Install Workflow Plugins

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/plugins

Commands to install the core workflows SDK and specific plugin packages via pip.

```bash
pip install mistralai-workflows
pip install mistralai-workflows-plugins-mistralai
pip install mistralai-workflows-plugins-webhook
```

--------------------------------

### Install Compression Dependencies

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/compression

Install the required extra dependencies to enable payload compression support.

```bash
uv add "mistralai[workflow-payload-compression]"
```

--------------------------------

### Verify Installation

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/installation

Runs a Python command to confirm the Workflows package is correctly installed.

```bash
uv run python -c "import mistralai.workflows; print('Workflows is installed successfully!')"
```

--------------------------------

### Install encryption dependencies

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/encryption

Install the required encryption extra to enable AES-GCM support.

```bash
uv add "mistralai[workflow-payload-encryption]"
```

--------------------------------

### Install Workflows with Optional Dependencies

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/installation

Installs the Workflows package with specific extras for Mistral integration or cloud storage providers.

```bash
uv add "mistralai-workflows[mistralai]"
```

```bash
# AWS S3 support
uv add "mistralai-workflows[s3]"

# Azure Blob Storage support
uv add "mistralai-workflows[azure]"

# Google Cloud Storage support
uv add "mistralai-workflows[gcs]"

# All storage providers
uv add "mistralai-workflows[storage]"
```

--------------------------------

### Install Mistral workflows plugin

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/durable_agents

Install the necessary package to enable durable agent functionality.

```bash
uv add 'mistralai-workflows[mistralai]'
```

--------------------------------

### Deployment response structure

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/deployments

Example JSON response returned when querying deployment information.

```json
{
  "id": "019d2585-21fc-7063-90ae-31a283c784a1",
  "name": "invoice-service",
  "is_active": true,
  "created_at": "2026-03-25 15:02:55.234755+00:00",
  "updated_at": "2026-04-02 17:42:41.828219+00:00",
  "workers": [
    {
      "name": "my-worker-host",
      "created_at": "2026-03-25 15:15:07.911486+00:00",
      "updated_at": "2026-04-02 15:42:13.679937+00:00"
    }
  ]
}
```

--------------------------------

### Trigger workflow execution

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/your_first_workflow

Examples for triggering a workflow using the Makefile or the Python SDK.

```bash
make execute workflow=hello-world input='{"name": "World"}'
```

```python
from mistralai.client import Mistral

client = Mistral(api_key="your_api_key")

execution = client.workflows.execute_workflow(
    workflow_identifier="hello-world",
    input={"name": "Alice"},
)
print(execution.model_dump_json(indent=2))
```

--------------------------------

### Start the worker

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/your_first_workflow

Discovers and registers workflow classes to begin processing tasks.

```python
discovered = discover_workflows()
await mistralai_workflows.run_worker(discovered)
```

--------------------------------

### Install Workflow Payload Compression Dependencies

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/compression

Install the necessary compression extras on all workers and clients to enable payload compression.

```bash
uv add "mistralai[workflow-payload-compression]"
```

```bash
uv add "mistralai[workflow-payload-compression]"
```

--------------------------------

### Install Mistral workflows plugin

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/connectors

Install the required package to enable Connector slot functionality in your workflow environment.

```bash
uv add "mistralai-workflows[mistralai]"
```

--------------------------------

### Configure pyproject.toml for Custom Libraries

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/plugins

Example configuration for a custom workflow package, ensuring the mistralai-workflows dependency is included.

```toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "acme-workflows"
version = "0.1.0"
dependencies = [
    "mistralai-workflows>=2.0.0",
]
```

--------------------------------

### Example Event Structure

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/consuming_events

A sample JSON payload representing a workflow execution event.

```json
{
  "stream": "token",
  "broker_sequence": 1,
  "timestamp": "2025-01-15T10:30:00Z",
  "data": {
    "event_type": "WORKFLOW_EXECUTION_STARTED",
    "event_id": "evt_abc123",
    "event_timestamp": 1736938200000000000,
    "workflow_name": "my-workflow",
    "workflow_exec_id": "abc123",
    "root_workflow_exec_id": "abc123",
    "parent_workflow_exec_id": null,
    "workflow_run_id": "run_456",
    "attributes": {}
  },
  "workflow_context": {
    "workflow_name": "my-workflow",
    "workflow_exec_id": "abc123",
    "parent_workflow_exec_id": null
  }
}
```

--------------------------------

### Configure Concurrency Limits in Python

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/concurrency

Example configuration for setting concurrency limits within a workflow execution.

```python
max_concurrent_executions_per_worker=50
max_concurrent_scheduled_tasks=150
```

--------------------------------

### Import Custom Workflow Components

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/plugins

Importing activities and workflows from a custom installed package.

```python
from acme_workflows.activities import my_custom_activity
from acme_workflows.workflows import my_rag_pipeline
```

--------------------------------

### Run the worker via Makefile

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/your_first_workflow

Starts the worker process using the project's Makefile.

```bash
make start-worker
```

--------------------------------

### Workflow Not Running Error Response

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/error_codes

Example response for WF_1102, indicating the current status of the workflow.

```json
{
  "detail": "Workflow not running",
  "code": "WF_1102",
  "status": "COMPLETED"
}
```

--------------------------------

### Implementing a chat completion workflow

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/plugins

Example of using the mistralai_chat_complete activity within a defined workflow class.

```python
import mistralai.workflows as workflows
from mistralai.workflows import workflow
from mistralai.workflows.plugins.mistralai import (
    ChatCompletionRequest,
    UserMessage,
    mistralai_chat_complete,
)
from pydantic import BaseModel

class Input(BaseModel):
    text: str

@workflow.define(name="summarize")
class SummarizeWorkflow:
    @workflow.entrypoint
    async def run(self, params: Input) -> str:
        request = ChatCompletionRequest(
            model="mistral-large-latest",
            messages=[UserMessage(content=f"Summarize this text:\n\n{params.text}")],
        )
        result = await mistralai_chat_complete(request)
        return result.choices[0].message.content
```

--------------------------------

### Define a Workflow with Offloaded Payloads

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/payload_offloading

Example of a workflow entrypoint that processes offloaded payloads without unwrapping them.

```python
import mistralai.workflows as workflows

@workflows.workflow.define(name="transcribe-and-summarize")
class TranscribeWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, audio_id: str) -> str:
        result = await transcribe(TranscriptionPayload(audio_id=audio_id))

        # Pass the offloaded field through. Do not unwrap in the workflow
        summary = await summarize(SummaryInput(transcript=result.transcript))
        return summary.text
```

--------------------------------

### Example Server-Sent Events response

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/consuming_events

The streaming API returns events in the SSE format, each containing a workflow context and broker sequence.

```text
data: {"stream":"token","data":{"event_type":"CUSTOM_TASK_IN_PROGRESS",...},"broker_sequence":1,"workflow_context":{...}}

data: {"stream":"token","data":{"event_type":"CUSTOM_TASK_IN_PROGRESS",...},"broker_sequence":2,"workflow_context":{...}}

data: {"stream":"token","data":{"event_type":"WORKFLOW_EXECUTION_COMPLETED",...},"broker_sequence":3,"workflow_context":{...}}
```

--------------------------------

### Configure Client Workflow Encoding

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/compression

Use the Mistral Python SDK to configure compression settings before starting a workflow.

```python
import asyncio

from mistralai.client import Mistral
from mistralai.extra.workflows import (
    WorkflowEncodingConfig,
    configure_workflow_encoding,
)
from mistralai.extra.workflows.encoding import (
    PayloadCompressionConfig,
    ZstdCompressionConfig,
)

async def main():
    client = Mistral(api_key="your_api_key")

    await configure_workflow_encoding(
        WorkflowEncodingConfig(
            payload_compression=PayloadCompressionConfig(
                min_size_bytes=1024 * 1024,  # 1MB (default)
                algorithm_config=ZstdCompressionConfig(level=3),
            )
        ),
        client=client,
    )

    execution = client.workflows.execute_workflow(
        workflow_identifier="my-workflow",
        input={"large_data": "..."},  # Compressed if >= 1MB
    )
    print(execution)


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

--------------------------------

### Set up a cookbook project

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/cookbook_examples

Initialize a new project by selecting a cookbook template from the CLI.

```bash
uvx mistralai-workflows-cli setup
```

--------------------------------

### Scaffold a new project

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/your_first_workflow

Initializes a new Python project with the Workflows SDK configured.

```bash
uvx mistralai-workflows-cli@latest setup
```

--------------------------------

### Implement Continue-As-New in Python

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/continue_as_new

Use should_continue_as_new to monitor history size and continue_as_new to restart the workflow with updated state parameters.

```python
import mistralai.workflows as workflows
from pydantic import BaseModel

class ProcessorState(BaseModel):
    page: int = 0
    total_processed: int = 0

@workflows.workflow.define(name="long-running-processor")
class LongRunningProcessor:
    @workflows.workflow.entrypoint
    async def run(self, page: int = 0, total_processed: int = 0):  # State parameters are required
        while True:
            # Process current page
            result = await process_page(page)
            total_processed += result.count
            page += 1

            # When history grows large, restart with a fresh history
            if workflows.workflow.should_continue_as_new():
                workflows.workflow.continue_as_new(ProcessorState(page=page, total_processed=total_processed))
```

--------------------------------

### Configure a Streamable HTTP MCP Agent

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/durable_agents

Initializes an agent with a Streamable HTTP MCP server configuration.

```python
from mistralai.workflows.plugins.mistralai import MCPStreamableHTTPConfig

mcp_config = MCPStreamableHTTPConfig(
    url="https://your-mcp-server.com/mcp",
    name="remote-tools",
)

agent = workflows_mistralai.Agent(
    model="mistral-medium-latest",
    name="streamable-http-mcp-agent",
    description="Agent with access to a Streamable HTTP MCP server",
    mcp_clients=[mcp_config],
)
```

--------------------------------

### Implement a basic agent workflow

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/durable_agents

Create a complete workflow that defines an activity as a tool and executes it via a RemoteSession.

```python
import mistralai.workflows as workflows
import mistralai.workflows.plugins.mistralai as workflows_mistralai
from mistralai.client.models import TextChunk

@workflows.activity()
async def get_interest_rate(year: int) -> dict:
    """Get the interest rate for a given year.

    Args:
        year: The year to get the interest rate for
    """
    # Your implementation here
    return {"interest_rate": 1.62}

@workflows.workflow.define(name="finance_agent_workflow")
class FinanceAgentWorkflow:
    @workflows.workflow.entrypoint
    async def entrypoint(self, question: str) -> dict:
        session = workflows_mistralai.RemoteSession()

        agent = workflows_mistralai.Agent(
            model="mistral-medium-latest",
            name="finance-agent",
            description="Agent for financial queries",
            instructions="Use tools to answer financial questions.",
            tools=[get_interest_rate],
        )

        outputs = await workflows_mistralai.Runner.run(
            agent=agent,
            inputs=question,
            session=session,
        )

        answer = "\n".join([
            output.text for output in outputs
            if isinstance(output, TextChunk)
        ])

        return {"answer": answer}
```

--------------------------------

### GET /v1/workflows/schedules

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/scheduling

Lists all schedules for all active workflows in the workspace.

```APIDOC
## GET /v1/workflows/schedules

### Description
Retrieves a list of all schedules.

### Method
GET

### Endpoint
https://api.mistral.ai/v1/workflows/schedules
```

--------------------------------

### Configure Stdio MCP Server

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/durable_agents

Connect to a local command-line MCP server using MCPStdioConfig.

```python
import mistralai.workflows.plugins.mistralai as workflows_mistralai
from mistralai.workflows.plugins.mistralai import MCPStdioConfig

mcp_config = MCPStdioConfig(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-everything"],
    name="server-everything",
)

agent = workflows_mistralai.Agent(
    model="mistral-medium-latest",
    name="mcp-agent",
    description="Agent with access to MCP tools",
    mcp_clients=[mcp_config],
)
```

--------------------------------

### GET /v1/workflows/schedules/{schedule_id}

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/scheduling

Retrieves the specification of a specific schedule by its ID.

```APIDOC
## GET /v1/workflows/schedules/{schedule_id}

### Description
Gets a schedule by schedule ID.

### Method
GET

### Endpoint
https://api.mistral.ai/v1/workflows/schedules/{schedule_id}

### Parameters
#### Path Parameters
- **schedule_id** (string) - Required - The unique identifier of the schedule.
```

--------------------------------

### Get a specific schedule

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/scheduling

Retrieves the full specification of a schedule by its unique identifier.

```python
schedule = client.workflows.schedules.get_schedule(
    schedule_id="your-schedule-id",
)
print(schedule.paused, schedule.future_executions)
```

```bash
curl -H "Authorization: Bearer $MISTRAL_API_KEY" \
  "https://api.mistral.ai/v1/workflows/schedules/your-schedule-id"
```

--------------------------------

### Configure Activity Field Offloading

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/payload_offloading

Environment variables for configuring storage providers for activity field offloading.

```bash
ACTIVITY_ATTRIBUTES_OFFLOADING__MIN_SIZE_BYTES=1048576   # 1MB (default)
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__STORAGE_PROVIDER=azure
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__CONTAINER_NAME=my-workflow-payloads
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__AZURE_CONNECTION_STRING="..."
```

```bash
ACTIVITY_ATTRIBUTES_OFFLOADING__MIN_SIZE_BYTES=1048576   # 1MB (default)
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__STORAGE_PROVIDER=s3
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__BUCKET_NAME=my-workflow-payloads
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__REGION_NAME=us-east-1  # optional
# Credentials are optional. Uses AWS default credential chain if not set
# ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__AWS_ACCESS_KEY_ID=your-access-key
# ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__AWS_SECRET_ACCESS_KEY=your-secret-key
```

```bash
ACTIVITY_ATTRIBUTES_OFFLOADING__MIN_SIZE_BYTES=1048576   # 1MB (default)
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__STORAGE_PROVIDER=gcs
ACTIVITY_ATTRIBUTES_OFFLOADING__STORAGE_CONFIG__BUCKET_ID=my-workflow-payloads
# Uses Application Default Credentials (ADC) from environment
```

--------------------------------

### Async I/O Best Practices

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/activities

Always use asynchronous libraries for I/O to avoid blocking the worker process.

```python
# ✅ Async HTTP
async with httpx.AsyncClient() as client:
    response = await client.post(url, json=data)

# ❌ Blocking HTTP, stalls the worker
response = requests.post(url, json=data)
```

--------------------------------

### GET /v1/workflows/executions/{execution_id}

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/executions

Retrieves the current state of a specific workflow execution by its ID.

```APIDOC
## GET /v1/workflows/executions/{execution_id}

### Description
Fetches the current state and status of a workflow execution.

### Method
GET

### Endpoint
/v1/workflows/executions/{execution_id}

### Parameters
#### Path Parameters
- **execution_id** (string) - Required - The unique identifier of the execution.
```

--------------------------------

### Configure storage provider via environment variables

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/payload_offloading

Set environment variables to enable payload offloading for S3 or GCS providers.

```bash
TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__STORAGE_PROVIDER=s3
TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__BUCKET_NAME=workflow-payloads
TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__REGION_NAME=us-east-1  # optional
# Credentials are optional. Uses AWS default credential chain if not set
# TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__AWS_ACCESS_KEY_ID=your-access-key
# TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__AWS_SECRET_ACCESS_KEY=your-secret-key
```

```bash
TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__STORAGE_PROVIDER=gcs
TEMPORAL_PAYLOAD_OFFLOADING__STORAGE_CONFIG__BUCKET_ID=workflow-payloads
# Uses Application Default Credentials (ADC) from environment
```

--------------------------------

### GET /v1/workflows/executions/{execution_id}/history

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/executions

Retrieves the full event log for a specific workflow execution.

```APIDOC
## GET /v1/workflows/executions/{execution_id}/history

### Description
Retrieves the complete event history for the specified execution.

### Method
GET

### Endpoint
/v1/workflows/executions/{execution_id}/history

### Parameters
#### Path Parameters
- **execution_id** (string) - Required - The unique identifier of the execution.
```

--------------------------------

### Injecting Dependencies into Activities

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/dependency_injection

Demonstrates using Depends() to provide configuration, database instances, and session context to an activity.

```python
import asyncio
import mistralai.workflows as workflows
from mistralai.workflows import Depends
from contextlib import contextmanager
from pydantic import BaseModel

def get_config() -> dict:
    return {"timeout": 30, "retries": 3, "api_url": "https://api.example.com"}

class FakeDB:
    def __init__(self):
        self.records = []

    def insert(self, name: str):
        self.records.append(name)
        return len(self.records)

async def get_db() -> FakeDB:
    return FakeDB()

@contextmanager
def get_session():
    session = {"logged_in": True, "events": []}
    try:
        yield session
    finally:
        session["logged_in"] = False

@workflows.activity()
async def create_user(
    name: str,
    db: FakeDB = Depends(get_db),
    config: dict = Depends(get_config),
    session: dict = Depends(get_session),
) -> dict:
    record_id = db.insert(name)
    session["events"].append("user_created")
    return {
        "status": "success",
        "record_id": record_id,
        "timeout_used": config["timeout"],
    }

class Input(BaseModel):
    name: str

@workflows.workflow.define(name="di_workflow")
class DIWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, params: Input) -> dict:
        return await create_user(params.name)

async def main():
    result = await workflows.execute_workflow(
        DIWorkflow,
        params=Input(name="Alice"),
    )
    print(result)
```

```text
{'status': 'success', 'record_id': 1, 'timeout_used': 30}
```

--------------------------------

### Combine local and remote activities in Python

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/activities/local_activities

Demonstrates mixing local execution for fast lookups with standard remote activity execution for external API calls.

```python
@workflows.workflow.define(name="mixed-workflow")
class MixedWorkflow:
    @workflows.workflow.entrypoint
    async def execute(self, email: str) -> dict:
        # Fast lookup runs locally
        with run_activities_locally():
            domain = await quick_lookup(email)

        # Slow operation runs as regular activity
        verified = await external_api_call(email)

        return {"domain": domain, "verified": verified}
```

--------------------------------

### GET /v1/workflows/deployments/{name}

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/deployments

Retrieves the details of a specific deployment, including its active status and associated workers.

```APIDOC
## GET /v1/workflows/deployments/{name}

### Description
Retrieves the details of a specific deployment by its name.

### Method
GET

### Endpoint
https://api.mistral.ai/v1/workflows/deployments/{name}

### Parameters
#### Path Parameters
- **name** (string) - Required - The name of the deployment to retrieve.

### Response
#### Success Response (200)
- **id** (string) - The unique identifier of the deployment.
- **name** (string) - The name of the deployment.
- **is_active** (boolean) - Indicates if the deployment is currently active.
- **created_at** (string) - Timestamp of creation.
- **updated_at** (string) - Timestamp of last update.
- **workers** (array) - List of workers associated with the deployment.

#### Response Example
{
  "id": "019d2585-21fc-7063-90ae-31a283c784a1",
  "name": "invoice-service",
  "is_active": true,
  "created_at": "2026-03-25 15:02:55.234755+00:00",
  "updated_at": "2026-04-02 17:42:41.828219+00:00",
  "workers": [
    {
      "name": "my-worker-host",
      "created_at": "2026-03-25 15:15:07.911486+00:00",
      "updated_at": "2026-04-02 15:42:13.679937+00:00"
    }
  ]
}
```

--------------------------------

### Retrieve workflow execution ID

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/workflows

Gets the unique identifier for the current workflow execution for logging or correlation.

```python
import mistralai.workflows as workflows
execution_id = workflows.get_execution_id()
```

--------------------------------

### Optimize Rate Limit Key Usage

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/rate_limiting

Demonstrates how to consolidate multiple activities under a single rate limit key to reduce the number of internal workers and pollers created.

```python
# ❌ Avoid — each unique key creates an extra internal worker
@workflows.activity(rate_limit=workflows.RateLimit(time_window_in_sec=1, max_execution=10, key="ocr"))
async def run_ocr(params: OcrParams) -> OcrResult: ...

@workflows.activity(rate_limit=workflows.RateLimit(time_window_in_sec=1, max_execution=10, key="classify"))
async def classify_document(params: ClassifyParams) -> ClassifyResult: ...

# ✅ Prefer — share one key across activities that hit the same resource
@workflows.activity(rate_limit=workflows.RateLimit(time_window_in_sec=1, max_execution=10, key="document_pipeline"))
async def run_ocr(params: OcrParams) -> OcrResult: ...

@workflows.activity(rate_limit=workflows.RateLimit(time_window_in_sec=1, max_execution=10, key="document_pipeline"))
async def classify_document(params: ClassifyParams) -> ClassifyResult: ...
```

--------------------------------

### Fire and forget child workflow

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/sub_workflows

Start a child workflow without waiting for completion by setting wait=False.

```python
handle = await workflow.execute_workflow(
    ChildWorkflow,
    params=child_params,
    execution_timeout=timedelta(hours=1),
    wait=False,
)
# Parent continues immediately — child runs independently
# Optionally await later: result = await handle
```

--------------------------------

### Use user credentials in activities

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/on_behalf_of

Initialize the Mistral client with use_executor_credentials=True to perform actions as the triggering user.

```python
from datetime import timedelta

import mistralai.workflows as workflows
from mistralai.workflows.client import get_mistral_client

@workflows.workflow.define(name="user_search", on_behalf_of=True)
class UserSearchWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, query: str) -> str:
        return await search_with_user_identity(query)

@workflows.activity(start_to_close_timeout=timedelta(seconds=30))
async def search_with_user_identity(query: str) -> str:
    client = get_mistral_client(use_executor_credentials=True)
    response = await client.chat.complete_async(
        model="mistral-medium-latest",
        messages=[{"role": "user", "content": query}],
    )
    return response.choices[0].message.content
```

--------------------------------

### Invalid Request Error Response

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/error_codes

Example response for WF_1001, including a list of valid reset events.

```json
{
  "detail": "Invalid event ID for reset",
  "code": "WF_1001",
  "valid_reset_events": [3, 7, 12]
}
```

--------------------------------

### Configure activity settings for local execution in Python

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/activities/local_activities

Timeouts and retry policies remain functional for local activities, though rate limiting is not supported.

```python
@workflows.activity(
    start_to_close_timeout=timedelta(seconds=5),
    retry_policy_max_attempts=3
)
async def timed_local_activity(data: str) -> str:
    # Timeouts and retry policies apply even when running locally.
    # Rate limiting is NOT supported for local activities.
    return data.upper()
```

--------------------------------

### Workflow input JSON schema

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/workflows

Example of the top-level input keys generated when using a Pydantic model.

```json
{"report_type": "daily", "include_details": true}
```

--------------------------------

### Manage Connector Credentials

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/connectors

Demonstrates creating, updating, listing, and deleting user credentials for connectors using the Mistral client.

```python
import os
from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

# Bearer connector: store a named credential and mark it default
await client.beta.connectors.create_or_update_user_credentials_async(
    connector_id_or_name="github_app",
    name="github-pat-full",
    credentials={"bearer_token": os.environ["GITHUB_PAT"]},
    is_default=True,
)

# OAuth2 connector: request an auth URL, the user completes it in a browser
result = await client.beta.connectors.get_auth_url_async(
    connector_id_or_name="outlook_calendar",
    credentials_name="personal",
)
print(result.auth_url)

# List and delete
await client.beta.connectors.list_user_credentials_async(
    connector_id_or_name="github_app",
)
await client.beta.connectors.delete_user_credentials_async(
    connector_id_or_name="github_app",
    credentials_name="github-pat-old",
)
```

--------------------------------

### Start a worker in Python

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/workers

Use run_worker to register workflow classes and begin polling for tasks until the process is terminated.

```python
import asyncio
import mistralai.workflows as workflows

async def main():
    await workflows.run_worker([MyWorkflow])

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

--------------------------------

### Implement TodoList progress tracking in Python

Source: https://docs.mistral.ai/studio-api/workflows/interacting-with-workflows/conversational_workflows/progress_tracking

Define a workflow with multiple steps using TodoListItem and manage their execution status using the TodoList context manager.

```python
import mistralai.workflows as workflows
import mistralai.workflows.plugins.mistralai as workflows_mistralai

@workflows.workflow.define(
    name="expense-processing-workflow",
    workflow_display_name="Expense Processing",
    workflow_description="Process expense with step-by-step progress",
)
class ExpenseProcessingWorkflow(workflows.InteractiveWorkflow):
    @workflows.workflow.entrypoint
    async def run(self, expense_id: str) -> workflows_mistralai.ChatAssistantWorkflowOutput:
        # Define the steps
        validate_item = workflows_mistralai.TodoListItem(
            title="Validate expense",
            description="Check expense details and receipts"
        )
        approve_item = workflows_mistralai.TodoListItem(
            title="Get approval",
            description="Route to manager for approval"
        )
        process_item = workflows_mistralai.TodoListItem(
            title="Process payment",
            description="Submit for reimbursement"
        )

        async with workflows_mistralai.TodoList(
            items=[validate_item, approve_item, process_item]
        ) as todo_list:
            # Step 1: Validate (using context manager for automatic status)
            async with validate_item:
                pass  # ... validation logic ...

            # Step 2: Approve (using context manager for automatic status)
            async with approve_item:
                pass  # ... approval logic ...

            # Step 3: Process (using context manager for automatic status)
            async with process_item:
                pass  # ... processing logic ...

        return workflows_mistralai.ChatAssistantWorkflowOutput(
            content=[workflows_mistralai.TextOutput(text=f"Expense {expense_id} processed successfully")]
        )
```

--------------------------------

### Execute Linear Summarization Workflow

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/cookbook_examples

Commands and JSON input for running the Linear project summarization workflow.

```bash
make execute-linear-summary
```

```bash
make execute-linear-summary \
  input='{"team":"Engineering","project":"*"}'
```

```json
{
  "team": "Engineering",
  "project": "*"
}
```

--------------------------------

### Primitive and Multi-parameter Inputs

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/workflows

Workflows accept JSON-serializable types as parameters for the entrypoint method.

```python
@workflow.define(name="report_workflow")
class ReportWorkflow:
    @workflow.entrypoint
    async def run(self, report_type: str, include_details: bool = False) -> dict:
        ...
```

```json
{"report_type": "daily", "include_details": false}
```

--------------------------------

### GET /v1/workflows/events/stream

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/consuming_events

Establishes a Server-Sent Events (SSE) connection to stream workflow and activity events based on provided filters.

```APIDOC
## GET /v1/workflows/events/stream

### Description
Streams workflow and activity events using Server-Sent Events (SSE). Use query parameters to filter the stream by scope, workflow, activity, or execution IDs.

### Method
GET

### Endpoint
https://api.mistral.ai/v1/workflows/events/stream

### Parameters
#### Query Parameters
- **scope** (string) - Optional - Filter by 'workflow', 'activity', or '*'.
- **stream** (string) - Optional - Filter by stream name (e.g., 'token').
- **workflow_name** (string) - Optional - Filter by workflow name.
- **workflow_exec_id** (string) - Optional - Filter by execution ID.
- **root_workflow_exec_id** (string) - Optional - Filter by root workflow execution ID.
- **parent_workflow_exec_id** (string) - Optional - Filter by parent workflow execution ID.
- **activity_name** (string) - Optional - Filter by activity name.
- **activity_id** (string) - Optional - Filter by activity execution ID.
- **start_seq** (integer) - Optional - Resume from a specific sequence number.
- **metadata_filters** (string) - Optional - Filter by metadata (JSON object).
- **workflow_event_types** (string) - Optional - Filter by workflow event types.

### Request Example
curl -N -H "Authorization: Bearer $MISTRAL_API_KEY" \
  "https://api.mistral.ai/v1/workflows/events/stream?workflow_exec_id=${EXEC_ID}&stream=token"

### Response
#### Success Response (200)
- **data** (object) - The event payload containing event_type, broker_sequence, and workflow_context.

#### Response Example
data: {"stream":"token","data":{"event_type":"WORKFLOW_EXECUTION_COMPLETED",...},"broker_sequence":3,"workflow_context":{...}}
```

--------------------------------

### Configure catchup window

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/scheduling

Sets the duration in seconds for retroactively firing missed runs after an outage.

```python
policy=models.SchedulePolicy(
    catchup_window_seconds=86400,
)
```

```bash
curl -X POST "https://api.mistral.ai/v1/workflows/schedules" \
  -H "Authorization: Bearer $MISTRAL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_identifier": "report-workflow",
    "deployment_name": "production",
    "schedule": {
      "input": {},
      "calendars": [{"hour": [{"start": 9}], "minute": [{"start": 0}]}],
      "policy": {
        "catchup_window_seconds": 86400
      }
    }
  }'
```

--------------------------------

### GET /v1/workflows/deployments

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/deployments

List all active deployments in your workspace. You can filter by workflow name or include inactive deployments using query parameters.

```APIDOC
## GET /v1/workflows/deployments

### Description
List all active deployments in your workspace.

### Method
GET

### Endpoint
https://api.mistral.ai/v1/workflows/deployments

### Parameters
#### Query Parameters
- **workflow_name** (string) - Optional - Filter deployments by workflow name.
- **active_only** (boolean) - Optional - Include inactive deployments if set to false.

### Response
#### Success Response (200)
- **deployments** (array) - List of deployment objects.

#### Response Example
{
  "deployments": [
    {
      "id": "019d2585-21fc-7063-90ae-31a283c784a1",
      "name": "invoice-service",
      "is_active": true,
      "created_at": "2026-03-25 15:02:55.234755+00:00",
      "updated_at": "2026-04-02 17:42:34.960883+00:00"
    }
  ]
}
```

--------------------------------

### GET /v1/workflows/events/stream

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/consuming_events

Subscribe to a stream of events across multiple executions. Useful for dashboards, observability tools, or monitoring parent workflows and their descendants.

```APIDOC
## GET /v1/workflows/events/stream

### Description
Subscribe across executions with rich filtering.

### Method
GET

### Endpoint
/v1/workflows/events/stream
```

--------------------------------

### Define an Approval Workflow with Confirmation

Source: https://docs.mistral.ai/studio-api/workflows/interacting-with-workflows/conversational_workflows/forms_and_confirmations

Demonstrates creating an interactive workflow that pauses for user confirmation using AcceptDeclineConfirmation.

```python
import mistralai.workflows as workflows
import mistralai.workflows.plugins.mistralai as workflows_mistralai

@workflows.workflow.define(
    name="approval-workflow",
    workflow_display_name="Approval",
    workflow_description="Confirm an action",
)
class ApprovalWorkflow(workflows.InteractiveWorkflow):
    @workflows.workflow.entrypoint
    async def run(self) -> workflows_mistralai.ChatAssistantWorkflowOutput:
        confirmation = await self.wait_for_input(
            workflows_mistralai.AcceptDeclineConfirmation(
                description="Do you want to proceed with this action?",
                accept_label="Yes, proceed",
                decline_label="Cancel",
            )
        )

        if workflows_mistralai.is_accepted(confirmation):
            return workflows_mistralai.ChatAssistantWorkflowOutput(
                content=[workflows_mistralai.TextOutput(text="Action confirmed!")]
            )
        else:
            return workflows_mistralai.ChatAssistantWorkflowOutput(
                content=[workflows_mistralai.TextOutput(text="Action cancelled.")]
            )
```

--------------------------------

### client.workflows.executions.reset_workflow

Source: https://docs.mistral.ai/studio-api/workflows/managing-workflows-in-production/reset_workflow

Resets a workflow execution to a specific event ID. This terminates the current run and starts a new one from the specified point using the latest code.

```APIDOC
## client.workflows.executions.reset_workflow

### Description
Resets a workflow execution to a specific event ID. This terminates the current run and starts a new one from the specified point using the latest code.

### Parameters
- **execution_id** (str) - Required - The execution to reset
- **event_id** (int) - Required - Event ID to reset to (must be a WORKFLOW_TASK_COMPLETED event)
- **reason** (str) - Required - Reason for the reset (recorded in event history)
- **exclude_signals** (bool) - Optional - Skip replaying signals received after the reset point
- **exclude_updates** (bool) - Optional - Skip replaying updates received after the reset point

### Request Example
client.workflows.executions.reset_workflow(
    execution_id="your-execution-id",
    event_id=42,
    reason="Bug fixed in activity logic",
    exclude_signals=True,
    exclude_updates=True
)
```

--------------------------------

### GET /v1/workflows/executions/{execution_id}/stream

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/consuming_events

Subscribe to a stream of events for a specific workflow execution by its ID. Supports filtering by event source and resuming from a specific sequence number.

```APIDOC
## GET /v1/workflows/executions/{execution_id}/stream

### Description
Subscribe to a single execution by ID. Supports resume from a previous SSE position.

### Method
GET

### Endpoint
/v1/workflows/executions/{execution_id}/stream

### Parameters
#### Path Parameters
- **execution_id** (string) - Required - The unique identifier of the workflow execution.

#### Query Parameters
- **event_source** (string) - Optional - Filter by event source (e.g., 'workflow', 'activity').
- **last_event_id** (integer) - Optional - Resume after a specific sequence number (maps to start_seq). Can also be sent as a 'Last-Event-ID' HTTP header.

### Request Example
curl -N -H "Authorization: Bearer $MISTRAL_API_KEY" "https://api.mistral.ai/v1/workflows/executions/${EXEC_ID}/stream"
```

--------------------------------

### Configure Workflow Encoding with Compression, Offloading, and Encryption

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/compression

Initializes a WorkflowEncodingConfig with compression, S3-based offloading, and full payload encryption.

```python
from mistralai.extra.workflows import WorkflowEncodingConfig
from mistralai.extra.workflows.encoding import (
    BlobStorageConfig,
    PayloadCompressionConfig,
    PayloadEncryptionConfig,
    PayloadEncryptionMode,
    PayloadOffloadingConfig,
)

config = WorkflowEncodingConfig(
    payload_compression=PayloadCompressionConfig(),
    payload_offloading=PayloadOffloadingConfig(
        storage_config=BlobStorageConfig(
            storage_provider="s3",
            bucket_name="workflow-payloads",
        ),
    ),
    payload_encryption=PayloadEncryptionConfig(
        mode=PayloadEncryptionMode.FULL,
        main_key="<your_hex_key>",
    ),
)
```

```python
from mistralai.extra.workflows import WorkflowEncodingConfig
from mistralai.extra.workflows.encoding import (
    BlobStorageConfig,
    PayloadCompressionConfig,
    PayloadEncryptionConfig,
    PayloadEncryptionMode,
    PayloadOffloadingConfig,
)

config = WorkflowEncodingConfig(
    payload_compression=PayloadCompressionConfig(),
    payload_offloading=PayloadOffloadingConfig(
        storage_config=BlobStorageConfig(
            storage_provider="s3",
            bucket_name="workflow-payloads",
        ),
    ),
    payload_encryption=PayloadEncryptionConfig(
        mode=PayloadEncryptionMode.FULL,
        main_key="<your_hex_key>",
    ),
)
```

--------------------------------

### Injecting Persistent API Clients

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/dependency_injection

Initializes an API client once and injects the persistent instance into activities.

```python
import os
import mistralai.workflows as workflows
from mistralai.workflows import Depends

async def get_payment_client() -> PaymentServiceClient:
    client = PaymentServiceClient(api_key=os.environ["PAYMENT_API_KEY"])
    await client.initialize()
    return client

@workflows.activity()
async def charge(user_id: str, amount: int, client: PaymentServiceClient = Depends(get_payment_client)) -> str:
    return await client.charge(user_id=user_id, amount_cents=amount)
```

--------------------------------

### Configure custom metrics endpoint

Source: https://docs.mistral.ai/studio-api/workflows/observability

Routes metrics to a custom OTLP endpoint while keeping logs and traces in Mistral. Requires signal-specific headers for authentication.

```bash
OTEL_METRICS_ENDPOINT=https://otel.example.com
OTEL_EXPORTER_OTLP_METRICS_HEADERS=Authorization=Bearer <your-token>
```

--------------------------------

### Define a Vibe-compatible workflow in Python

Source: https://docs.mistral.ai/studio-api/workflows/interacting-with-workflows/conversational_workflows/publish_in_vibe

Use the ChatAssistantWorkflowOutput class to ensure your workflow is compatible with Vibe Work.

```python
import mistralai.workflows as workflows
import mistralai.workflows.plugins.mistralai as workflows_mistralai

@workflows.workflow.define(
    name="expense-summary-workflow",
    workflow_display_name="Expense Summary",
    workflow_description="Generates a summary of expenses",
)
class ExpenseSummaryWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, department: str) -> workflows_mistralai.ChatAssistantWorkflowOutput:
        summary = f"Expense summary for {department}: Total $12,500"

        return workflows_mistralai.ChatAssistantWorkflowOutput(
            content=[workflows_mistralai.TextOutput(text=summary)]
        )
```

--------------------------------

### Implement ConfirmationInput in Python

Source: https://docs.mistral.ai/studio-api/workflows/interacting-with-workflows/conversational_workflows/forms_and_confirmations

ConfirmationInput renders a list of options as buttons that trigger an immediate form submission upon selection.

```python
import mistralai.workflows as workflows
import mistralai.workflows.plugins.mistralai as workflows_mistralai

@workflows.workflow.define(
    name="type-selection-workflow",
    workflow_display_name="Type Selection",
    workflow_description="Select your favorite type",
)
class TypeSelectionWorkflow(workflows.InteractiveWorkflow):
    @workflows.workflow.entrypoint
    async def run(self) -> workflows_mistralai.ChatAssistantWorkflowOutput:
        await workflows_mistralai.send_assistant_message("Let's find out your type preference!")

        selection = await self.wait_for_input(
            workflows_mistralai.ConfirmationInput(
                options=[
                    ("fire", "Fire"),
                    ("water", "Water"),
                    ("grass", "Grass"),
                    ("electric", "Electric"),
                ],
                description="What is your favorite type?",
            )
        )

        selected_type = selection.choice  # Returns the value, e.g., "fire"
        return workflows_mistralai.ChatAssistantWorkflowOutput(
            content=[workflows_mistralai.TextOutput(text=f"You selected {selected_type}!")]
        )
```

--------------------------------

### Configure worker environment variables

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/workers

Set the API key and deployment name to connect the worker to the correct workspace and task queue.

```bash
MISTRAL_API_KEY=your_key DEPLOYMENT_NAME=invoice-service uv run python my_worker.py
```

--------------------------------

### Schedule a workflow with calendar specifications

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/scheduling

Create a schedule using calendar fields to define specific execution times.

```bash
curl -X POST "https://api.mistral.ai/v1/workflows/schedules" \
  -H "Authorization: Bearer $MISTRAL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_identifier": "report-workflow",
    "deployment_name": "production",
    "schedule": {
      "input": {},
      "calendars": [
        {
          "hour": [{"start": 9}],
          "minute": [{"start": 30}],
          "day_of_week": [{"start": 1, "end": 5}]
        }
      ]
    }
  }'
```

--------------------------------

### Execute child workflows

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/workflows

Demonstrates executing child workflows with varying wait and termination policies.

```python
from datetime import timedelta
from mistralai.workflows import workflow

result = await workflow.execute_workflow(
    ChildWorkflow,
    params=child_params,
    execution_timeout=timedelta(hours=1)
)
```

```python
handle = await workflow.execute_workflow(
    ChildWorkflow,
    params=child_params,
    execution_timeout=timedelta(hours=1),
    wait=False,
)
# Parent continues immediately; child runs independently
# Optionally await later: result = await handle
```

```python
from mistralai.workflows import ParentClosePolicy

handle = await workflow.execute_workflow(
    ChildWorkflow,
    params=child_params,
    execution_timeout=timedelta(hours=1),
    wait=False,
    parent_close_policy=ParentClosePolicy.TERMINATE,
)
```

--------------------------------

### Configure SSE MCP Server

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/durable_agents

Connect to a remote MCP server using Server-Sent Events with MCPSSEConfig.

```python
from mistralai.workflows.plugins.mistralai import MCPSSEConfig

mcp_config = MCPSSEConfig(
    url="https://your-mcp-server.com/sse",
    timeout=60,
    name="remote-tools",
    headers={"Authorization": "Bearer your-token"},  # Optional
)

agent = workflows_mistralai.Agent(
    model="mistral-medium-latest",
    name="sse-mcp-agent",
    description="Agent with access to remote MCP tools",
    mcp_clients=[mcp_config],
)
```

--------------------------------

### Resume event stream from sequence

Source: https://docs.mistral.ai/studio-api/workflows/building-workflows/consuming_events

Use the start_seq parameter to resume event consumption after a connection drop, ensuring no events are missed.

```python
last_seq = 0
while True:
    try:
        async for event in client.workflows.events.get_stream_events(
            workflow_exec_id=execution_id,
            start_seq=last_seq,
        ):
            last_seq = event.broker_sequence + 1  # advance past this event
            process(event)
        break  # Completed normally
    except ConnectionError:
        await asyncio.sleep(1)
        # Loop will resume from last_seq
```

--------------------------------

### Use deterministic SDK replacements

Source: https://docs.mistral.ai/studio-api/workflows/getting-started/core_concepts/workflows

Replace standard non-deterministic Python functions with SDK-provided alternatives to ensure replay consistency.

```python
from mistralai.workflows import workflow

current_time = workflow.now()    # not datetime.now()
request_id   = workflow.uuid4()  # not uuid.uuid4()
rand_value   = workflow.random() # not random.random()
```
Tokens:
Raw