Connecteurs dans les workflows

Utilisez les Connecteurs dans un workflow pour appeler des services externes (GitHub, Notion, Slack, Outlook, etc.) sans avoir à gérer vous-même les identifiants. Le workflow déclare les connecteurs nécessaires ; la plateforme Mistral résout les identifiants à l'exécution et déclenche les flux OAuth à la demande.

Sans emplacement de connecteur, chaque workflow qui interagit avec une API externe doit gérer lui-même le stockage des identifiants, l'authentification OAuth et l'isolation des tokens par utilisateur. Les emplacements centralisent les trois :

• Aucun secret dans le code du workflow : les identifiants sont résolus dynamiquement par la plateforme.
• OAuth automatique : si l'appelant n'a pas encore autorisé, le workflow s'interrompt et fournit une URL d'authentification, puis reprend une fois l'autorisation effectuée.
• Identité par utilisateur : les workflows s'exécutent pour le compte de l'appelant, chaque utilisateur utilise donc ses propres identifiants enregistrés.
• Authentification interchangeable : les connecteurs bearer (PAT) et OAuth2 utilisent le même code workflow.

Les emplacements de connecteur sont fournis avec le plugin Mistral :

uv add "mistralai-workflows[mistralai]"

uv add "mistralai-workflows[mistralai]"

Vous devez également avoir enregistré au moins un Connecteur pour votre espace de travail. Créez-en un via Studio › Contexte › Connecteurs, ou via l'API Connecteurs.

Ajoutez des identifiants avant d'exécuter un workflow :

• Les connecteurs authentifiés Bearer (par ex. GitHub PAT) nécessitent d'enregistrer les identifiants dans Studio avant usage.
• Les connecteurs OAuth2 fonctionnent également mieux avec des identifiants ajoutés en amont. À défaut, le workflow déclenche un flux OAuth à la première exécution sans identifiant (voir Comment fonctionne le flux OAuth alternatif).

Chaque utilisateur stocke ses propres identifiants par connecteur dans Studio. Vous pouvez conserver un seul identifiant ou en stocker plusieurs nommés (par exemple deux PAT GitHub avec des autorisations différentes, ou un compte Microsoft personnel et un compte professionnel) et choisir lequel utiliser pour chaque exécution de workflow.

1. Ouvrez Studio › Contexte › Connecteurs puis sélectionnez un connecteur.
2. Passez à l’onglet Identifiants.
3. Cliquez sur + Ajouter des identifiants, donnez-lui un nom (alphanumérique et tirets), puis effectuez le collage du token bearer ou suivez le flux OAuth.
4. Un identifiant est toujours défini comme par défaut. Pour modifier le choix lorsqu'aucun nom n'est spécifié, éditez un identifiant puis définissez-le par défaut.

Les identifiants sont stockés par utilisateur : chaque appelant utilise ses propres identifiants lors de l'exécution du workflow.

Vous pouvez aussi créer, lister et supprimer des identifiants de manière programmatique via client.beta.connectors. Pratique pour automatiser la création à grande échelle, faire la rotation des tokens ou automatiser le flux OAuth.

import os
from mistralai import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

# Connecteur Bearer : enregistrer un identifiant nommé et le définir par défaut
await client.beta.connectors.create_or_update_user_credentials_async(
    connector_id_or_name="github_app",
    name="github-pat-full",
    credentials={"bearer_token": os.environ["GITHUB_PAT"]},
    is_default=True,
)

# Connecteur OAuth2 : obtenir une URL d'authentification, l'utilisateur l'ouvre dans son navigateur
result = await client.beta.connectors.get_auth_url_async(
    connector_id_or_name="outlook_calendar",
    credentials_name="personal",
)
print(result.auth_url)

# Lister et supprimer
await client.beta.connectors.list_user_credentials_async(
    connector_id_or_name="github_app",
)
await client.beta.connectors.delete_user_credentials_async(
    connector_id_or_name="github_app",
    credentials_name="github-pat-old",
)

import os
from mistralai import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

# Connecteur Bearer : enregistrer un identifiant nommé et le définir par défaut
await client.beta.connectors.create_or_update_user_credentials_async(
    connector_id_or_name="github_app",
    name="github-pat-full",
    credentials={"bearer_token": os.environ["GITHUB_PAT"]},
    is_default=True,
)

# Connecteur OAuth2 : obtenir une URL d'authentification, l'utilisateur l'ouvre dans son navigateur
result = await client.beta.connectors.get_auth_url_async(
    connector_id_or_name="outlook_calendar",
    credentials_name="personal",
)
print(result.auth_url)

# Lister et supprimer
await client.beta.connectors.list_user_credentials_async(
    connector_id_or_name="github_app",
)
await client.beta.connectors.delete_user_credentials_async(
    connector_id_or_name="github_app",
    credentials_name="github-pat-old",
)

Lorsqu'une exécution de workflow démarre, l'intercepteur d'authentification du worker effectue un pré-check sur chaque emplacement de connecteur déclaré avec @uses_connectors. Si des identifiants valides existent pour l'appelant, le corps du workflow s'exécute immédiatement. Sinon (cas typique de première utilisation OAuth2), le worker s'interrompt, demande une URL d'authentification à l'API Mistral, la transmet au client comme événement auth_url, puis attend pendant que l'utilisateur termine l'autorisation dans son navigateur. Une fois les identifiants stockés, le workflow reprend.

L'activité de polling envoie un heartbeat pendant l'attente, ce qui évite qu'un utilisateur lent ne provoque un time-out du worker. L'URL d'authentification a une fenêtre de 10 minutes avant déclenchement de ConnectorAuthTimeout.

Un workflow connecteur comporte trois éléments : une déclaration d'emplacement, une activité qui utilise le connecteur, et une classe de workflow qui orchestre l'ensemble.

Les emplacements se déclarent au niveau du module. Chaque emplacement contient le nom du connecteur tel qu'enregistré dans Studio :

from mistralai.workflows.plugins.mistralai.connectors import connector

github_connector = connector("github_app")
notion_connector = connector("Notion")

from mistralai.workflows.plugins.mistralai.connectors import connector

github_connector = connector("github_app")
notion_connector = connector("Notion")

connector(name) accepte les paramètres suivants :

Les activités reçoivent un ToolCallClient par injection de dépendance. Depends(slot) résout l'emplacement vers un client authentifié à l'exécution.

from typing import Any

import mistralai.workflows as workflows
from mistralai.workflows import Depends
from mistralai.workflows.plugins.mistralai.connectors import ToolCallClient, connector

github_connector = connector("github_app")


@workflows.activity(name="create-github-issue")
async def create_github_issue(
    owner: str,
    repo: str,
    title: str,
    body: str,
    github: ToolCallClient = Depends(github_connector),
) -> None:
    await github.call_tool(
        tool_name="issue_write",
        arguments={
            "method": "create",
            "owner": owner,
            "repo": repo,
            "title": title,
            "body": body,
        },
    )

from typing import Any

import mistralai.workflows as workflows
from mistralai.workflows import Depends
from mistralai.workflows.plugins.mistralai.connectors import ToolCallClient, connector

github_connector = connector("github_app")


@workflows.activity(name="create-github-issue")
async def create_github_issue(
    owner: str,
    repo: str,
    title: str,
    body: str,
    github: ToolCallClient = Depends(github_connector),
) -> None:
    await github.call_tool(
        tool_name="issue_write",
        arguments={
            "method": "create",
            "owner": owner,
            "repo": repo,
            "title": title,
            "body": body,
        },
    )

call_tool(tool_name, arguments) redirige l'appel vers le Connecteur MCP et retourne la réponse brute de l'outil.

Utilisez @uses_connectors pour enregistrer les emplacements, et on_behalf_of=True afin d'exécuter le workflow avec l'identité de l'appelant :

import pydantic
import mistralai.workflows as workflows
from mistralai.workflows.plugins.mistralai.connectors import connector, uses_connectors

github_connector = connector("github_app")


class GitHubIssuePrompt(pydantic.BaseModel):
    owner: str
    repo: str
    title: str
    body: str


@workflows.workflow.define(name="github-issue-creator", on_behalf_of=True)
@uses_connectors(github_connector)
class GitHubIssueCreatorWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, prompt: GitHubIssuePrompt) -> None:
        await create_github_issue(
            prompt.owner,
            prompt.repo,
            prompt.title,
            prompt.body,
        )

import pydantic
import mistralai.workflows as workflows
from mistralai.workflows.plugins.mistralai.connectors import connector, uses_connectors

github_connector = connector("github_app")


class GitHubIssuePrompt(pydantic.BaseModel):
    owner: str
    repo: str
    title: str
    body: str


@workflows.workflow.define(name="github-issue-creator", on_behalf_of=True)
@uses_connectors(github_connector)
class GitHubIssueCreatorWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, prompt: GitHubIssuePrompt) -> None:
        await create_github_issue(
            prompt.owner,
            prompt.repo,
            prompt.title,
            prompt.body,
        )

Remarques :

• on_behalf_of=True exécute le workflow sous l'identité de l'appelant. Obligatoire pour la résolution d'identifiants par utilisateur.
• Passez plusieurs emplacements dans un appel si le workflow a besoin de plusieurs connecteurs : @uses_connectors(github_connector, notion_connector).
• Appliquez @uses_connectors après @workflow.define. L’ordre des décorateurs est important.

Au démarrage du worker, le plugin enregistre automatiquement un ConnectorAuthInterceptor qui gère le pré-check et la pause OAuth décrits dans Fonctionnement du flux OAuth de secours.

Ouvrez Studio › Workflows, sélectionnez votre workflow puis cliquez sur Démarrer le workflow.

• Si vous avez plusieurs identifiants nommés pour un connecteur, la fenêtre de lancement vous laisse choisir lequel utiliser pour cette exécution.
• Pour ajouter ou mettre à jour des identifiants par connecteur avant de lancer un workflow, rendez-vous dans Studio › Contexte › Connecteurs puis ouvrez l’onglet Identifiants.
• En solution de secours : si vous lancez un workflow sur un connecteur OAuth2 sans identifiants enregistrés, le panneau d'exécution affiche une invite OAuth (icône de clé orange). Terminez le flux dans un onglet navigateur pour que le workflow reprenne automatiquement.

Utilisez execute_with_connector_auth_async pour automatiser le flux OAuth. Le helper vérifie l'exécution, détecte les demandes d'authentification, appelle votre callback on_auth_required avec l'URL, puis attend que l'utilisateur termine le flux.

import asyncio
import webbrowser

from mistralai.client import Mistral
from mistralai.extra.workflows.connector_auth import (
    ConnectorAuthTaskState,
    execute_with_connector_auth_async,
)
from mistralai.extra.workflows.connector_slot import ConnectorSlot


async def on_auth_required(state: ConnectorAuthTaskState) -> None:
    if state.auth_url:
        webbrowser.open(state.auth_url)
    input("Appuyez sur Entrée après avoir terminé le flux OAuth...")


async def main() -> None:
    async with Mistral(api_key="<your-api-key>") as client:
        response = await execute_with_connector_auth_async(
            client=client,
            workflow_identifier="github-issue-creator",
            input_data={
                "owner": "my-org",
                "repo": "my-repo",
                "title": "Bug : quelque chose ne fonctionne pas",
                "body": "Étapes de reproduction...",
            },
            on_auth_required=on_auth_required,
        )
        print(response)


asyncio.run(main())

import asyncio
import webbrowser

from mistralai.client import Mistral
from mistralai.extra.workflows.connector_auth import (
    ConnectorAuthTaskState,
    execute_with_connector_auth_async,
)
from mistralai.extra.workflows.connector_slot import ConnectorSlot


async def on_auth_required(state: ConnectorAuthTaskState) -> None:
    if state.auth_url:
        webbrowser.open(state.auth_url)
    input("Appuyez sur Entrée après avoir terminé le flux OAuth...")


async def main() -> None:
    async with Mistral(api_key="<your-api-key>") as client:
        response = await execute_with_connector_auth_async(
            client=client,
            workflow_identifier="github-issue-creator",
            input_data={
                "owner": "my-org",
                "repo": "my-repo",
                "title": "Bug : quelque chose ne fonctionne pas",
                "body": "Étapes de reproduction...",
            },
            on_auth_required=on_auth_required,
        )
        print(response)


asyncio.run(main())

Si l'appelant dispose déjà d'identifiants valides pour tous les emplacements requis, l'étape OAuth est ignorée et le workflow s'exécute directement.

Si vous avez plusieurs identifiants nommés pour un connecteur, transmettez un ConnectorSlot par emplacement pour choisir celui à utiliser lors de cette exécution. Les noms des slots doivent correspondre aux emplacements déclarés avec @uses_connectors :

from mistralai.extra.workflows.connector_slot import ConnectorSlot

connector_slots = [
    ConnectorSlot(connector_name="github_app", credentials_name="github-pat-full"),
    ConnectorSlot(connector_name="Notion", credentials_name="work-notion"),
]

response = await execute_with_connector_auth_async(
    client=client,
    workflow_identifier="github-issue-creator",
    input_data={...},
    connectors=connector_slots,
    on_auth_required=on_auth_required,
)

from mistralai.extra.workflows.connector_slot import ConnectorSlot

connector_slots = [
    ConnectorSlot(connector_name="github_app", credentials_name="github-pat-full"),
    ConnectorSlot(connector_name="Notion", credentials_name="work-notion"),
]

response = await execute_with_connector_auth_async(
    client=client,
    workflow_identifier="github-issue-creator",
    input_data={...},
    connectors=connector_slots,
    on_auth_required=on_auth_required,
)

Le même code workflow peut circuler en équipe : chaque utilisateur l'exécute avec ses propres identifiants. Omettez credentials_name pour tomber sur la valeur par défaut de l'utilisateur pour ce connecteur.

Passez un emplacement de connecteur directement à un agent durable pour le laisser invoquer les outils connecteur de façon autonome dans sa boucle de conversation. Conservez bien @uses_connectors sur le workflow afin que l’intercepteur d’authentification fonctionne toujours :

from mistralai.workflows.plugins.mistralai import Agent, Runner
from mistralai.workflows.plugins.mistralai.connectors import connector, uses_connectors
import mistralai.workflows as workflows

github_connector = connector("github_app")


@workflows.workflow.define(name="github-agent", on_behalf_of=True)
@uses_connectors(github_connector)
class GitHubAgentWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, repo: str) -> str:
        agent = Agent(
            name="github-pr-lister",
            model="mistral-medium-latest",
            instructions=f"Lister les dernières pull requests sur {repo}.",
            connectors=[github_connector],
        )
        result = await Runner.run(agent=agent, inputs=f"Résume les PRs sur {repo}.")
        return result.final_output

from mistralai.workflows.plugins.mistralai import Agent, Runner
from mistralai.workflows.plugins.mistralai.connectors import connector, uses_connectors
import mistralai.workflows as workflows

github_connector = connector("github_app")


@workflows.workflow.define(name="github-agent", on_behalf_of=True)
@uses_connectors(github_connector)
class GitHubAgentWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, repo: str) -> str:
        agent = Agent(
            name="github-pr-lister",
            model="mistral-medium-latest",
            instructions=f"Lister les dernières pull requests sur {repo}.",
            connectors=[github_connector],
        )
        result = await Runner.run(agent=agent, inputs=f"Résume les PRs sur {repo}.")
        return result.final_output

L'agent reçoit les outils du connecteur dans sa toolbox et les appelle à chaque tour. L'authentification OAuth et la résolution des identifiants s'effectuent toujours automatiquement via l'intercepteur de workflow.