Connectors in Workflows

Use Connectors inside a Workflow to call external services (GitHub, Notion, Slack, Outlook, etc.) without managing credentials yourself. The Workflow declares which Connectors it needs, the Mistral platform resolves credentials at runtime and triggers OAuth flows on demand.

Without Connector slots, every Workflow that talks to an external API has to handle its own credential storage, OAuth dance, and per-user token isolation. Slots centralize all three:

• No secrets in Workflow code: credentials are resolved at runtime by the platform.
• Automatic OAuth: if the caller hasn't authorized yet, the Workflow pauses and emits an auth URL, then resumes when the flow completes.
• Per-user identity: Workflows run on behalf of the caller, so each user uses their own stored credentials.
• Swappable auth: bearer (PAT) and OAuth2 Connectors use the same Workflow code.

Connector slots ship with the Mistral plugin:

uv add "mistralai-workflows[mistralai]"

uv add "mistralai-workflows[mistralai]"

You also need at least one Connector registered for your workspace. Create one from Studio › Context › Connectors, or via the Connectors API.

Add credentials before running a Workflow:

• Bearer-authenticated Connectors (e.g. GitHub PAT) require credentials in Studio first.
• OAuth2 Connectors also work best with credentials added first. As a fallback, the Workflow triggers an OAuth flow on demand the first time it runs without credentials (see How the fallback OAuth flow works).

Each user stores their own credentials per Connector in Studio. You can keep a single credential or store several named credentials (for example two GitHub PATs with different scopes, or a personal and a work Microsoft account) and pick which one to use per Workflow execution.

1. Open Studio › Context › Connectors and select a Connector.
2. Switch to the Credentials tab.
3. Click + Add credentials, give it a name (alphanumeric and hyphens), and complete the bearer token paste or OAuth flow.
4. One credential is always the default. To change which one runs when no name is specified, edit a credential and mark it as default.

Credentials are stored per user: each caller uses their own stored credentials when the Workflow runs.

You can also create, list, and delete credentials programmatically via client.beta.connectors. Use this when you need to provision credentials at scale, rotate tokens, or script the OAuth handoff.

import os
from mistralai import Mistral

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

# Bearer connector: store a named credential and mark it default
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,
)

# OAuth2 connector: request an auth URL, the user completes it in a browser
result = await client.beta.connectors.get_auth_url_async(
    connector_id_or_name="outlook_calendar",
    credentials_name="personal",
)
print(result.auth_url)

# List and delete
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"])

# Bearer connector: store a named credential and mark it default
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,
)

# OAuth2 connector: request an auth URL, the user completes it in a browser
result = await client.beta.connectors.get_auth_url_async(
    connector_id_or_name="outlook_calendar",
    credentials_name="personal",
)
print(result.auth_url)

# List and delete
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",
)

When a Workflow execution starts, the worker's auth interceptor runs a preflight on every Connector slot declared with @uses_connectors. If valid credentials exist for the caller, the Workflow body runs immediately. If not (typical first OAuth2 use), the worker pauses, gets an auth URL from the Mistral API, forwards it to the client as an auth_url event, and waits while the user completes authorization in their browser. Once the credentials land in storage, the Workflow resumes.

The polling activity heartbeats while it waits, so a slow user doesn't cause the worker to time out. The auth URL has a 10-minute window before ConnectorAuthTimeout fires.

A Connector Workflow has three pieces: a slot declaration, an activity that calls the Connector, and a Workflow class that ties them together.

Slots are declared at module level. Each slot holds the Connector name as registered in 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) accepts these parameters:

Activities receive a ToolCallClient via dependency injection. Depends(slot) resolves the slot to an authenticated client at runtime.

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) dispatches the call to the MCP Connector and returns the raw tool response.

Use @uses_connectors to register the slots, and on_behalf_of=True so the Workflow runs with the caller's identity:

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,
        )

Notes:

• on_behalf_of=True runs the Workflow under the caller's identity. Required to resolve per-user credentials.
• Pass several slots in one call when the Workflow needs more than one Connector: @uses_connectors(github_connector, notion_connector).
• Apply @uses_connectors after @workflow.define. The order matters.

When the worker starts, the plugin auto-registers a ConnectorAuthInterceptor that handles the preflight and OAuth pause described in How the fallback OAuth flow works.

Open Studio › Workflows, pick your Workflow, click Start workflow.

• If you have multiple named credentials for a Connector, the launch dialog lets you pick which one to use for this execution.
• To add or update credentials per Connector before starting a Workflow, go to Studio › Context › Connectors and open the Credentials tab.
• As a fallback, if you start a Workflow with an OAuth2 Connector without any credentials, the execution panel shows an OAuth prompt (orange key icon). Complete the flow in a browser tab and the Workflow resumes automatically.

Use execute_with_connector_auth_async to handle the OAuth dance programmatically. The helper polls the execution, detects auth requests, calls your on_auth_required callback with the URL, and waits for the user to complete the flow.

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("Press Enter after completing the OAuth flow...")


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: something is broken",
                "body": "Steps to reproduce...",
            },
            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("Press Enter after completing the OAuth flow...")


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: something is broken",
                "body": "Steps to reproduce...",
            },
            on_auth_required=on_auth_required,
        )
        print(response)


asyncio.run(main())

If the caller already has valid credentials for every required slot, the OAuth step is skipped and the Workflow runs straight through.

If you have several named credentials for a Connector, pass a ConnectorSlot per slot to choose which one to use for that execution. Slot names must match the slots declared with @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,
)

The same Workflow code can be shared across a team while each user runs it with their own credentials. Omit credentials_name to fall back to the user's default credential for that Connector.

Pass a Connector slot directly to a Durable Agent to let the agent call Connector tools autonomously during its conversation loop. Keep @uses_connectors on the Workflow so the auth interceptor still runs:

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"List recent pull requests on {repo}.",
            connectors=[github_connector],
        )
        result = await Runner.run(agent=agent, inputs=f"Summarize PRs on {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"List recent pull requests on {repo}.",
            connectors=[github_connector],
        )
        result = await Runner.run(agent=agent, inputs=f"Summarize PRs on {repo}.")
        return result.final_output

The agent receives the Connector's tools in its tool belt and invokes them on each turn. OAuth and credential resolution still happen automatically via the Workflow interceptor.