워크플로에서 커넥터 사용하기
워크플로에서 커넥터 사용하기 (Connectors in Workflows)
Mistral Workflows 안에서 Connectors를 사용하는 방법을 다루는 쿡북이에요. 워커에 커넥터 의존성을 선언하고, 특정 자격 증명으로 액티비티에서 커넥터 도구를 호출하며, 자동 OAuth 처리를 통해 워크플로를 실행하는 방법을 보여줘요.
출처: 문서
본문
Workflows 내부에서 Connectors를 사용해 봅시다 — 워커에 커넥터 의존성을 선언하고, 특정 자격 증명으로 액티비티에서 커넥터 도구를 호출하며, 자동 OAuth 처리를 통해 워크플로를 실행하는 방법을 다뤄요.
API 상태 (API status): 워크플로 커넥터 통합은 mistralai-workflows-plugins-mistralai를 사용해요. 이들은 베타(beta) 기능으로 변경될 수 있어요.
사전 준비사항 (Prerequisites)
이 쿡북은 두 가지 별개의 역할을 다뤄요.
| Role | What it does | Where it runs |
|---|---|---|
| Worker | Hosts the workflow and calls connector tools | A long-running server process, separate from client scripts |
| Client | Triggers workflow execution and handles OAuth redirects | Any script or AI Studio |
시작 전에 필요한 것 (What you need before starting)
이 쿡북을 완료하려면 Mistral API 키가 필요해요. Studio에서 API keys 섹션으로 이동해서 새 API 키를 만들어요.
워크플로를 실행할 전용 Mistral 워크스페이스를 마련하세요. 워크플로 실행은 워크플로 정의와 같은 워크스페이스를 사용해야 해요.
등록된 Connector가 하나 이상 있어야 해요. Build a Database Advisor Agent를 보면 전체 커넥터 수명주기 예제를 볼 수 있고, Studio에서 직접 만들 수도 있어요.
Notion, Gmail 같은 OAuth 인증 커넥터의 경우: 사전 저장된 자격 증명이 필요 없어요 — 인증 흐름이 워크플로 실행 시점에 자동으로 트리거돼요. Bearer 인증 커넥터(예: GitHub PAT)의 경우 자격 증명을 워크플로 실행 전에 Mistral 콘솔에 저장해야 해요.
커넥터가 있는 워커 만들기 (Create a Worker with Connectors)
개념 (Concepts)
커넥터를 사용하는 워크플로는 세 가지 구성 요소가 있어요.
| Building block | What it does |
|---|---|
connector(name) |
Declares a named connector slot — a dependency on a Connector |
@uses_connectors(slot, ...) |
Attaches declared slots to a workflow class so the runtime knows which connectors to authenticate |
ToolCallClient |
Activity-level client for calling connector tools, injected via Depends |
ConnectorAuthInterceptor는 run_worker가 시작될 때 플러그인이 자동으로 등록해요. 이 인터셉터는 매 워크플로 실행 전에 인증 사전 검사(preflight)를 실행해요: 유효한 자격 증명이 있으면(credentials_name이 지정된 경우 그에 맞는 것, 그 외에는 기본 자격 증명이 있으면) 워크플로가 즉시 진행되고, 없으면 OAuth 흐름을 트리거해서 사용자가 인증할 때까지 기다려요. Bearer 온더플라이(on-the-fly) 인증은 현재 지원되지 않아요.
워커와 예제 클라이언트 설치 (Worker and example client install)
미리 만들어진 워커·워크플로 환경을 설정하려면 다음 명령을 실행해요.
uvx mistralai-workflows-cli setup
이 명령은 Workflows SDK가 이미 구성된 실행 가능한 Python 프로젝트, 최소 예제 워크플로, 워커 실행 및 실행 트리거를 위한 헬퍼 명령을 스캐폴딩해요.
명령은 Mistral 콘솔에서 Mistral API 키를 생성하도록 안내해요. 프롬프트를 따라 API 키를 생성한 다음, 요청할 때 명령에 전달하세요. API 키는 한 번만 접근할 수 있어요.
1단계 — 커넥터 슬롯 선언 (Step 1 — Declare connector slots)
workflows 디렉터리에 새 Python 파일을 만들어요(예: workflows/connectors_example.py).
커넥터 슬롯은 모듈 수준에서 선언돼요. 각 슬롯은 커넥터 이름과 인증 구성을 담아요.
from mistralai.workflows.plugins.mistralai.connectors import connector
github_connector = connector("github_app")
notion_connector = connector("notion")
connector()은 다음을 받아요.
| Parameter | Default | Description |
|---|---|---|
name |
— | Connector name or ID as registered in Studio |
auto_auth |
True |
Run OAuth preflight before the workflow starts |
credentials_name |
None |
Pin to a specific shared credential name (ex: workspace scoped). Not supported yet, only runtime credentials are supported |
2단계 — 커넥터 도구를 호출하는 액티비티 작성 (Step 2 — Write an activity that calls a connector tool)
액티비티는 Depends를 사용한 의존성 주입으로 ToolCallClient를 받아요.
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)는 호출을 Connector로 보내고 원시 도구 응답을 반환해요.
3단계 — 워크플로 클래스 정의 (Step 3 — Define the workflow class)
@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는 호출자의 정체성으로 워크플로를 실행해요 — 사용자별 커넥터 자격 증명에 필요.@uses_connectors(...)는 슬롯을 등록해서 인증 인터셉터가 워크플로 본문이 실행되기 전에 어떤 커넥터를 인증할지 알게 해요.
참고: 같은 워크플로에서 여러 커넥터를 사용하고 싶다면 @uses_connectors(github_app, notion)처럼 쓸 수 있어요.
⚠️ 경고: @uses_connectors와 @workflow.define의 순서가 중요해요! @uses_connectors를 워크플로 정의 다음에 적용해야 해요.
4단계 — 워커 실행 (Step 4 — Run the worker)
완전한 자체 포함 워커 파일:
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]))
워커 시작:
make start-worker
작동 방식 (How it works):
- 이 명령은 워커를 시작하고 Mistral API에 연결하며, 작업을 기다릴 수 있도록 워크플로를 등록해요. 이 명령에 대한 자세한 내용은 Makefile을 참조하세요.
ConnectorAuthInterceptor는 플러그인 시스템이 자동으로 로드해요 — 수동 설정이 필요 없어요.- 매 워크플로 실행 전에 인터셉터가
auto_auth=True인 모든 커넥터 슬롯에 대한 자격 증명을 확인해요.- 유효한 자격 증명 발견 → 즉시 진행
- 자격 증명 없는 OAuth2 커넥터 → auth URL을 내보내고 사용자 인증을 기다림
- 저장된 자격 증명이 없는 Bearer 커넥터 →
ConnectorError발생
일반적인 오류와 해결책 (Common errors & fixes):
| Error | Cause | Fix |
|---|---|---|
ConnectorError: Credential 'x' not found |
Named credential doesn't exist for this connector | Create the credential first, or omit credentials_name to use default credentials if available |
ConnectorAuthTimeout |
User didn't complete the OAuth flow within 10 minutes | Re-run the workflow and complete the browser auth step promptly |
ConnectorError: ... requires bearer authentication |
Bearer connector has no stored credential | Add a credential via the Mistral dashboard before running |
ConnectorError: Extension bindings reference unknown connectors |
A runtime binding names a connector not in @uses_connectors |
Check that connector_name in the binding matches a declared slot |
커넥터가 있는 워크플로 실행 (Execute a Workflow with Connectors)
mistralai SDK의 execute_with_connector_auth_async를 사용해서 워크플로를 트리거해요. 이 헬퍼는 이벤트를 폴링하고, 워커가 커넥터 인증이 필요하다고 신호하면 OAuth URL을 출력하고 워크플로가 재개되기 전에 사용자가 흐름을 완료할 때까지 기다려요.
예제 클라이언트 스크립트 (Example client script)
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()))
단순 실행 (사전 등록 커넥터) (Simple execution, pre-registered connectors)
커넥터가 이미 저장된 bearer 토큰을 사용할 때 실행은 간단해요.
실행:
make execute workflow=github-issue-creator input='{"owner": "your-username", "repo": "your-repo", "title": "Hello World", "body": "Hello World"}'
OAuth/Bearer 실행 (런타임 커넥터 바인딩) (Oauth/Bearer execution with runtime connector binding)
워크플로를 실행할 때 각 커넥터에 사용할 자격 증명을 지정하려면 스크립트의 bindings 매개변수를 사용해요. 이 매개변수는 워크플로 실행 확장을 등록하고 이 사용자에 어떤 자격 증명을 사용할지 워커에 전달해요.
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
바인딩 필드 (Binding fields):
| Field | Description |
|---|---|
connector_name |
Must match a connector slot declared with @uses_connectors on the workflow |
credentials_name |
Select specific stored credentials for this execution |
OAuth 흐름 (The OAuth flow)
커넥터가 OAuth를 요구하고 사용자에게 저장된 자격 증명이 없으면 워크플로가 일시 중지돼요. 워크플로가 CLI로 실행되면 on_auth_required 콜백이 URL을 출력하고 기다려요.
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.
사용자가 브라우저에서 인증하면 워커가 새 자격 증명을 감지하고 워크플로가 자동으로 재개돼요.
작동 방식 (How it works):
execute_with_connector_auth_async는 워크플로를 실행하고 Mistral Workflows API에서 작업 이벤트를 폴링해요.- 워커가
connector_auth_started이벤트를 내보내면on_auth_required가 OAuth URL과 함께 호출돼요. - 워커의 장기 실행 하트비트 액티비티가 자격 증명이 나타날 때까지 credentials API를 폴링해요.
- 자격 증명이
list_tools호출로 검증되면 워크플로가 진행돼요. - 워크플로가 완료되면 클라이언트 호출이 반환되고, 실패하면 예외를 발생시켜요.
일반적인 오류와 해결책 (Common errors & fixes):
| Error | Cause | Fix |
|---|---|---|
ConnectorAuthTimeout |
OAuth flow not completed within 10 minutes | Re-run and complete the browser auth step |
404 Not Found on workflow execute |
Workflow not registered or worker not running | Start the worker first and verify the workflow name matches exactly |
ConnectorError: Extension bindings reference unknown connectors |
bindings names a connector not declared with @uses_connectors |
Match connector_name to a slot declared in the workflow |
Studio로 워크플로 실행 (Execute a workflow via Studio)
자격 증명을 지정하지 않으면 워커가 기본 자격 증명을 사용해요.
자격 증명을 기본으로 승격하는 방법은 Multiple Authentication 쿡북을 참조하세요.
실행 패널에서의 OAuth 흐름 (OAuth flow in the execution panel)
주어진 커넥터에 대한 자격 증명이 없고 커넥터가 OAuth2라면, 워커가 인증 흐름을 트리거해요 — 실행 패널에서 인증하라는 이벤트를 받게 돼요(주황색 키 아이콘은 작업이 필요함을 나타냄).
흐름을 완료하면 새로 생성된 자격 증명이 기본 값으로 저장되고 워크플로가 자동으로 재개돼요.
요약 (Summary)
이 쿡북은 Mistral Workflows 안에서 Connectors를 사용하는 방법을 다뤘어요 — 커넥터 슬롯 선언, 액티비티에서 Connector 도구 호출, 클라이언트 스크립트나 Studio에서 자동 OAuth 처리로 워크플로 실행하기까지요.
이 쿡북이 다루는 내용 (What this cookbook covers):
- Workflows 워커에서 커넥터 슬롯 선언
ToolCallClient로 Connector 도구를 호출하는 액티비티 작성@uses_connectors로 워크플로 클래스 정의- 워커 실행 및 워크플로 실행
- 자동 OAuth 흐름 처리가 있는 클라이언트 측 실행
- Studio에서 워크플로 트리거
사용한 Mistral 기능 (Mistral features used):
- Workflows (beta)
- Connectors (beta)
기타 서비스 (Other services):
- GitHub MCP — Bearer 인증 Connector
- Notion — OAuth2 인증 Connector
더 알아보기 (Learn more)
- 원본 문서: Connectors in Workflows