On-behalf-of workflows

Some workflows need to act as the user who triggered them — for example, accessing the user's connectors, querying resources scoped to their workspace, or calling the Mistral API with their identity. On-behalf-of (OBO) mode makes this possible.

When a workflow is marked as OBO, the platform captures the triggering user's identity at execution time and makes it available to the worker. The worker can then obtain short-lived credentials that represent that user, without ever seeing their API key or session token.

At execution time, the platform attaches the triggering user's identity to the execution context. When an activity requests user credentials, the SDK:

1. Retrieves the execution token from the current workflow context.
2. Exchanges it with the platform for a short-lived JWT representing the triggering user.
3. Injects the JWT into the Authorization header of every outgoing request.
4. Caches and refreshes the JWT automatically.

From the activity's perspective, the client behaves like a normal Mistral client — the identity swap is transparent.

Set on_behalf_of=True on the workflow definition:

import mistralai.workflows as workflows

@workflows.workflow.define(name="user_report", on_behalf_of=True)
class UserReportWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, report_type: str) -> dict:
        ...

import mistralai.workflows as workflows

@workflows.workflow.define(name="user_report", on_behalf_of=True)
class UserReportWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, report_type: str) -> dict:
        ...

Inside an activity, create a Mistral client with use_executor_credentials=True:

from datetime import timedelta

import mistralai.workflows as workflows
from mistralai.workflows.client import get_mistral_client

@workflows.workflow.define(name="user_search", on_behalf_of=True)
class UserSearchWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, query: str) -> str:
        return await search_with_user_identity(query)

@workflows.activity(start_to_close_timeout=timedelta(seconds=30))
async def search_with_user_identity(query: str) -> str:
    client = get_mistral_client(use_executor_credentials=True)
    response = await client.chat.complete_async(
        model="mistral-medium-latest",
        messages=[{"role": "user", "content": query}],
    )
    return response.choices[0].message.content

from datetime import timedelta

import mistralai.workflows as workflows
from mistralai.workflows.client import get_mistral_client

@workflows.workflow.define(name="user_search", on_behalf_of=True)
class UserSearchWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, query: str) -> str:
        return await search_with_user_identity(query)

@workflows.activity(start_to_close_timeout=timedelta(seconds=30))
async def search_with_user_identity(query: str) -> str:
    client = get_mistral_client(use_executor_credentials=True)
    response = await client.chat.complete_async(
        model="mistral-medium-latest",
        messages=[{"role": "user", "content": query}],
    )
    return response.choices[0].message.content

You can call the Mistral API as the worker within an OBO workflow — for example, to ensure requests are rate-limited against the worker's workspace rather than the user's. Create a second client with use_executor_credentials=False (the default):

@workflows.activity(start_to_close_timeout=timedelta(seconds=30))
async def my_activity(query: str) -> str:
    # Calls the API as the triggering user
    user_client = get_mistral_client(use_executor_credentials=True)

    # Calls the API as the worker (uses the worker's own API key)
    worker_client = get_mistral_client(use_executor_credentials=False)
    ...

@workflows.activity(start_to_close_timeout=timedelta(seconds=30))
async def my_activity(query: str) -> str:
    # Calls the API as the triggering user
    user_client = get_mistral_client(use_executor_credentials=True)

    # Calls the API as the worker (uses the worker's own API key)
    worker_client = get_mistral_client(use_executor_credentials=False)
    ...

This lets you choose the right identity on a per-call basis within the same activity.

• The caller must have a full user identity. Executions triggered by API keys without user_id / organization_id headers are rejected.
• The identity JWT is short-lived and scoped to the execution. If the execution ends, the JWT can no longer be refreshed.