Planification des workflows

Exécutez des workflows selon un calendrier cron, un intervalle fixe ou un agenda. Les planifications persistent après un redémarrage des workers et sont gérées indépendamment du code de vos workers.

Note

Les planifications étaient auparavant définies avec le décorateur @workflow.define(schedules=[...]). Cette approche est obsolète. Pour plus de détails et des instructions de migration, consultez Obsolète : Planifications basées sur les décorateurs.

Fonctionnement

Mistral AI Workflows stockent les planifications et déclenchent les exécutions de workflow aux horaires prévus. Les politiques de chevauchement (overlap) et de rattrapage (catchup) contrôlent ce qui se passe lorsque des exécutions sont manquées ou se chevauchent. Vous pouvez créer, lister et supprimer des planifications à tout moment, sans redémarrer vos workers.

Créer une planification

Pour créer une planification, transmettez un workflow_identifier et une ScheduleDefinition. Le workflow_identifier doit correspondre au name utilisé dans @workflow.define().

from mistralai.client import Mistral, models

client = Mistral()

response = await client.workflows.schedules.schedule_workflow(
    workflow_identifier="report-workflow",
    deployment_name="production",
    schedule=models.ScheduleDefinition(
        input={"report_type": "daily"},
        calendars=[
            models.ScheduleCalendar(
                hour=[models.ScheduleRange(start=9)],
                minute=[models.ScheduleRange(start=0)],
            )
        ],  # daily at 9:00 AM UTC
    ),
)
print(response.schedule_id)

from mistralai.client import Mistral, models

client = Mistral()

response = await client.workflows.schedules.schedule_workflow(
    workflow_identifier="report-workflow",
    deployment_name="production",
    schedule=models.ScheduleDefinition(
        input={"report_type": "daily"},
        calendars=[
            models.ScheduleCalendar(
                hour=[models.ScheduleRange(start=9)],
                minute=[models.ScheduleRange(start=0)],
            )
        ],  # daily at 9:00 AM UTC
    ),
)
print(response.schedule_id)

response.schedule_id est l'identifiant de la planification créée. Utilisez-le pour obtenir, mettre à jour, mettre en pause, reprendre et supprimer la planification — comme indiqué dans la section Gestion des planifications.

workflow_identifier : le nom du workflow issu de @workflow.define(name=...), ou son identifiant d'enregistrement.

deployment_name : facultatif. Requis s'il existe plusieurs déploiements nommés pour le workflow. Achemine la planification vers un déploiement spécifique. Si omis, la plateforme utilise le déploiement par défaut, mais retourne une erreur si le workflow en compte plusieurs. Voir Déploiements pour plus de détails.

Spécifications des planifications

Pour définir une planification, utilisez une ScheduleDefinition. Une ScheduleDefinition peut combiner des calendars, des intervals et des cron_expressions dans n'importe quelle ordre. Une exécution est déclenchée dès qu'une des définitions correspond à la condition actuelle. Utilisez skip pour exclure des horaires spécifiques du résultat.

Calendriers

Les calendriers constituent le moyen recommandé pour définir des planifications. Il s'agit de la représentation native de Mistral Workflows, et les expressions cron sont converties en calendriers côté serveur. Les calendriers prennent également en charge des plages et des pas que les expressions cron ne peuvent pas exprimer proprement.

Un ScheduleCalendar correspond quand tous ses champs correspondent simultanément. Chaque champ accepte une liste de valeurs ScheduleRange(start, end, step) :

ScheduleRange(start=9) : exactement 9
ScheduleRange(start=1, end=5) : de 1 à 5 inclus
ScheduleRange(start=0, end=59, step=15) : 0, 15, 30, 45

Champs et plages associés : second (0 à 59), minute (0 à 59), hour (0 à 23), day_of_month (1 à 31), month (1 à 12), year, day_of_week (0 à 6, où 0 correspond à dimanche).

L'exemple suivant montre un ScheduleCalendar qui s'exécute du lundi au vendredi à 9h30 UTC :

schedule=models.ScheduleDefinition(
    input={},
    calendars=[
        models.ScheduleCalendar(
            hour=[models.ScheduleRange(start=9)],        # 9 AM
            minute=[models.ScheduleRange(start=30)],     # at :30
            day_of_week=[models.ScheduleRange(start=1, end=5)],  # Mon-Fri (1=Mon, 5=Fri)
        )
    ],
)

schedule=models.ScheduleDefinition(
    input={},
    calendars=[
        models.ScheduleCalendar(
            hour=[models.ScheduleRange(start=9)],        # 9 AM
            minute=[models.ScheduleRange(start=30)],     # at :30
            day_of_week=[models.ScheduleRange(start=1, end=5)],  # Mon-Fri (1=Mon, 5=Fri)
        )
    ],
)

Intervalles

Pour déclencher une exécution à intervalles fixes répétés, utilisez intervals. Les intervalles sont définis avec une chaîne de durée ISO 8601 pour every et un offset optionnel pour décaler le début dans chaque période.

Chaînes de durée courantes : PT30M (30 minutes), PT1H (1 heure), PT4H (4 heures), P1D (1 jour), PT2H30M (2,5 heures).

L'exemple suivant utilise every="PT4H" et offset="PT30M" pour décaler chaque exécution de 30 minutes dans la fenêtre. Il se déclenche à 00:30, 04:30, 08:30, 12:30, 16:30 et 20:30 UTC. Sans offset, les exécutions se déclenchent à la limite de l'intervalle, par exemple à 00:00, 04:00 et 08:00.

schedule=models.ScheduleDefinition(
    input={},
    intervals=[
        models.ScheduleInterval(
            every="PT4H",    # repeat every 4 hours
            offset="PT30M", # fire 30 min after each 4-hour boundary
        )
    ],
)

schedule=models.ScheduleDefinition(
    input={},
    intervals=[
        models.ScheduleInterval(
            every="PT4H",    # repeat every 4 hours
            offset="PT30M", # fire 30 min after each 4-hour boundary
        )
    ],
)

Expressions cron

Le paramètre cron_expressions utilise la syntaxe standard à 5 champs commune aux cron jobs : minute hour day-of-month month day-of-week. Toutes les heures sont en UTC sauf si vous définissez time_zone_name.

Note

Les expressions cron sont converties en spécifications de calendrier côté serveur. Utilisez calendars directement lorsque vous avez besoin de plages, de pas ou de plusieurs valeurs que la syntaxe cron ne permet pas d'exprimer proprement.

L'exemple suivant utilise cron_expressions pour exécuter un workflow du lundi au vendredi à 9h00 UTC :

schedule=models.ScheduleDefinition(
    input={},
    cron_expressions=["0 9 * * 1-5"],  # 9 AM UTC, Monday-Friday
)

schedule=models.ScheduleDefinition(
    input={},
    cron_expressions=["0 9 * * 1-5"],  # 9 AM UTC, Monday-Friday
)

cron_expressions, calendars et intervals sont combinés. Chaque spécification active déclenche une exécution indépendamment.

Délimitation et exclusion

Pour restreindre une planification à une fenêtre temporelle, utilisez start_at et end_at. Pour exclure des heures spécifiques du déclenchement, utilisez skip (même structure ScheduleCalendar).

from datetime import datetime, timezone

schedule=models.ScheduleDefinition(
    input={},
    calendars=[
        models.ScheduleCalendar(
            hour=[models.ScheduleRange(start=9)],        # 9 AM
            minute=[models.ScheduleRange(start=30)],     # at :30
        )
    ],
    start_at=datetime(2026, 1, 1, 9, 30, tzinfo=timezone.utc),   # first run: Jan 1 2026 at 09:30 UTC
    end_at=datetime(2026, 12, 31, 9, 30, tzinfo=timezone.utc),   # last run: Dec 31 2026 at 09:30 UTC
    skip=[
        # skip December 25th
        models.ScheduleCalendar(
            month=[models.ScheduleRange(start=12)],
            day_of_month=[models.ScheduleRange(start=25)],
        )
    ],
)

from datetime import datetime, timezone

schedule=models.ScheduleDefinition(
    input={},
    calendars=[
        models.ScheduleCalendar(
            hour=[models.ScheduleRange(start=9)],        # 9 AM
            minute=[models.ScheduleRange(start=30)],     # at :30
        )
    ],
    start_at=datetime(2026, 1, 1, 9, 30, tzinfo=timezone.utc),   # first run: Jan 1 2026 at 09:30 UTC
    end_at=datetime(2026, 12, 31, 9, 30, tzinfo=timezone.utc),   # last run: Dec 31 2026 at 09:30 UTC
    skip=[
        # skip December 25th
        models.ScheduleCalendar(
            month=[models.ScheduleRange(start=12)],
            day_of_month=[models.ScheduleRange(start=25)],
        )
    ],
)

Politiques de planification

Politique de chevauchement

La politique de chevauchement contrôle ce qui se passe lorsqu'une nouvelle exécution est prévue alors que l'exécution précédente est encore active.

Pour définir la politique, transmettez une valeur overlap de 1 à 6 selon les comportements suivants :

Valeur	Nom	Comportement
`1`	`SKIP`	Ignore la nouvelle exécution. Par défaut. À utiliser pour les tâches de synchronisation où seules les dernières données comptent.
`2`	`BUFFER_ONE`	Met en file d'attente une seule exécution en attente. Les autres exécutions sont ignorées jusqu'au démarrage de celle-ci. Garantit une seule exécution de suivi, sans créer d'arriéré.
`3`	`BUFFER_ALL`	Met en file d'attente toutes les exécutions en attente. Peut créer un arriéré illimité. Utilisez cette option avec prudence.
`4`	`CANCEL_OTHER`	Annule correctement l'exécution en cours et lance la nouvelle.
`5`	`TERMINATE_OTHER`	Termine de force l'exécution en cours et lance la nouvelle.
`6`	`ALLOW_ALL`	Démarre toutes les exécutions simultanément. À utiliser uniquement lorsque les exécutions sont entièrement indépendantes et que le système peut absorber le parallélisme.

schedule=models.ScheduleDefinition(
    input={"source": "postgres"},
    calendars=[
        models.ScheduleCalendar(
            hour=[models.ScheduleRange(start=8, end=18)],  # 8 AM-6 PM UTC
            minute=[models.ScheduleRange(start=0)],
        )
    ],
    policy=models.SchedulePolicy(
        overlap=models.ScheduleOverlapPolicy.BUFFER_ONE,
    ),
)

schedule=models.ScheduleDefinition(
    input={"source": "postgres"},
    calendars=[
        models.ScheduleCalendar(
            hour=[models.ScheduleRange(start=8, end=18)],  # 8 AM-6 PM UTC
            minute=[models.ScheduleRange(start=0)],
        )
    ],
    policy=models.SchedulePolicy(
        overlap=models.ScheduleOverlapPolicy.BUFFER_ONE,
    ),
)

Fenêtre de rattrapage

Pour contrôler la durée pendant laquelle les exécutions manquées sont déclenchées rétroactivement après une panne, utilisez catchup_window_seconds. Les exécutions planifiées avant cette fenêtre sont ignorées. Par défaut, 31536000 (365 jours). Modifiez cette valeur en fonction des besoins de votre cas d'utilisation.

policy=models.SchedulePolicy(
    catchup_window_seconds=86400,
)

policy=models.SchedulePolicy(
    catchup_window_seconds=86400,
)

Pause en cas d'échec

Pour mettre automatiquement la planification en pause lorsqu'une exécution de workflow déclenchée échoue, définissez pause_on_failure=True. La planification reste en pause jusqu'à ce qu'elle soit reprise manuellement.

policy=models.SchedulePolicy(
    pause_on_failure=True,
)

policy=models.SchedulePolicy(
    pause_on_failure=True,
)

Dispersion aléatoire

jitter ajoute un délai aléatoire entre 0 et la durée spécifiée à chaque exécution planifiée. Utilisez une chaîne de durée ISO 8601. Utilisez ce paramètre pour répartir la charge lorsque de nombreuses planifications seraient autrement synchronisées, par exemple des centaines de planifications horaires déclenchées toutes à :00. Le délai est borné : il ne fait pas passer une exécution après l'heure planifiée suivante.

L'exemple suivant s'exécute toutes les heures, avec une fenêtre de dispersion de 5 minutes, ce qui signifie que les exécutions seront retardées aléatoirement entre l'heure pile et jusqu'à 5 minutes après.

schedule=models.ScheduleDefinition(
    input={"tenant_id": "acme"},
    intervals=[
        models.ScheduleInterval(every="PT1H"),  # every hour
    ],
    jitter="PT5M",  # actual fire time is random within [scheduled_time, scheduled_time + 5 min)
)

schedule=models.ScheduleDefinition(
    input={"tenant_id": "acme"},
    intervals=[
        models.ScheduleInterval(every="PT1H"),  # every hour
    ],
    jitter="PT5M",  # actual fire time is random within [scheduled_time, scheduled_time + 5 min)
)

Nombre maximal d'exécutions

Pour arrêter automatiquement la planification après un nombre fixe d'exécutions, définissez max_executions. Une fois la limite atteinte, aucune exécution supplémentaire n'est déclenchée. null (par défaut) signifie que les exécutions sont illimitées et que la planification se poursuit indéfiniment.

L'exemple suivant s'exécute chaque lundi pendant 4 semaines, puis s'arrête :

schedule=models.ScheduleDefinition(
    input={},
    cron_expressions=["0 9 * * 1"],  # every Monday
    max_executions=4,                # run for 4 weeks then stop
)

schedule=models.ScheduleDefinition(
    input={},
    cron_expressions=["0 9 * * 1"],  # every Monday
    max_executions=4,                # run for 4 weeks then stop
)

Gestion des planifications

Note

Pour consulter les schémas complets des requêtes et réponses pour toutes les opérations de planification, voir la référence de l'API.

Liste des planifications

L'exemple suivant liste toutes les planifications de tous les workflows actifs dans l'espace de travail :

response = client.workflows.schedules.get_schedules()
for s in response.schedules:
    print(s.schedule_id, s.workflow_name)

response = client.workflows.schedules.get_schedules()
for s in response.schedules:
    print(s.schedule_id, s.workflow_name)

Obtenir une planification

L'exemple suivant montre comment obtenir une planification par son identifiant. La réponse retourne la spécification complète de la planification.

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

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

Mettre à jour une planification

Mettez à jour l'horaire, la politique ou les données d'entrée d'une planification sans la supprimer et la recréer. Seuls les champs que vous incluez sont modifiés. Les autres conservent leurs valeurs existantes.

client.workflows.schedules.update_schedule(
    schedule_id="your-schedule-id",
    schedule=models.PartialScheduleDefinition(
        calendars=[
            models.ScheduleCalendar(
                hour=[models.ScheduleRange(start=10)],
                minute=[models.ScheduleRange(start=0)],
            )
        ],
    ),
)

client.workflows.schedules.update_schedule(
    schedule_id="your-schedule-id",
    schedule=models.PartialScheduleDefinition(
        calendars=[
            models.ScheduleCalendar(
                hour=[models.ScheduleRange(start=10)],
                minute=[models.ScheduleRange(start=0)],
            )
        ],
    ),
)

Pause et reprise

Une planification en pause arrête de déclencher des exécutions jusqu'à ce qu'elle soit explicitement reprise. Les deux méthodes acceptent une note note optionnelle, enregistrée avec l'état de pause ou de reprise.

# Pause
client.workflows.schedules.pause_schedule(
    schedule_id="your-schedule-id",
    note="Paused for maintenance",
)

# Resume
client.workflows.schedules.resume_schedule(
    schedule_id="your-schedule-id",
    note="Maintenance complete",
)

# Pause
client.workflows.schedules.pause_schedule(
    schedule_id="your-schedule-id",
    note="Paused for maintenance",
)

# Resume
client.workflows.schedules.resume_schedule(
    schedule_id="your-schedule-id",
    note="Maintenance complete",
)

Supprimer une planification

client.workflows.schedules.unschedule_workflow(
    schedule_id="your-schedule-id",
)

client.workflows.schedules.unschedule_workflow(
    schedule_id="your-schedule-id",
)

Déclencher une exécution manuellement

Déclenchez une exécution immédiate en dehors de la planification normale. Par défaut, la politique de chevauchement de la planification s'applique. Si elle est définie sur SKIP et qu'une exécution est déjà active, l'exécution déclenchée est ignorée. Transmettez overlap pour remplacer la politique pour ce déclenchement unique. Définissez-le sur ALLOW_ALL pour exécuter immédiatement quelle que soit l'exécution en cours.

client.workflows.schedules.trigger(
    schedule_id="your-schedule-id",
    overlap=models.ScheduleOverlapPolicy.ALLOW_ALL,
)

client.workflows.schedules.trigger(
    schedule_id="your-schedule-id",
    overlap=models.ScheduleOverlapPolicy.ALLOW_ALL,
)

Information

Vous pouvez également créer, modifier, supprimer, mettre en pause, reprendre et déclencher des planifications directement dans AI Studio.

Obsolète : planifications basées sur les décorateurs

Avertissement

L'argument schedules= du décorateur @workflow.define est obsolète. Utilisez plutôt client.workflows.schedules.schedule_workflow().

Deux problèmes rendent cette approche fragile :

Les modifications des planifications nécessitent un redémarrage du worker. Le worker lit les définitions de planification une seule fois au démarrage. Les mises à jour des expressions cron ou des politiques ne sont pas prises en compte par un worker en cours d'exécution.
Conflits multi-workers. Chaque instance de worker enregistre sa propre copie des planifications intégrées au code. Si deux workers pour le même workflow exécutent des configurations ScheduleDefinition différentes, la plateforme voit des enregistrements contradictoires et le comportement est indéfini.

Pour migrer, appelez client.workflows.schedules.schedule_workflow() en dehors de votre worker, par exemple depuis un script de déploiement ou la console de pilotage. Supprimez ensuite l'argument schedules=[...] de @workflow.define.

import mistralai.workflows as workflows
from mistralai.workflows.models import ScheduleDefinition, SchedulePolicy, ScheduleOverlapPolicy

schedule = ScheduleDefinition(
    input={"report_type": "daily"},
    cron_expressions=["0 9 * * *"],
    policy=SchedulePolicy(
        catchup_window_seconds=86400,
        overlap=ScheduleOverlapPolicy.SKIP,
    )
)

@workflows.workflow.define(name="report_workflow", schedules=[schedule])
class ReportWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, report_type: str = "daily") -> None:
        pass

import mistralai.workflows as workflows
from mistralai.workflows.models import ScheduleDefinition, SchedulePolicy, ScheduleOverlapPolicy

schedule = ScheduleDefinition(
    input={"report_type": "daily"},
    cron_expressions=["0 9 * * *"],
    policy=SchedulePolicy(
        catchup_window_seconds=86400,
        overlap=ScheduleOverlapPolicy.SKIP,
    )
)

@workflows.workflow.define(name="report_workflow", schedules=[schedule])
class ReportWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, report_type: str = "daily") -> None:
        pass

Notes

Les expressions cron suivent la syntaxe standard à 5 champs (minute hour day-of-month month day-of-week).
Les planifications utilisent UTC par défaut. Définissez time_zone_name sur un nom de fuseau horaire IANA (par exemple, "Europe/Paris") pour utiliser un fuseau horaire local à la place.
Chaque planification transporte sa propre charge utile input, directement transmise au point d'entrée du workflow.