Triage Slack messages into Intercom and a feature tracker with Mistral Apps

graph LR
    Slack["Slack: a new message in the channel"] -->|message event| Trigger["Studio webhook trigger<br/>(triggers/slack.json filter)"]
    Trigger -->|starts| Workflow["slack-triage workflow"]
    Workflow -->|classify| Model["mistral-medium-latest"]
    Workflow -->|bug| Intercom["Intercom connector:<br/>ticket"]
    Workflow -->|feature request| Tracker["Tracker connector:<br/>the app's /api/v1/feature_requests"]
    Workflow -->|replies in the thread| Slack

mistral login
mistral apps init slack-triage-workflow --caps fastapi,postgres,workflows,apps
cd slack-triage-workflow

slack-triage-workflow/
├── apps/
│   ├── api/src/api/routers/api/v1/
│   │   ├── __init__.py
│   │   └── feature_requests.py
│   └── worker/src/worker/
│       ├── activities.py
│       └── workflows/slack_triage.py
├── packages/py/
│   ├── cli/src/cli/commands/
│   │   ├── bind_secrets.py
│   │   ├── connector.py
│   │   ├── intercom_ids.py
│   │   └── triage_message.py
│   ├── db/src/db/
│   │   ├── accessors/feature_request.py
│   │   ├── migrations/versions/0001_feature_request.py
│   │   └── models/feature_request.py
│   └── env/src/env/triage.py
└── triggers/
    └── slack.json

MISTRAL_API_KEY=your-mistral-api-key
DEPLOYMENT_NAME=slack-triage-your-name

rm -rf .agents node_modules .nxignore package.json bun.lock bunfig.toml nx.json \
  tools/install.sh tools/db.sh tools/lib.sh \
  apps/api/project.json apps/worker/project.json packages/py/db/project.json

**/.nx
.nx

            hint="no selected capability owns a table; add a model and an Alembic revision for it",

def _find_repo_root() -> Path:
    for parent in Path(__file__).resolve().parents:
        if (parent / "apps.json").is_file() and (parent / "pyproject.toml").is_file():
            return parent
    return Path.cwd()

# Executed (`bash tools/uv.sh <args>`) -> strict mode, then exec uv. Sourced (by hand from bash or
# zsh) -> only export the environment and leave the caller's shell options alone:

# Executed directly -> run uv with this environment. Sourced -> only export the environment.

``DATABASE_URL``; skipped otherwise.

dependencies = [
    "utils",
    "env",
    "mistralai>=2.9.4",
    "mistralai-workflows[mistralai]>=3.16.0",
    "httpx>=0.28.1",
    "pydantic>=2.14.1",
    "structlog>=25.5.0",
    "watchfiles>=1.3",
]

dependencies = [
    "typer>=0.27",
    "utils",
    "env",
    "mistralai>=2.9.4",
    "structlog>=25.5.0",
    "httpx>=0.28.1",
]

import asyncio
import signal

import structlog
from env.logging import env
from mistralai import workflows
from mistralai_capabilities.workflows import client as workflows_client
from utils.logging import configure_logging

uv sync --all-packages

"""Settings for the Slack triage workflow.

The connector names, the Intercom IDs and the Slack token have no defaults: set them in `.env` locally,
or as workspace secrets for the deployed worker.
"""

from typing import Annotated

from env._base import BaseEnv
from env._mistral_secrets import WorkspaceSecret


class Env(BaseEnv):
    triage_model: str = "mistral-medium-latest"
    slack_bot_token: Annotated[str | None, WorkspaceSecret()] = None
    intercom_connector_name: Annotated[str | None, WorkspaceSecret()] = None
    intercom_ticket_type_id: Annotated[str | None, WorkspaceSecret()] = None
    intercom_workspace_id: Annotated[str | None, WorkspaceSecret()] = None
    tracker_connector_name: Annotated[str | None, WorkspaceSecret()] = None
    # Where ticket links in Slack replies point when not https://app.intercom.com, such as
    # https://app.eu.intercom.com for an EU workspace.
    intercom_app_url: str | None = None

    def require(self, name: str) -> str:
        """The value of setting ``name``, or an error naming the variable to set."""
        value = getattr(self, name)
        if not value:
            raise RuntimeError(f"{name.upper()} is not set")
        return value


env = Env()

TRIAGE_MODEL=mistral-medium-latest

TRIAGE_MODEL=
SLACK_BOT_TOKEN=
INTERCOM_CONNECTOR_NAME=
INTERCOM_TICKET_TYPE_ID=
INTERCOM_WORKSPACE_ID=
TRACKER_CONNECTOR_NAME=
INTERCOM_APP_URL=

"""A feature request filed in the tracker."""

from datetime import UTC, datetime

from sqlmodel import Field, SQLModel


def _now() -> datetime:
    return datetime.now(UTC)


class FeatureRequest(SQLModel, table=True):
    __tablename__ = "feature_request"

    id: int | None = Field(default=None, primary_key=True)
    summary: str
    description: str
    customer: str
    source_channel: str
    slack_thread_url: str
    urgency: str
    status: str = "open"
    created_at: datetime = Field(default_factory=_now)

"""Create the feature_request table.

Revision ID: 0001
Revises:
Create Date: 2026-10-27

"""

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0001"
down_revision: str | Sequence[str] | None = None
branch_labels: str | Sequence[str] | None = ("feature_request",)
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    op.create_table(
        "feature_request",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("summary", sa.String(), nullable=False),
        sa.Column("description", sa.String(), nullable=False),
        sa.Column("customer", sa.String(), nullable=False),
        sa.Column("source_channel", sa.String(), nullable=False),
        sa.Column("slack_thread_url", sa.String(), nullable=False),
        sa.Column("urgency", sa.String(), nullable=False),
        sa.Column("status", sa.String(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
    )


def downgrade() -> None:
    op.drop_table("feature_request")

"""Reads and writes of the ``feature_request`` table."""

from sqlmodel import col, select

from db.engine import get_session_maker
from db.models.feature_request import FeatureRequest


async def create_feature_request(request: FeatureRequest) -> FeatureRequest:
    async with get_session_maker()() as session:
        session.add(request)
        await session.commit()
        await session.refresh(request)
    return request


async def get_feature_request(request_id: int) -> FeatureRequest | None:
    async with get_session_maker()() as session:
        return await session.get(FeatureRequest, request_id)


async def list_feature_requests(limit: int = 50) -> list[FeatureRequest]:
    async with get_session_maker()() as session:
        rows = await session.execute(
            select(FeatureRequest).order_by(col(FeatureRequest.id).desc()).limit(limit)
        )
        return list(rows.scalars())

"""The tracker API. Every route needs a caller signed in through the Apps gateway."""

from fastapi import HTTPException, Request, params, status
from mistralai_capabilities.apps.tokens import token_from


def require_caller(request: Request) -> None:
    # The gateway signs the caller in and forwards a token; a request without one bypassed it.
    if not token_from(request.headers):
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED, "Call this API through the app's gateway"
        )


dependencies: tuple[params.Depends, ...] = (params.Depends(require_caller),)

from datetime import datetime
from typing import Literal

from db.accessors.feature_request import (
    create_feature_request,
    get_feature_request,
    list_feature_requests,
)
from db.models.feature_request import FeatureRequest
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel

router = APIRouter(tags=["feature_requests"])

Urgency = Literal["low", "medium", "high"]


class NewFeatureRequest(BaseModel):
    summary: str
    description: str
    customer: str
    source_channel: str
    slack_thread_url: str
    urgency: Urgency


class FeatureRequestOut(NewFeatureRequest):
    id: int
    status: str
    created_at: datetime


@router.post("", status_code=status.HTTP_201_CREATED)
async def create(body: NewFeatureRequest) -> FeatureRequestOut:
    """File a feature request and return it with its ID."""
    record = await create_feature_request(FeatureRequest(**body.model_dump()))
    return FeatureRequestOut.model_validate(record, from_attributes=True)


@router.get("")
async def list_all(limit: int = 50) -> list[FeatureRequestOut]:
    """List the most recent feature requests, newest first."""
    return [
        FeatureRequestOut.model_validate(r, from_attributes=True)
        for r in await list_feature_requests(limit)
    ]


@router.get("/{request_id}")
async def get(request_id: int) -> FeatureRequestOut:
    """Get one feature request by ID."""
    record = await get_feature_request(request_id)
    if record is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "No such feature request")
    return FeatureRequestOut.model_validate(record, from_attributes=True)

"""The Slack triage activities: read the message, file it, and reply in its thread.

Intercom and the tracker are Studio HTTP connectors. The connectors gateway adds each connector's stored
credential, so this code holds only the Mistral API key and the connector names.
"""

from datetime import timedelta
from typing import Any, Literal

import httpx
from env.mistral import env as mistral_env
from env.triage import env
from mistralai import workflows
from pydantic import BaseModel
from utils.mistral import mistral_api_url

TIMEOUT = timedelta(seconds=30)
INTERCOM_HEADERS = {"Intercom-Version": "2.11"}


class Classification(BaseModel):
    category: Literal["bug", "feature_request"]
    title: str
    urgency: Literal["low", "medium", "high"]


class SlackMessage(BaseModel):
    channel_id: str
    user_id: str
    ts: str
    text: str
    author: str = ""
    channel: str = ""
    permalink: str = ""


async def _slack(method: str, **params: str) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=20) as client:
        res = await client.post(
            f"https://slack.com/api/{method}",
            headers={"Authorization": f"Bearer {env.require('slack_bot_token')}"},
            data=params,
        )
    body: dict[str, Any] = res.json()
    if not body.get("ok"):
        raise RuntimeError(f"Slack {method} failed: {body.get('error')}")
    return body


async def call_connector(name: str, method: str, path: str, **kwargs: Any) -> Any:
    """Call an HTTP connector's API at ``path`` through the Studio connectors gateway."""
    if not mistral_env.mistral_api_key:
        raise RuntimeError("MISTRAL_API_KEY is not set")
    async with httpx.AsyncClient(timeout=20) as client:
        res = await client.request(
            method,
            f"{mistral_api_url()}/v1/connectors-gateway/{name}/http{path}",
            headers={
                "Authorization": f"Bearer {mistral_env.mistral_api_key}",
                **kwargs.pop("headers", {}),
            },
            **kwargs,
        )
    res.raise_for_status()
    return res.json()


@workflows.activity(start_to_close_timeout=TIMEOUT)
async def describe_message(message: SlackMessage) -> SlackMessage:
    user = (await _slack("users.info", user=message.user_id))["user"]
    channel = (await _slack("conversations.info", channel=message.channel_id))[
        "channel"
    ]
    link = await _slack(
        "chat.getPermalink", channel=message.channel_id, message_ts=message.ts
    )
    return message.model_copy(
        update={
            "author": user["profile"].get("display_name")
            or user.get("real_name")
            or user["name"],
            "channel": channel["name"],
            "permalink": link["permalink"],
        }
    )


# Not retried: a timeout after the record was created would file it twice.
@workflows.activity(start_to_close_timeout=TIMEOUT, retry_policy_max_attempts=1)
async def create_feature_request(message: SlackMessage, triage: Classification) -> str:
    record = await call_connector(
        env.require("tracker_connector_name"),
        "POST",
        "/api/v1/feature_requests",
        json={
            "summary": triage.title,
            "description": message.text,
            "customer": message.author,
            "source_channel": f"#{message.channel}",
            "slack_thread_url": message.permalink,
            "urgency": triage.urgency,
        },
    )
    return f"Feature request #{record['id']}"


@workflows.activity(start_to_close_timeout=TIMEOUT, retry_policy_max_attempts=1)
async def create_bug_ticket(message: SlackMessage, triage: Classification) -> str:
    intercom = env.require("intercom_connector_name")

    # A ticket needs an existing contact; a 409 means one was created for this Slack user before.
    external_id = f"slack-{message.user_id}"
    try:
        await call_connector(
            intercom,
            "POST",
            "/contacts",
            headers=INTERCOM_HEADERS,
            json={"role": "user", "external_id": external_id, "name": message.author},
        )
    except httpx.HTTPStatusError as e:
        if e.response.status_code != 409:
            raise

    ticket = await call_connector(
        intercom,
        "POST",
        "/tickets",
        headers=INTERCOM_HEADERS,
        json={
            "ticket_type_id": env.require("intercom_ticket_type_id"),
            "contacts": [{"external_id": external_id}],
            "ticket_attributes": {
                "_default_title_": triage.title,
                "_default_description_": (
                    f"{message.text}\n\nUrgency: {triage.urgency}\nReported in #{message.channel}: {message.permalink}"
                ),
            },
        },
    )
    workspace = env.require("intercom_workspace_id")
    app_url = (env.intercom_app_url or "https://app.intercom.com").rstrip("/")
    return f"{app_url}/a/inbox/{workspace}/inbox/conversation/{ticket['id']}"


@workflows.activity(start_to_close_timeout=TIMEOUT)
async def reply_in_thread(message: SlackMessage, text: str) -> None:
    await _slack(
        "chat.postMessage", channel=message.channel_id, thread_ts=message.ts, text=text
    )

from typing import Any

from mistralai import workflows
from mistralai.workflows import workflow
from pydantic import BaseModel

with workflow.unsafe.imports_passed_through():
    from env.triage import env
    from mistralai.client.models import (
        ChatCompletionRequest,
        SystemMessage,
        UserMessage,
    )
    from mistralai.workflows.plugins.mistralai import chat_parse_to_model

    from worker.activities import (
        Classification,
        SlackMessage,
        create_bug_ticket,
        create_feature_request,
        describe_message,
        reply_in_thread,
    )

PROMPT = """You triage messages posted in a customer support Slack channel.
Classify the message as a "bug" (something is broken or behaves unexpectedly) or a
"feature_request" (a request for something new or different). If unsure, call it a bug.
Also give a short title (under 80 characters) and an urgency: low, medium or high."""


class SlackEvent(BaseModel):
    """The HTTP request a Studio Slack trigger delivers; the Slack event is in `body.event`."""

    body: dict[str, Any]


class TriageResult(BaseModel):
    category: str
    title: str
    urgency: str
    reference: str


@workflows.workflow.define(
    name="slack-triage",
    workflow_display_name="Slack triage",
    workflow_description=(
        "Classify a Slack message as a bug or a feature request, file it in Intercom or the tracker, "
        "and reply in the thread with the result."
    ),
)
class SlackTriage:
    @workflows.workflow.entrypoint
    async def run(self, request: SlackEvent) -> TriageResult | None:
        event = request.body["event"]
        # The trigger filter skips these too; this check keeps the bot from answering its own replies
        # if the filter is left out.
        if event.get("bot_id") or event.get("subtype") or event.get("thread_ts"):
            return None
        message = await describe_message(
            SlackMessage(
                channel_id=event["channel"],
                user_id=event["user"],
                ts=event["ts"],
                text=event["text"],
            )
        )

        triage = await chat_parse_to_model(
            Classification,
            ChatCompletionRequest(
                model=env.triage_model,
                messages=[
                    SystemMessage(content=PROMPT),
                    UserMessage(content=message.text),
                ],
            ),
        )

        if triage.category == "bug":
            link = await create_bug_ticket(message, triage)
            reply = f"Thanks {message.author}, this looks like a bug. I opened a support ticket: <{link}|{triage.title}>"
        else:
            link = await create_feature_request(message, triage)
            reply = f"Thanks {message.author}, I filed this as {link}: {triage.title}"

        await reply_in_thread(message, reply)
        return TriageResult(
            category=triage.category,
            title=triage.title,
            urgency=triage.urgency,
            reference=link,
        )

"""Create a Studio HTTP connector whose credential is shared with the workspace."""

from typing import Annotated

import typer
from mistralai.client.models import (
    BearerAuthMethod,
    ConnectorAuthenticationHeader,
    CreateHTTPConnectorRequest,
    NoneAuthMethod,
)
from utils.mistral import mistral_client

app = typer.Typer()


@app.command(name="connector", help=__doc__)
def main(
    name: Annotated[
        str, typer.Argument(help="Connector name, for example intercom or tracker")
    ],
    url: Annotated[
        str, typer.Argument(help="API base URL, for example https://api.intercom.io")
    ],
    token: Annotated[
        str,
        typer.Option(prompt=True, hide_input=True, help="Token the connector sends"),
    ],
    header: Annotated[
        str | None,
        typer.Option(
            help="Send the token in this header, such as X-Api-Key, instead of as a bearer token"
        ),
    ] = None,
) -> None:
    if header:
        auth = NoneAuthMethod(headers=[ConnectorAuthenticationHeader(name=header)])
        credentials = {"headers": {header: token.strip()}}
    else:
        auth = BearerAuthMethod()
        credentials = {"bearer_token": token.strip()}
    client = mistral_client()
    client.beta.connectors.create(
        request=CreateHTTPConnectorRequest(
            name=name,
            title=name,
            description=f"{name} API",
            server=url,
            visibility="shared_workspace",
            auth_methods=[auth],
        )
    )
    # Workspace scope: a credential scoped to your user is invisible to triggered runs.
    client.beta.connectors.create_credentials(
        connector_id_or_name=name,
        consumer_scope="workspace",
        name="default",
        credentials=credentials,
    )
    typer.echo(f"Created connector {name} -> {url} with a workspace credential")