FastAPI Mistral Ticket Router: Safe GCC Guide

🚀 Technical Briefing: This tutorial is part of our deep-dive series on Agentic Workflows at Gate of AI. For the full technical breakdown, interactive code sandbox, and the native Arabic translation, visit the original article here.

Build a secure FastAPI ticket-routing foundation with strict request validation, deterministic escalation policies, SQLite audit records, and a safe integration boundary for a verified Mistral API implementation.

Important Accuracy Note Before You Start

The supplied source material confirms that Mistral AI is a Paris-based AI company and describes it as an OpenAI competitor. It does not provide official API documentation, supported model IDs, Python SDK package details, chat-completion method signatures, structured-output parameters, pricing, regional hosting commitments, or deployment guarantees.

For that reason, this tutorial does not present unverified Mistral SDK code as production-ready fact. Instead, it builds a complete, runnable FastAPI ticket router with a deterministic local classifier and a clearly defined provider boundary. After consulting current official Mistral documentation and completing your own evaluation, replace the local classifier implementation at that boundary with the verified Mistral integration appropriate to your account and approved model.

This is a safer engineering approach than copying an unverified model alias, SDK method, or JSON-mode option into a customer-facing workflow. The rest of the service—input contracts, audit storage, authorization, routing policy, tests, and operational controls—remains useful regardless of which approved AI provider or model you connect later.

What You Will Build

You will build a FastAPI service exposing POST /tickets/classify. A client sends a support ticket with a ticket ID, subject, body, customer tier, and source. The service validates the request, classifies it into a limited taxonomy, applies deterministic escalation rules, records an audit event in SQLite, and returns a typed JSON response.

The working baseline intentionally uses transparent keyword rules rather than pretending that an unverified external model call is available. It is not intended to replace an evaluated LLM. Its purpose is to give your team a safe, testable routing baseline and a precise interface for an eventual Mistral-backed classifier.

This architecture is useful for service desks, customer support teams, internal IT queues, security intake, logistics exceptions, and account-management workflows. It is particularly relevant in GCC organisations adopting AI under programmes such as Saudi Vision 2030 and the UAE National Strategy for Artificial Intelligence: automation should preserve review paths, clear ownership, and auditable decisions when tickets may contain customer, employee, or security-sensitive information.

Prerequisites and Installation

• Python 3.10 or newer.
• Basic familiarity with virtual environments, HTTP APIs, and JSON.
• FastAPI, Uvicorn, Pydantic Settings, and pytest.
• A future Mistral API account only when you are ready to implement the provider adapter from current official documentation.

Create the project and install dependencies:

mkdir fastapi-ticket-router
cd fastapi-ticket-router

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip

cat > requirements.txt <<'EOF'
fastapi>=0.115.0
uvicorn[standard]>=0.30.0
pydantic>=2.8.0
pydantic-settings>=2.4.0
pytest>=8.3.0
EOF

pip install -r requirements.txt
mkdir -p app tests
touch app/__init__.py

Create a local environment file. The application API key protects your own endpoint; it is separate from any future AI-provider credential. Do not commit this file to source control.

cat > .env <<'EOF'
APP_API_KEY=replace-with-a-long-random-secret
DATABASE_PATH=./ticket_router.db
MAX_TICKET_CHARACTERS=12000
EOF

cat > .gitignore <<'EOF'
.venv/
.env
__pycache__/
.pytest_cache/
*.pyc
ticket_router.db
EOF

For production, place secrets in the encrypted secret-management facility approved by your cloud or platform team. Never send an AI-provider API key to browser JavaScript, mobile clients, public repositories, or client-side environment variables.

Step 1: Define Strict Contracts and Configuration

LLM integration should sit behind a deterministic API contract. Downstream systems should receive a finite set of categories, bounded confidence values, and explicit review flags—not arbitrary free-form text. Create app/config.py and app/schemas.py:

cat > app/config.py <<'EOF'
from functools import lru_cache
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    app_api_key: SecretStr
    database_path: str = "./ticket_router.db"
    max_ticket_characters: int = Field(default=12000, ge=500, le=50000)


@lru_cache
def get_settings() -> Settings:
    return Settings()
EOF

cat > app/schemas.py <<'EOF'
from enum import Enum
from pydantic import BaseModel, Field, field_validator


class CustomerTier(str, Enum):
    free = "free"
    standard = "standard"
    business = "business"
    enterprise = "enterprise"


class TicketCategory(str, Enum):
    billing = "billing"
    account_access = "account_access"
    technical_issue = "technical_issue"
    security = "security"
    sales = "sales"
    feature_request = "feature_request"
    cancellation = "cancellation"
    abuse = "abuse"
    other = "other"


class Urgency(str, Enum):
    low = "low"
    normal = "normal"
    high = "high"
    critical = "critical"


class TicketInput(BaseModel):
    ticket_id: str = Field(min_length=3, max_length=100, pattern=r"^[A-Za-z0-9_-]+$")
    subject: str = Field(min_length=3, max_length=300)
    body: str = Field(min_length=10, max_length=12000)
    customer_tier: CustomerTier = CustomerTier.standard
    source: str = Field(default="api", max_length=50)

    @field_validator("subject", "body")
    @classmethod
    def reject_blank_text(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("must not be blank")
        return value


class ClassificationResult(BaseModel):
    category: TicketCategory
    urgency: Urgency
    confidence: float = Field(ge=0.0, le=1.0)
    recommended_team: str = Field(min_length=2, max_length=80)
    customer_summary: str = Field(min_length=10, max_length=500)
    requires_human_review: bool
    review_reason: str | None = Field(default=None, max_length=300)


class ClassificationResponse(BaseModel):
    ticket_id: str
    classifier: str
    result: ClassificationResult
EOF

The enums are a safety boundary. If a future AI model proposes a category outside this list, validate and reject it before any routing action occurs. Keep prompt instructions, provider output, and final routing policy separate.

Step 2: Add the Deterministic Classifier and Policy Layer

Create a baseline classifier. Security-related language is always routed to Security Operations and always requires human review. This is deliberate: high-risk escalation should not depend solely on a probability returned by a model.

cat > app/classifier.py <<'EOF'
from app.schemas import ClassificationResult, TicketCategory, TicketInput, Urgency

SECURITY_TERMS = ("api key", "credential", "breach", "compromised", "unauthorized", "fraud")
BILLING_TERMS = ("invoice", "refund", "charge", "payment", "subscription")
ACCESS_TERMS = ("login", "password", "mfa", "sso", "locked out")
TECHNICAL_TERMS = ("error", "outage", "bug", "api", "integration", "slow")


def apply_safety_policy(result: ClassificationResult) -> ClassificationResult:
    updated = result.model_copy(deep=True)
    if updated.category == TicketCategory.security:
        updated.recommended_team = "Security Operations"
        updated.requires_human_review = True
        updated.review_reason = "Security-related ticket requires human verification."
    elif updated.urgency == Urgency.critical:
        updated.requires_human_review = True
        updated.review_reason = "Critical urgency requires immediate human review."
    elif updated.confidence < 0.70:
        updated.requires_human_review = True
        updated.review_reason = "Confidence is below the automated-routing threshold."
    elif not updated.requires_human_review:
        updated.review_reason = None
    return updated


def classify_ticket(ticket: TicketInput) -> ClassificationResult:
    text = f"{ticket.subject} {ticket.body}".lower()
    category = TicketCategory.other
    team = "General Support"
    urgency = Urgency.normal
    confidence = 0.72

    if any(term in text for term in SECURITY_TERMS):
        category, team, urgency, confidence = TicketCategory.security, "Security Operations", Urgency.high, 0.90
    elif any(term in text for term in BILLING_TERMS):
        category, team, confidence = TicketCategory.billing, "Billing Support", 0.82
    elif any(term in text for term in ACCESS_TERMS):
        category, team, confidence = TicketCategory.account_access, "Identity Support", 0.82
    elif any(term in text for term in TECHNICAL_TERMS):
        category, team, confidence = TicketCategory.technical_issue, "Technical Support", 0.78

    result = ClassificationResult(
        category=category,
        urgency=urgency,
        confidence=confidence,
        recommended_team=team,
        customer_summary=f"Ticket received: {ticket.subject}",
        requires_human_review=False,
        review_reason=None,
    )
    return apply_safety_policy(result)
EOF

When implementing the future Mistral adapter, preserve the function contract: accept TicketInput, return ClassificationResult, validate all provider output with Pydantic, and call apply_safety_policy after validation. Do not let a provider response directly select an operational queue.

Step 3: Store Audit Records in SQLite

An audit record should identify the ticket, classifier, final validated decision, and timestamp. SQLite is suitable for local development, demos, and single-instance low-volume services. Use a managed relational database with backups, access controls, and migrations for larger production workloads.

cat > app/database.py <<'EOF'
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from app.schemas import ClassificationResponse


def connection(path: str) -> sqlite3.Connection:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    return sqlite3.connect(path)


def initialize_database(path: str) -> None:
    with connection(path) as db:
        db.execute("""
        CREATE TABLE IF NOT EXISTS ticket_audits (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ticket_id TEXT NOT NULL,
            classifier TEXT NOT NULL,
            result_json TEXT NOT NULL,
            created_at TEXT NOT NULL
        )
        """)


def save_audit(path: str, response: ClassificationResponse) -> None:
    with connection(path) as db:
        db.execute(
            "INSERT INTO ticket_audits(ticket_id, classifier, result_json, created_at) VALUES (?, ?, ?, ?)",
            (response.ticket_id, response.classifier,
             json.dumps(response.result.model_dump(mode="json")),
             datetime.now(timezone.utc).isoformat()),
        )
EOF

Audit data can contain personal or commercially sensitive information. Define retention, deletion, access, encryption, and incident-response procedures before processing production tickets. GCC deployments should also assess applicable contractual, sectoral, and data-residency requirements with qualified legal and security stakeholders rather than assuming that a particular cloud or AI configuration meets local obligations.

Step 4: Expose the FastAPI Endpoint

cat > app/main.py <<'EOF'
import hmac
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, Header, HTTPException, status
from app.classifier import classify_ticket
from app.config import Settings, get_settings
from app.database import initialize_database, save_audit
from app.schemas import ClassificationResponse, TicketInput


@asynccontextmanager
async def lifespan(app: FastAPI):
    initialize_database(get_settings().database_path)
    yield


app = FastAPI(title="FastAPI Ticket Router", version="1.0.0", lifespan=lifespan)


def require_api_key(
    x_api_key: str | None = Header(default=None, alias="X-API-Key"),
    settings: Settings = Depends(get_settings),
) -> None:
    expected = settings.app_api_key.get_secret_value()
    if x_api_key is None or not hmac.compare_digest(x_api_key, expected):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing API key")


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/tickets/classify", response_model=ClassificationResponse, dependencies=[Depends(require_api_key)])
async def route_ticket(ticket: TicketInput, settings: Settings = Depends(get_settings)) -> ClassificationResponse:
    if len(ticket.body) > settings.max_ticket_characters:
        raise HTTPException(status_code=422, detail="Ticket body exceeds configured maximum")
    response = ClassificationResponse(
        ticket_id=ticket.ticket_id,
        classifier="deterministic-baseline",
        result=classify_ticket(ticket),
    )
    save_audit(settings.database_path, response)
    return response
EOF

uvicorn app.main:app --reload --host 127.0.0.1 --port 8000

Test the endpoint in another terminal:

curl --request POST http://127.0.0.1:8000/tickets/classify \
  --header "Content-Type: application/json" \
  --header "X-API-Key: replace-with-a-long-random-secret" \
  --data '{"ticket_id":"SUP-1042","subject":"Possible API key exposure","body":"A production API key appeared in CI logs. Please investigate unauthorized use.","customer_tier":"enterprise","source":"support"}'

The response must route the ticket to Security Operations and set requires_human_review to true. This is enforced by code, not by an optional model instruction.

Test the Non-Negotiable Policy

cat > tests/test_policy.py <<'EOF'
from app.classifier import classify_ticket
from app.schemas import TicketInput


def test_security_ticket_requires_human_review() -> None:
    result = classify_ticket(TicketInput(
        ticket_id="SUP-1",
        subject="Credential exposure",
        body="Our API key may be compromised and used by an unknown party.",
    ))
    assert result.recommended_team == "Security Operations"
    assert result.requires_human_review is True
    assert result.review_reason is not None
EOF

pytest -q

Do not unit-test live provider calls as part of the standard test suite. Live calls are network-dependent, potentially billable, and variable. Instead, use unit tests for schemas and deterministic policy; then run controlled integration tests against an approved provider environment using sanitized fixtures.

Replacing the Baseline with a Verified Mistral Integration

Before changing the classifier, obtain the current official Mistral documentation for your account and verify the SDK version, authentication approach, supported model ID, request schema, response schema, structured-output option, error handling, rate limits, and data-processing terms. None of these implementation details are established in the supplied context.

Implement the provider adapter behind the existing classifier boundary. Serialize ticket content as data rather than instructions; validate the resulting JSON against ClassificationResult; reject malformed or out-of-taxonomy values; record the approved model identifier in the audit trail; and apply apply_safety_policy after the model output is validated.

Evaluate the integration on a versioned, sanitized dataset before release. Track category quality, security false negatives, human-review rate, latency, schema-valid response rate, override rate, and cost per ticket. For critical security, financial, healthcare, or public-sector workflows, design a safe failure mode: if the AI provider is unavailable or output validation fails, send the ticket to a human-review queue instead of silently guessing.

For organisations working with SDAIA, G42, NEOM, stc, e&, or other regional enterprises, the key question is not simply whether an AI model can classify a ticket. The production question is whether the end-to-end system has approved data handling, accountable operators, reliable auditability, and a documented escalation path. This FastAPI foundation supports that discipline while leaving the model provider replaceable.