Async and Best Practices

Async I/O: A Technical Requirement

All activities in Mistral Workflows must use asynchronous input/output operations. This isn’t a style recommendation; it’s a technical constraint. A blocking synchronous call like requests.post() freezes the worker and prevents any other activity from running.

This requirement stems from the worker’s architecture, which executes activities within an asyncio event loop. When a coroutine awaits a network response asynchronously, it yields control back to the loop, allowing other activities to proceed. A synchronous call, by contrast, yields nothing.

import requests

# INCORRECT: Blocks the entire worker
@workflows.activity()
async def blocking_activity(url: str) -> dict:
    # requests.post() is synchronous — it blocks the event loop
    response = requests.post(url, json={"query": "test"})
    return response.json()

The symptom in production can be confusing: you launch ten activities in parallel with asyncio.gather, but they actually execute one after another. If each waits two seconds for a remote API, you measure twenty seconds instead of the expected two, with no errors raised.

Choose Natively Asynchronous Libraries

The fix is to replace every blocking library with its async equivalent. For HTTP calls, httpx is the default choice.

import httpx

# CORRECT: Non-blocking
@workflows.activity()
async def async_activity(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json={"query": "test"})
    return response.json()

aiohttp serves the same purpose with a slightly different API based on sessions.

import aiohttp

@workflows.activity()
async def aiohttp_activity(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json={"query": "test"}) as response:
            return await response.json()

For databases, asyncpg replaces psycopg2 for PostgreSQL. Note the try/finally block: the connection must close even if the query raises an exception. Otherwise, successive retries will exhaust the connection pool.

import asyncpg

# CORRECT: Asynchronous PostgreSQL driver
@workflows.activity()
async def async_db_query(user_id: str) -> dict:
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    try:
        row = await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )
        return dict(row) if row else {"error": "user not found"}
    finally:
        await conn.close()

File access follows the same logic with aiofiles, even for small files where you might be tempted to keep using open().

import aiofiles

# CORRECT: Asynchronous file reading
@workflows.activity()
async def read_file_async(path: str) -> dict:
    async with aiofiles.open(path, mode="r") as f:
        content = await f.read()
    return {"content": content, "size": len(content)}

The Mistral SDK itself is no exception: every method has a variant suffixed with _async, and that is the one you must call from an activity.

from mistralai.client import Mistral

@workflows.activity()
async def call_mistral(prompt: str) -> dict:
    client = Mistral()
    # Use complete_async, NOT complete
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[{"role": "user", "content": prompt}]
    )
    return {"response": response.choices[0].message.content}

Keep this mapping table handy during code reviews: any line from the left column spotted in an activity is a potential performance bug.

When No Async Version Exists

Some business libraries, such as a proprietary SDK or a legacy client, simply do not offer an asynchronous variant. In these cases, asyncio.to_thread() lets you run the blocking code in a separate thread without freezing the event loop.

import asyncio
from some_library import blocking_function

@workflows.activity()
async def compatibility_activity(params: dict) -> dict:
    """Wraps a blocking call in a thread."""
    result = await asyncio.to_thread(
        blocking_function,
        params["arg1"],
        params["arg2"]
    )
    return {"result": result}

This is a compromise, not an equivalent solution: each call consumes a thread from the pool, which has a finite size. For an activity called massively in parallel, you merely shift the bottleneck rather than removing it. Always prefer a natively async library when one exists.

Designing Clean Activities

Beyond asynchrony, four habits distinguish a robust activity from one that will cost you hours of debugging.

The first is scope. An activity should have one and only one responsibility, because it is also the unit of retry and persistence: a catch-all activity that fails while sending an email will replay the extraction and analysis from the start.

# GOOD: Each activity has a clear responsibility
@workflows.activity()
async def extract_text(url: str) -> dict:
    # ...
    pass

@workflows.activity()
async def analyze_sentiment(text: str) -> dict:
    # ...
    pass

# BAD: Catch-all activity
@workflows.activity()
async def do_everything(url: str) -> dict:
    text = ...  # extraction
    sentiment = ...  # analysis
    # send email
    # save to DB
    pass

The second concerns resources. The async with context manager guarantees the client closes even if an exception occurs, something manual calls inevitably forget sooner or later.

@workflows.activity()
async def clean_request(params: dict) -> dict:
    # Use async with to guarantee closure
    async with httpx.AsyncClient() as client:
        response = await client.get(params["url"])
    # The client is automatically closed here
    return response.json()

The third is more subtle: not all errors deserve a retry. A timeout, a 429, or a 500 are transient and should bubble up to trigger a retry. A 4xx error, however, indicates an invalid request that retrying three times won’t fix. Instead, return an explicit result and let the workflow decide.

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=2),
    retry_policy_max_attempts=3,
)
async def robust_activity(params: dict) -> dict:
    async with httpx.AsyncClient(timeout=60) as client:
        try:
            response = await client.post(params["url"], json=params["data"])
            response.raise_for_status()
            return {"status": "success", "data": response.json()}
        except httpx.TimeoutException:
            raise  # Will be retried by the platform
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                raise  # Rate limited — will be retried
            elif e.response.status_code >= 500:
                raise  # Server error — will be retried
            else:
                # Client error (4xx) — do not retry
                return {"status": "error", "code": e.response.status_code}

Finally, the fourth habit stems from the 2 MB limit per invocation: return only what the workflow actually needs to decide the next step. A count, a sample, and the URL to the full result are almost always sufficient.

@workflows.activity()
async def extract_data(source: str) -> dict:
    data = await retrieve_large_data(source)

    # Return only what the workflow needs
    return {
        "item_count": len(data),
        "summary": data[:10],  # Only the first 10
        "full_url": await upload_results(data)
    }

Key Takeaways

• Async I/O is mandatory: use httpx, asyncpg, aiofiles instead of their synchronous equivalents
• Blocking calls (requests, psycopg2, open) freeze the entire worker
• Use asyncio.to_thread() as a last resort for libraries without an async version
• Each activity must have one clear responsibility
• Always close connections with async with
• Handle errors explicitly: some should be retried, others should not
• Limit return sizes to stay under the 2 MB limit