FastAPI Mistral Ticket Router: Safe GCC Guide

Gate of AI published a tutorial on building 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. The tutorial deliberately avoids presenting unverified Mistral SDK code as production-ready, instead offering a deterministic local classifier and a clearly defined provider boundary. It is aimed at GCC organizations adopting AI under programs such as Saudi Vision 2030 and the UAE National Strategy for Artificial Intelligence.

🚀 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. 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. 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. 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. 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 : python 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. 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. python 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. 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. python 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. python 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. python 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. 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.