Waiting for Conditions

Workflows can pause execution until a specific condition is met, enabling event-driven patterns like human-in-the-loop and approval flows.

Basic usage

Use workflow.wait_condition() to block until a predicate returns True:

from datetime import timedelta
from mistralai.workflows import workflow

await workflow.wait_condition(
    lambda: self.ready,
    timeout=timedelta(minutes=5)
)

from datetime import timedelta
from mistralai.workflows import workflow

await workflow.wait_condition(
    lambda: self.ready,
    timeout=timedelta(minutes=5)
)

Example: approval flow

Combine with signals to implement a human-in-the-loop pattern:

import mistralai.workflows as workflows
from mistralai.workflows import workflow
from datetime import timedelta

@workflows.workflow.define(name="approval_workflow")
class ApprovalWorkflow:
    def __init__(self):
        self.approved = False

    @workflows.workflow.signal(name="approve")
    async def approve(self):
        self.approved = True

    @workflows.workflow.entrypoint
    async def run(self, request_id: str) -> str:
        # Wait up to 24 hours for approval
        await workflow.wait_condition(
            lambda: self.approved,
            timeout=timedelta(hours=24)
        )

        if not self.approved:
            return f"Request {request_id} timed out"

        return f"Request {request_id} approved"

import mistralai.workflows as workflows
from mistralai.workflows import workflow
from datetime import timedelta

@workflows.workflow.define(name="approval_workflow")
class ApprovalWorkflow:
    def __init__(self):
        self.approved = False

    @workflows.workflow.signal(name="approve")
    async def approve(self):
        self.approved = True

    @workflows.workflow.entrypoint
    async def run(self, request_id: str) -> str:
        # Wait up to 24 hours for approval
        await workflow.wait_condition(
            lambda: self.approved,
            timeout=timedelta(hours=24)
        )

        if not self.approved:
            return f"Request {request_id} timed out"

        return f"Request {request_id} approved"

Timeout behavior

When a timeout is specified, wait_condition raises a asyncio.TimeoutError if the condition is not met within the duration. Handle this to implement timeout logic:

import asyncio

try:
    await workflow.wait_condition(
        lambda: self.ready,
        timeout=timedelta(minutes=30)
    )
except asyncio.TimeoutError:
    # Handle timeout
    return "Operation timed out"

import asyncio

try:
    await workflow.wait_condition(
        lambda: self.ready,
        timeout=timedelta(minutes=30)
    )
except asyncio.TimeoutError:
    # Handle timeout
    return "Operation timed out"