Create an Activity — Mistral Workflows: Durable Orchestration

What Is an Activity?

An activity is the basic operational unit in a workflow. This is where the real work happens: external API calls, database queries, sending emails, file processing, and calls to AI models. The division of labor is simple and will serve as your compass throughout this course: any operation involving input/output (I/O) or side effects must be encapsulated in an activity.

While the workflow orchestrates, the activity executes. This is the fundamental distinction, and it has a direct consequence: because risky work is isolated within activities, the engine can retry, measure, and replay without ever duplicating side effects.

Anatomy of an Activity

An activity is a Python async function decorated with @workflows.activity():

import mistralai.workflows as workflows

@workflows.activity()
async def get_user_profile(user_id: str) -> dict:
    """Retrieves a user profile from the API."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

Four elements in these few lines deserve your attention. The @workflows.activity() decorator registers the function as an activity with the engine, making it callable from a workflow. The async def is mandatory: all activities must be asynchronous, without exception. Parameters must be JSON-serializable, meaning simple types—str, int, float, bool, list, dict—or Pydantic models; passing a connection object or file handle as an argument will fail serialization. The same rule applies to the return value, which travels across the network before being written to the event log.

Concrete Activity Examples

The first case, likely the most common in this course, encapsulates a call to a Mistral model. Note that the activity also returns token consumption: once in the log, this kind of metadata lets you reconstruct the cost of an execution after the fact.

import mistralai.workflows as workflows
from mistralai.client import Mistral

@workflows.activity()
async def generate_summary(text: str) -> dict:
    """Generates a summary of the provided text via Mistral."""
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "You are an assistant specialized in text summarization."},
            {"role": "user", "content": f"Summarize this text in 3 sentences:\n\n{text}"}
        ]
    )
    return {
        "summary": response.choices[0].message.content,
        "tokens_used": response.usage.total_tokens
    }

The second case illustrates database access. The try/finally block is not decorative: if the activity fails and is retried, leaving a connection open on every attempt will quickly exhaust the PostgreSQL pool.

import mistralai.workflows as workflows
import asyncpg

@workflows.activity()
async def save_report(report: dict) -> dict:
    """Saves a report to PostgreSQL."""
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    try:
        row = await conn.fetchrow(
            "INSERT INTO reports (title, content, created_at) VALUES ($1, $2, NOW()) RETURNING id",
            report["title"],
            report["content"]
        )
        return {"id": row["id"], "status": "saved"}
    finally:
        await conn.close()

The third case, sending email, is where side effects are most visible: once the message is sent, no retry can catch it.

import mistralai.workflows as workflows
import aiosmtplib

@workflows.activity()
async def send_notification(recipient: str, subject: str, body: str) -> dict:
    """Sends a notification email."""
    message = f"Subject: {subject}\nTo: {recipient}\n\n{body}"
    await aiosmtplib.send(
        message,
        hostname="smtp.example.com",
        port=587,
        username="[email protected]",
        password="secret",
        use_tls=True
    )
    return {"status": "sent", "recipient": recipient}

Core Characteristics

First, a welcome freedom: unlike workflows, activities have no determinism constraints. You can freely use datetime.now(), uuid.uuid4(), random.random(), or any other non-deterministic call inside an activity. That is precisely what activities are for—absorbing everything the workflow is not allowed to do.

The platform then treats each activity call as an atomic unit: it either succeeds entirely or fails entirely, with no intermediate state visible to the workflow. If your activity does three things and the second one crashes, the workflow sees only a failure, not partial success. This is why you should avoid stacking five distinct operations into a single function.

Once an activity succeeds, its result is permanently recorded in the workflow’s event log. During a replay, the activity is never re-executed: only the recorded result is used. In return, because an activity may be retried on failure—network timeout, transient error—it must be idempotent: executed multiple times with the same parameters, it must produce the same result. Switching from a POST to a PUT on a stable identifier is often enough to achieve this property, as here where the email serves as the key:

@workflows.activity()
async def create_or_update_client(email: str, name: str) -> dict:
    """Idempotent: uses an upsert instead of an insert."""
    async with httpx.AsyncClient() as client:
        response = await client.put(
            f"https://api.crm.com/clients/{email}",
            json={"name": name, "email": email}
        )
        return response.json()

Naming an Activity

The decorator accepts an explicit name, which significantly improves trace readability when a technical function bears an English name that no one else on the team links to the business domain:

@workflows.activity(name="Process Customer Emails")
async def process_customer_emails(batch_id: str) -> dict:
    # ...
    pass

This name will appear in OpenTelemetry traces and facilitate debugging. Make it a habit to set it from the start: the day you are hunting down why an execution took forty minutes, you will read meaningful labels rather than a list of function names.

Size Constraints

Two limits govern exchanges: 2 MB maximum for input parameters and 2 MB maximum for the return value. These stem directly from event sourcing, since these values are serialized and stored in the log.

The practical consequence is clear. If you need to process large files—a 40 MB scanned PDF, an export archive—do not pass them from activity to activity. Store them in an external service (S3, GCS) and circulate only the URL or identifier: the activity downloads the file, processes it, writes the result back to storage, and returns a reference of a few hundred bytes.

Key Takeaways

An activity is an async function decorated with @workflows.activity() that performs the real work (I/O, API, DB)
Parameters and return values must be JSON-serializable
Activities have no determinism constraints
A successful activity is never re-executed during a replay
Activities must be idempotent (same result if executed multiple times)
2 MB limit on input and output per activity