Workflows

Workflows Documentation

This skill provides comprehensive documentation and guidance for the Mistral Workflows framework, which is designed for building durable, fault-tolerant workflows with orchestrated activities.

About Workflows

Mistral Workflows is a durable-execution orchestration platform that accelerates the development and reliable execution of complex, AI-driven workflows. It combines a user-friendly API with a rich Python framework (mistralai.workflows) optimized for Mistral's AI services, providing fault-tolerant execution, automatic retries, and durable state that survives crashes — from simple sequences to long-running, stateful processes (seconds to years).

SDK Version and Imports

This skill targets the mistralai-workflows SDK v3.4.0 and higher:

uv add "mistralai-workflows>=3.4.0,<4.0.0"

Canonical import style used throughout this skill:

import mistralai.workflows as workflows

@workflows.workflow.define(name="my_workflow")
class MyWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, data: str) -> str:
        ...

Focused snippets may use from mistralai.workflows import workflow, activity, ... instead.

Always use the SDK; never import temporalio directly. The SDK re-exports everything user code needs so workflows stay portable and deterministic:

• workflow.now(), workflow.uuid4(), workflow.random() — deterministic replacements for datetime.now(), uuid.uuid4(), random inside workflows
• workflow.wait_condition(), workflow.continue_as_new(), workflow.execute_workflow()
• workflow.unsafe.imports_passed_through() / workflow.unsafe.skip_determinism_enforcement() — sandbox escape hatches
• activity_heartbeat() — heartbeat from inside an activity
• WorkflowError, ActivityError, ParentClosePolicy — all from mistralai.workflows

Call activities directly (await my_activity(args)); timeouts and retries live on the @activity(...) decorator, not at the call site.

Documentation Structure

The documentation is organized into several categories:

Getting Started

• Introduction: Overview of Mistral Workflows and its core architecture
• Installation: Guide to installing and setting up the Workflows framework (CLI scaffolding, optional deps)
• Core Concepts: Workflows, activities, workers, executions vs runs
• Python SDK: Programmatic API via the mistralai client (client.workflows.*)
• Your First Workflow: Step-by-step guide to creating your first workflow

Guides

• Workflows: Creating workflows, determinism enforcement (sandbox), input types, timeouts, signals/queries/updates, child workflows, continue-as-new
• Activities: Timeouts, retries, heartbeats, local activities, sticky sessions, nested activities
• Workflows Exception Handling: WorkflowsException, ErrorCode enum, factory methods
• Error Codes: API error codes WF_1000-WF_1600 with HTTP status, description, and resolution
• Signals, Queries, and Updates: Workflow communication patterns with input validation
• Scheduling: Cron expressions, ScheduleDefinition, SchedulePolicy, overlap handling
• Dependency Injection: FastAPI-style Depends() pattern
• Streaming: Task API, token streaming, progress updates
• Streaming Consumption: client.workflows.events.get_stream_events_async(), NATS subjects, SSE API
• Concurrency: execute_activities_in_parallel() with List/Chain/Offset executors
• Rate Limiting: Distributed rate limiting across workers
• Handling Large Data: OffloadableField, blob storage (S3/Azure/GCS)
• Payload Encoding: Payload offloading, AES-GCM encryption, key rotation
• Observability: OpenTelemetry traces, trace sampling
• Durable Agents: Agent, Runner, RemoteSession/LocalSession, MCP, multi-agent handoffs
• Conversational Workflows: InteractiveWorkflow, HITL, ChatInput/FormInput, Canvas editing, Rich UI components, Tool UI states
• Local Execution: No-infra dev mode with Pydantic model params
• Limitations: System constraints and limits
• Workflows Plugins: Mistral AI plugin, Webhook plugin, Nuage plugin, custom plugins
• Deployment Patterns: Best practices for deploying workflows
• Migration v2 to v3: Breaking changes and upgrade steps from SDK v2 through v3.4.0

Testing

• Testing Workflows: Integration testing with create_test_worker, hang prevention, sandbox pitfalls
• Diagnostics: Run wf-diagnose locally or on Kubernetes to collect a diagnostic report for support triage

Quick-test script — run any workflow in a local test environment with zero setup:

python .agents/skills/workflows/scripts/test_workflow.py <workflow_file> --input '{"key": "value"}' [--timeout 30]

Timeout policy for testing: Use aggressive (short) timeouts to keep the feedback loop tight. A hanging test wastes more time than a false timeout. Defaults:

If a workflow is known to be long-running (e.g. multi-step agent, large data processing), increase timeouts proportionally — but start short and only raise them when you see legitimate timeout failures, not preemptively.

Internal References

These are additional patterns and utilities not covered in the official docs:

• Execution IDs: Generate deterministic execution IDs for child workflows
• Pipeline Pattern: Build multi-step workflows with declarative StepSpec definitions
• Workflow Testing: Ensure workflow classes are properly registered in workers

When to Use This Skill

Use this skill when you need to:

1. Build durable workflows: Create long-running, fault-tolerant processes
2. Orchestrate activities: Coordinate multiple tasks and operations
3. Handle background jobs: Manage asynchronous processing and task queues
4. Create multi-step pipelines: Build complex workflows with multiple stages
5. Schedule tasks: Set up recurring or delayed execution of workflows
6. Develop LLM agents: Build durable AI agents with MCP tool support
7. Build conversational workflows: Create interactive workflows with HITL, forms, canvas, and rich UI
8. Ensure fault tolerance: Implement systems that can recover from failures automatically
9. Stream events: Real-time token streaming and progress updates via NATS

Key Features

• Fault tolerance: Automatic recovery from failures and retries
• Durable execution: Workflows can run for extended periods (seconds to years)
• Determinism enforcement: Sandbox-based determinism, enabled by default (opt out per-workflow with enforce_determinism=False)
• Rich Python framework: Easy-to-use decorators and APIs (mistralai.workflows)
• Built-in observability: Deep integration with OpenTelemetry for monitoring
• Streaming: NATS-backed real-time token and progress streaming
• Rate limiting: Distributed rate limiting shared across workers
• Dependency injection: FastAPI-style Depends() pattern
• Large payload handling: OffloadableField with S3/Azure/GCS blob storage
• Conversational workflows: Interactive workflows with Vibe integration, forms, canvas, and rich UI components
• Durable agents: AI agents with MCP support, multi-agent handoffs, and persistent state
• Scalability: Designed to handle complex, distributed applications