temporal2mistral

Move Python code from the Temporal SDK to Mistral Workflows.

It rewrites what can be rewritten safely and gives you a report of the rest, file by file and line by line.

$ temporal2mistral check ./src           # report only, writes nothing
$ temporal2mistral apply ./src --diff    # show the rewrite
$ temporal2mistral apply ./src           # rewrite in place (commit first)

Alpha. Tested against temporalio 1.29 and mistralai-workflows 3.15. Mistral Workflows is in public preview, so its API can still move. This project is not affiliated with Temporal Technologies or Mistral AI.

Why not search and replace?

Mistral Workflows runs on Temporal, so most of the API maps one to one. A few things don't:

• Temporal sets activity timeouts and retries at each call (workflow.execute_activity(fn, x, start_to_close_timeout=...)). Mistral sets them once on the definition (@activity(start_to_close_timeout=...)) and you call the activity directly (await fn(x)). The call is usually in a different module from the definition, so the tool scans the whole project before rewriting anything.
• Mistral builds a JSON schema from every signature. Parameters and return types must be annotated, activities must be module-level async def functions, query and update handlers must return something, and child workflows and continue_as_new take a single pydantic model.
• The defaults differ. A Mistral workflow is cancelled after 1 hour unless you set execution_timeout (Temporal has no limit), and activities retry 3 times (Temporal retries forever).

Install

pip install temporal2mistral
pip install "temporal2mistral[mistral]"      # with the Mistral SDK, for --import-check

Needs Python 3.12 or later, like the Mistral SDK. From a clone: pip install -e ".[test,mistral]".

Example

Before:

@activity.defn
def compose_greeting(input: ComposeGreetingInput) -> str:
    activity.logger.info("Running activity with parameter %s" % input)
    return f"{input.greeting}, {input.name}!"


@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        workflow.logger.info("Running workflow with parameter %s" % name)
        return await workflow.execute_activity(
            compose_greeting,
            ComposeGreetingInput("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

After temporal2mistral apply:

def _compose_greeting_sync(input: ComposeGreetingInput) -> str:
    temporal_activity.logger.info("Running activity with parameter %s" % input)
    return f"{input.greeting}, {input.name}!"


@activity(start_to_close_timeout=timedelta(seconds=10), retry_policy_max_attempts=0)
async def compose_greeting(input: ComposeGreetingInput) -> str:
    return await asyncio.to_thread(_compose_greeting_sync, input)


@workflow.define(name="GreetingWorkflow")
class GreetingWorkflow:
    @workflow.entrypoint
    async def run(self, name: str) -> str:
        temporal_workflow.logger.info("Running workflow with parameter %s" % name)
        return await compose_greeting(ComposeGreetingInput("Hello", name))

And temporal2mistral check hello.py prints:

hello.py
    14  WARNING T2M031  Synchronous activity `compose_greeting` wrapped in an `async def` that runs it in a thread
               -> The original body now runs in a thread through `asyncio.to_thread`, like Temporal's `activity_executor`.
    14  INFO    T2M020  start_to_close_timeout, retry_policy_max_attempts moved to `@activity` on `compose_greeting` (from hello.py:24)
    14  INFO    T2M026  `retry_policy_max_attempts=0` (unlimited, Temporal's default) set on `compose_greeting`
               -> Remove it to use Mistral's default of 3 attempts, or rerun with --mistral-retry-defaults.
    15  INFO    T2M060  `activity.logger` is used through `temporal_activity` (Mistral activities run on Temporal)
    19  WARNING T2M010  Workflow `GreetingWorkflow` had no execution timeout on Temporal; Mistral cancels it after 1 hour by default
               -> Pass `execution_timeout=timedelta(...)` to `@workflow.define`, or rerun with `--execution-timeout 30d`.
    23  INFO    T2M060  `workflow.logger` is used through `temporal_workflow` (Mistral workflows run on Temporal)
               -> Works under a Mistral worker; raises in the SDK's local mode (execute_workflow outside a worker).

1 file(s) converted out of 1 using temporalio (1 scanned): 0 error(s), 2 warning(s), 4 info(s).

The output is correct but the imports aren't sorted, so run ruff, black or isort on it afterwards.

What gets converted

Activity options

Timeouts and retries are collected from every call site in the project, including module-level constants like RETRY = RetryPolicy(maximum_attempts=3). If two call sites disagree, the tool keeps the most permissive value and tells you. If schedule_to_close_timeout is the only timeout, it becomes start_to_close_timeout. Temporal's unlimited retries become retry_policy_max_attempts=0, unless you pass --mistral-retry-defaults.

Workflow inputs

Child workflows and continue_as_new take one pydantic model on Mistral. If the target's entrypoint already takes a single pydantic model, the argument is passed through. Otherwise the tool writes a small model next to the caller:

class ComposeGreetingWorkflowParams(BaseModel):
    """Input of the `ComposeGreetingWorkflow` workflow (generated by temporal2mistral)."""

    input: Any

...
return await workflow.execute_workflow(
    ComposeGreetingWorkflow, params=ComposeGreetingWorkflowParams(input=ComposeGreetingInput("Hello", name))
)

The child still validates its input against its own annotations, so dataclasses keep working.

Workers

A worker becomes run_worker when it only runs until it is stopped, has no option that matters on Mistral (interceptors, plugins, tuning...), and is the only user of a client created next to it by Client.connect (that line is removed too, unless the client has a data_converter, plugins, interceptors or runtime). Activities register themselves, and task_queue is dropped because Mistral routes executions to deployments. Other workers are reported with T2M070.

What you still have to do

The tool reports these and leaves the code alone:

• Workers the tool cannot convert safely (see above). The report points at each one and gives you the run_worker call.
• Client code, the code that starts workflows and talks to them: see Migrating client code below.
• Class-based activities whose class needs arguments to be built, and activities defined inside a function. Mistral activities are module-level functions: use Depends(...) with a function that builds the shared client.
• @workflow.init, update validators, dynamic workflows and handlers, Nexus, asynchronous activity completion, and query or update handlers that return nothing.
• Types with no JSON schema, such as plain classes that relied on a custom Temporal payload converter.
• Execution timeouts. Every workflow gets a warning until you set execution_timeout, or you can pass --execution-timeout 30d to set it on all of them.
• Workflows already running on your Temporal cluster. They can't be moved: let them finish there and start new executions on Mistral.

Migrating client code

The code that starts workflows and sends them signals, queries and updates moves from the Temporal client to the Mistral API client. The tool does not rewrite it, because client code is usually spread across an application (created at startup, used in request handlers...), and Mistral returns results as JSON values instead of instances of the workflow's return type. Check what your code does with them.

Instead, check reports every call with the exact Mistral code to write (T2M071). Take this Temporal starter, starter.py:

from temporalio.client import Client

from workflows import GreetingWorkflow


async def main() -> None:
    client = await Client.connect("localhost:7233")
    handle = await client.start_workflow(GreetingWorkflow.run, "Louis", id="greet", task_queue="greetings")
    await handle.signal(GreetingWorkflow.approve, "Ana")
    status = await handle.query(GreetingWorkflow.status)
    print(status, await handle.result())

temporal2mistral check prints:

starter.py
     7  WARNING T2M071  Temporal client in use: starting, signalling and querying workflows goes through the Mistral API
               -> Create `client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])` (from mistralai.client import Mistral). Each call to change is reported with its translation; see https://github.com/louiscauret/temporal2mistral#migrating-client-code
     8  WARNING T2M071  `client.start_workflow(...)` goes through the Mistral API client
               -> `handle = await client.workflows.execute_workflow_async(workflow_identifier="GreetingWorkflow", input={"name": "Louis"}, execution_id="greet")`; then `await handle.result()` becomes `(await client.workflows.wait_for_workflow_completion_async(handle.execution_id)).result`
     9  WARNING T2M071  `handle.signal(...)` goes through the Mistral API client
               -> `await client.workflows.executions.signal_workflow_execution_async(execution_id=handle.execution_id, name="approve", input={"approver": "Ana"})`
    10  WARNING T2M071  `handle.query(...)` goes through the Mistral API client
               -> `(await client.workflows.executions.query_workflow_execution_async(execution_id=handle.execution_id, name="status")).result`

Mistral, written from those hints:

import os

from mistralai.client import Mistral


async def main() -> None:
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    handle = await client.workflows.execute_workflow_async(
        workflow_identifier="GreetingWorkflow", input={"name": "Louis"}, execution_id="greet"
    )
    await client.workflows.executions.signal_workflow_execution_async(
        execution_id=handle.execution_id, name="approve", input={"approver": "Ana"}
    )
    status = (
        await client.workflows.executions.query_workflow_execution_async(
            execution_id=handle.execution_id, name="status"
        )
    ).result
    final = await client.workflows.wait_for_workflow_completion_async(handle.execution_id)
    print(status, final.result)

The correspondence:

Three rules to get right:

• The workflow is named, not passed as a class: workflow_identifier is the name given to @workflow.define(name=...), which the tool sets to the Temporal workflow name.
• input is a dict of the parameters by name ({"name": "Louis"} for run(self, name: str)). When the workflow, signal, query or update takes a single pydantic model, the input is that model: input=order.model_dump(mode="json").
• Handlers are named too: name="approve" is the method name, or the name= given to @workflow.signal(...), @workflow.query(...) or @workflow.update(...).

task_queue has no equivalent: Mistral routes executions to deployments.

Options

temporal2mistral check PATH... [--fail-on error|warning|never]
temporal2mistral apply PATH... [--diff | --out DIR]

  --execution-timeout DURATION   set execution_timeout on every workflow (30d, 12h, 90m...)
  --sync-activities thread|async how to convert synchronous activities (default: thread)
  --mistral-retry-defaults       use Mistral's 3 attempts where Temporal retried forever
  --import-check                 import every converted module with the Mistral SDK (see below)
  --import-root DIR              directory put on sys.path for --import-check
  --exclude GLOB                 skip paths (repeatable)
  --min-severity error|warning|info
  --format text|json             --report FILE writes the report to a file

check exits with status 1 when it finds errors, so you can run it in CI:

temporal2mistral check src --format json --report migration-report.json

--import-check

Some problems only show up when the code runs: forward references in input types, third-party types without a schema, and so on. With --import-check, each converted module is imported in a subprocess, from a converted copy of the project, with mistralai-workflows installed. Mistral validates definitions at import time, so anything it rejects is reported as T2M090 with the SDK's own message, in the module where the error happens (often a workflow module imported by a worker). If the original module doesn't import either (a missing dependency, for example), it is skipped and reported as T2M091.

Be aware that this runs your project's top-level code.

Diagnostics

Errors mean the converted code won't work until someone fixes it. Warnings mean it works but may behave differently. Infos record what was changed.

Tested on temporalio/samples-python

I test against Temporal's official Python samples (commit b0e4d84, 9 October 2026): 734 Python files outside tests/.

The other 129 modules need optional integrations that weren't installed (OpenAI Agents, LangGraph, Strands, Google ADK, boto3...). They were converted but not import-checked. All 7 rejected modules already have error diagnostics: one uses a plain class as input (T2M006), one has an update handler that returns None (T2M015), and five have recursive or forward-referenced input types and were already flagged for @workflow.init (T2M012) or a dynamic activity call (T2M032). For those, --import-check adds the SDK's exact message (T2M090).

To reproduce: python scripts/validate_corpus.py path/to/samples-python --out /tmp/converted. The full list of errors on the samples is in docs/samples-python-errors.txt.

More tests go beyond imports. tests/test_sdk_validation.py runs a converted workflow (child workflow, sync and async activities, retry policy, signal, query and update) in the Mistral SDK's local mode. tests/test_e2e_temporal.py runs a converted workflow, with a class-based activity, on Temporal's test server through the Mistral SDK's test worker. That one downloads the test server, so it runs in CI.

Development

pip install -e ".[test,mistral]" ruff
pytest                                   # the SDK tests run when mistralai-workflows is installed
T2M_SAMPLES=/path/to/samples-python pytest
ruff check . && ruff format src tests scripts

CONTRIBUTING.md explains how the code is organised and the style rules.

Roadmap

• Optionally convert input dataclasses to pydantic models.
• Convert class-based activities whose class takes arguments, when the worker builds it in an obvious way.

License

MIT