Queries

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

Key characteristics

Synchronous communication
Read-only operations (should not modify workflow state)
Can be called at any time during workflow execution
Must return quickly (not for long-running operations)

Basic example

import mistralai.workflows as workflows

@workflows.workflow.define()
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):
        # 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()
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):
        # Simulate work
        for i in range(1, 11):
            self.progress = i * 10
            await asyncio.sleep(1)
        self.completed = True

Input validation

Queries validate incoming payloads against their declared parameters. The system automatically:

Validates incoming payloads match the expected types
Rejects extra fields not declared in the handler signature
Returns HTTP 422 (Unprocessable Entity) with descriptive error messages on validation failure

For complex input structures, use Pydantic models:

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

Comparison

Feature	Communication Type	Modifies State	Returns Value	Can Execute Activities
Signal	Asynchronous	Yes	No	No
Query	Synchronous	No	Yes	No
Update	Synchronous	Yes	Yes	Yes