Usar conectores en flujos de trabajo

Usa conectores dentro de flujos de trabajo: declara dependencias de conectores en un trabajador, llama herramientas de conectores desde actividades con credenciales específicas y ejecuta flujos de trabajo con manejo automático de OAuth.

Estado de la API: La integración de conectores de flujo de trabajo usa mistralai-workflows-plugins-mistralai. Estas son características beta y pueden cambiar.

Requisitos previos

Este manual cubre dos roles separados:

Lo que necesitas antes de empezar

Para completar este manual, necesitarás una clave de API de Mistral. En Studio, navega a la sección de claves de API y crea una nueva clave de API.

• Dedica un espacio de trabajo de Mistral para ejecutar tu flujo de trabajo. La ejecución del flujo de trabajo debe usar el mismo espacio de trabajo que la definición del flujo de trabajo.
• Al menos un conector registrado. Consulta Crea un agente asesor de bases de datos para ver un ejemplo completo del ciclo de vida de un conector, o crea uno directamente en Studio.
• Para conectores autenticados con OAuth como Notion y Gmail: no se necesitan credenciales preexistentes; el flujo de autenticación se activa automáticamente en el momento de la ejecución del flujo de trabajo; para conectores autenticados con bearer (por ejemplo, GitHub PAT), las credenciales deben almacenarse en la consola de Mistral antes de ejecutar el flujo de trabajo.

Crea un trabajador con conectores

Conceptos

Un flujo de trabajo que usa conectores tiene tres bloques de construcción:

El ConnectorAuthInterceptor es registrado automáticamente por el plugin cuando run_worker se inicia. Ejecuta un pre-vuelo de autenticación antes de cada ejecución del flujo de trabajo: si existen credenciales válidas (que coincidan con credentials_name si se especifica, o si existen credenciales predeterminadas) el flujo de trabajo procede inmediatamente; si no, activa un flujo de OAuth y espera a que el usuario se autentique. La autenticación bearer sobre la marcha no es compatible actualmente.

Instalación de trabajador y cliente de ejemplo

Para configurar un entorno de trabajador y flujo de trabajo preconstruido, ejecuta el siguiente comando:

uvx mistralai-workflows-cli setup

Esto estructura un proyecto Python listo para ejecutar con el SDK de Workflows ya configurado, un flujo de trabajo de ejemplo mínimo y comandos de ayuda para ejecutar tu trabajador y activar ejecuciones.

El comando te pide que generes una clave de API de Mistral en la Consola de Mistral. Sigue las instrucciones para generar la clave de API y luego pásala al comando cuando se te solicite. Las claves de API solo son accesibles una vez.

Paso 1 — Declara slots de conector

Crea un nuevo archivo Python en el directorio workflows (por ejemplo, workflows/connectors_example.py).

Los slots de conector se declaran a nivel de módulo. Cada slot contiene el nombre del conector y la configuración de autenticación:

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

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

connector() acepta:

Paso 2 — Escribe una actividad que llame a una herramienta de conector

Las actividades reciben un ToolCallClient a través de la inyección de dependencias usando Depends:

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) envía la llamada al conector y devuelve la respuesta de la herramienta sin procesar.

Paso 3 — Define la clase de flujo de trabajo

Declara slots de conector en la clase de flujo de trabajo con @uses_connectors:

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

• on_behalf_of=True ejecuta el flujo de trabajo usando la identidad del llamador, requerido para credenciales de conector por usuario
• @uses_connectors(...) registra los slots para que el interceptor de autenticación sepa qué conectores autenticar antes de que se ejecute el cuerpo del flujo de trabajo.
  • Nota: si quieres usar varios conectores en el mismo flujo de trabajo, puedes usar @uses_connectors(github_app, notion)
• :warning: ¡El orden de @uses_connectors y @workflow.define importa! Necesitas aplicar el @uses_connectors después de la definición del flujo de trabajo

Paso 4 — Ejecuta el trabajador

Archivo de trabajador completo y autocontenido:

from __future__ import annotations

import asyncio

import pydantic
import structlog

import mistralai.workflows as workflows
from mistralai.workflows import Depends
from mistralai.workflows.core.config.config import config
from mistralai.workflows.core.logging import setup_logging
from mistralai.workflows.plugins.mistralai.connectors import (
    ToolCallClient,
    connector,
    uses_connectors,
)

logger = structlog.get_logger(__name__)

github_connector = connector("github_app")


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


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


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


if __name__ == "__main__":
    setup_logging(
        log_format=config.common.log_format,
        log_level=config.common.log_level,
        app_version=config.common.app_version,
    )
    asyncio.run(workflows.run_worker([GitHubIssueCreatorWorkflow]))

Inicia el trabajador:

make start-worker

Cómo funciona:

• El comando inicia el trabajador, se conecta a la API de Mistral y registra tu flujo de trabajo para que pueda esperar tareas. Para obtener detalles sobre este comando, consulta el Makefile.
• El ConnectorAuthInterceptor se carga automáticamente por el sistema de plugins; no se necesita configuración manual
• Antes de cada ejecución del flujo de trabajo, el interceptor verifica las credenciales para cada slot de conector auto_auth=True:
  • Credenciales válidas encontradas → procede inmediatamente
  • Conector OAuth2 sin credenciales → emite una URL de autenticación y espera a que el usuario se autentique
  • Conector Bearer sin credenciales almacenadas → lanza ConnectorError

Errores comunes y soluciones:

Ejecuta un flujo de trabajo con conectores

Usa execute_with_connector_auth_async del SDK mistralai para activar un flujo de trabajo. Este asistente sondea los eventos y, si el trabajador indica que se requiere autorización del conector, imprime la URL de OAuth y espera a que el usuario complete el flujo antes de que el flujo de trabajo se reanude.

Script de cliente de ejemplo

import asyncio
import os

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


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


async def on_auth_required(state: ConnectorAuthTaskState) -> None:
    """Default callback: opens the OAuth URL in the browser and waits."""
    if state.auth_url:
        logger.info(
            "Auth required — opening browser (connector=%s, auth_url=%s)",
            state.connector_name,
            state.auth_url,
        )
        webbrowser.open(state.auth_url)
    else:
        logger.info(
            "Auth required — authenticate the connector manually (connector=%s)",
            state.connector_name,
        )
    input("Press Enter after completing the OAuth flow...")



async def main(args) -> None:
    bindings = json.loads(args.bindings) if args.bindings else []
    connector_slots: Sequence[ConnectorSlot] = [
        ConnectorSlot(**binding) for binding in bindings
    ]

    logger.info("Running workflow with connector slots: %s", connector_slots)
    async with Mistral(api_key=args.api_key, server_url=args.server_url) as client:
        response = await execute_with_connector_auth_async(
            client=client,
            workflow_identifier="github-issue-creator",
            input_data=GitHubIssuePrompt(
                owner="my-org",
                repo="my-repo",
                title="Bug: something is broken",
                body="Steps to reproduce...",
            ),
            deployment_name=args.deployment_name,
            connectors=connector_slots,
            on_auth_required=on_auth_required,
        )
        print(response)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Search meetings")
    parser.add_argument("--api-key", required=True, help="Mistral API key")
    parser.add_argument(
        "--server_url",
        required=False,
        default="https://api.mistral.ai",
        help="Mistral server URL",
    )
    parser.add_argument("--deployment-name", required=True, help="Deployment name")
    parser.add_argument("--workflow_name", required=True, help="workflow to execute")
    parser.add_argument(
        "--bindings",
        default=None,
        help="dict containing connector bindings",
    )
    asyncio.run(main(parser.parse_args()))

Ejecución simple (conectores pre-registrados)

Cuando el conector usa un token bearer que ya está almacenado, la ejecución es sencilla: Ejecútalo:

make execute workflow=github-issue-creator input='{"owner": "your-username", "repo": "your-repo", "title": "Hello World", "body": "Hello World"}'

Ejecución de Oauth/Bearer con enlace de conector en tiempo de ejecución

Para especificar qué credenciales usar para cada conector al ejecutar el flujo de trabajo, usa el parámetro de enlaces del script. Registrará extensiones de ejecución de flujo de trabajo y comunicará al trabajador qué credenciales usar para este usuario.

uv run python -m 09_workflow_executor_with_connectors --api-key <your_api_key> --query meeting  --bindings '[{"connector_name": "github_app", "credentials_name": "galilou"}]' --workflow_name github-issue-creator --deployment-name default

Campos de enlace:

El flujo de OAuth

Cuando un conector requiere OAuth y el usuario no tiene credenciales almacenadas, el flujo de trabajo se pausa. Si el flujo de trabajo se ejecuta a través de la CLI, la devolución de llamada on_auth_required imprime la URL y espera:

Connector 'Notion' requires authorization.
Open this URL in your browser to authenticate:
  https://api.notion.com/v1/oauth/authorize?client_id=...

Waiting for authorization... (press Ctrl+C to cancel)
✓ Authorization complete.

Una vez que el usuario se autentica en su navegador, el trabajador detecta las nuevas credenciales y el flujo de trabajo se reanuda automáticamente.

Cómo funciona:

• execute_with_connector_auth_async ejecuta el flujo de trabajo y sondea los eventos de tareas de la API de Mistral Workflows
• Cuando el trabajador emite un evento connector_auth_started, se llama a on_auth_required con la URL de OAuth
• Una actividad de latido de larga duración en el trabajador sondea la API de credenciales hasta que aparecen las credenciales
• Una vez que las credenciales se verifican con una llamada list_tools, el flujo de trabajo continúa
• La llamada del cliente regresa cuando el flujo de trabajo se completa, o falla en caso de error

Errores comunes y soluciones:

Ejecuta un flujo de trabajo a través de Studio

Cuando no se especifican las credenciales, el trabajador usará tus credenciales predeterminadas.

Para promover credenciales a predeterminadas, consulta el Manual de autenticación múltiple.

Flujo de OAuth en el panel de ejecución

Si no tienes credenciales para un conector dado y el conector es OAuth2, el trabajador activa un flujo de autenticación; recibirás un evento en el panel de ejecución que te pedirá que te autentiques (un icono de llave naranja indica que se requiere una acción).

Una vez que completes el flujo, las credenciales recién creadas se almacenan como tus predeterminadas y el flujo de trabajo se reanuda automáticamente.

Resumen

Este manual cubrió cómo usar conectores dentro de los flujos de trabajo de Mistral: declarar slots de conectores, llamar a herramientas de conectores desde actividades y ejecutar flujos de trabajo con manejo automático de OAuth desde un script de cliente o desde Studio.

Lo que cubre este manual:

• Declarar slots de conectores en un trabajador de Workflows
• Escribir actividades que llaman a herramientas de conectores a través de ToolCallClient
• Definir una clase de flujo de trabajo con @uses_connectors
• Ejecutar un trabajador y ejecutar un flujo de trabajo
• Ejecución del lado del cliente con manejo automático del flujo de OAuth
• Activar un flujo de trabajo desde Studio

Características de Mistral usadas:

• Workflows (beta)
• Connectors (beta)

Otros servicios:

• GitHub MCP — Conector autenticado con bearer
• Notion — Conector autenticado con OAuth2

Ver la documentación