Queries

Queries allow external systems to read the current state of a running workflow synchronously.

Queries use synchronous communication and are read-only — they should not modify workflow state. They can be called at any time during execution but must return quickly, making them unsuitable for long-running operations.

The workflow below tracks its own progress as it runs. The get_status query handler reads that state and returns it immediately to the caller — without interrupting or modifying the running workflow.

import mistralai.workflows as workflows

@workflows.workflow.define(name="processing_workflow")
class ProcessingWorkflow:
    def __init__(self):
        self.progress = 0.0
        self.completed = False

    @workflows.workflow.query(name="get_status")
    def get_status(self) -> dict:
        return {
            "progress": self.progress,
            "completed": self.completed
        }

    @workflows.workflow.entrypoint
    async def run(self) -> None:
        # Simulate work
        for i in range(1, 11):
            self.progress = i * 10
            await asyncio.sleep(1)
        self.completed = True

import mistralai.workflows as workflows

@workflows.workflow.define(name="processing_workflow")
class ProcessingWorkflow:
    def __init__(self):
        self.progress = 0.0
        self.completed = False

    @workflows.workflow.query(name="get_status")
    def get_status(self) -> dict:
        return {
            "progress": self.progress,
            "completed": self.completed
        }

    @workflows.workflow.entrypoint
    async def run(self) -> None:
        # Simulate work
        for i in range(1, 11):
            self.progress = i * 10
            await asyncio.sleep(1)
        self.completed = True

Query handlers can accept parameters just like any other handler, and the SDK validates the payload before invoking the handler. Queries validate incoming payloads against their declared parameters. Incoming data is checked against the expected types, and any extra fields not declared in the handler signature are rejected. Validation failures return HTTP 422 (Unprocessable Entity) with a descriptive error message.

For complex input structures, use Pydantic models. This lets you pass structured parameters to a query handler while keeping full type safety:

import pydantic

class StatusRequest(pydantic.BaseModel):
    include_details: bool = False

@workflows.workflow.query(name="get_status")
def get_status(self, req: StatusRequest) -> dict:
    result = {"progress": self.progress}
    if req.include_details:
        result["steps"] = self.completed_steps
    return result

import pydantic

class StatusRequest(pydantic.BaseModel):
    include_details: bool = False

@workflows.workflow.query(name="get_status")
def get_status(self, req: StatusRequest) -> dict:
    result = {"progress": self.progress}
    if req.include_details:
        result["steps"] = self.completed_steps
    return result

Once your workflow is running, you can query its state at any time from the outside using the SDK or the API.

from mistralai.client import Mistral

client = Mistral(api_key="your_api_key")

result = client.workflows.executions.query_workflow_execution(
    execution_id="my-execution-id",
    name="get_status",
)
print(result.model_dump_json(indent=2))

from mistralai.client import Mistral

client = Mistral(api_key="your_api_key")

result = client.workflows.executions.query_workflow_execution(
    execution_id="my-execution-id",
    name="get_status",
)
print(result.model_dump_json(indent=2))