Build a Smart IT Help Desk using AI with SendGrid, Twilio and Qdrant

Time to read: 14 minutes

June 24, 2025
Written by
Jacob Muganda
Contributor
Opinions expressed by Twilio contributors are their own
Reviewed by
Amanda Lange
Twilion

Build a Smart IT Help Desk using AI with SendGrid, Twilio and Qdrant

What if your help desk could fix issues as they are being raised? Picture a support engine that grows smarter with every conversation, instantly triaging incoming issues and knowing when to bring a human into the loop by escalating only what truly demands human attention. An automated system like this would leave your team to focus on innovation, not inbox overload.

In this hands‑on tutorial, you’ll assemble a Smart IT Help Desk System that:

  • Listens on WhatsApp via the Twilio Business API for instant, familiar user interactions.
  • Analyzes issues with Large Language Models, turning raw symptom data into clear, actionable diagnostics.
  • Indexes and searches troubleshooting guides in Qdrant’s vector database to ground every answer in your own domain knowledge.
  • Automates ticket workflows classification, priority‑setting, asset updates, slashing resolution times.
  • Supports multimodal input by accepting screenshots or photos of errors, so agents see exactly what users see, cutting ambiguity and speeding fixes.
  • Escalates critical incidents with an automated email alert to your admin team for urgent, on‑site intervention and satisfaction surveys.

Prerequisites

Before you begin, make sure you have the following:

  • A Twilio account with the WhatsApp sandbox enabled. (You’ll use this for testing.)

You’ll need your Account SID, Auth Token, and the sandbox WhatsApp number.

  • An OpenAI API key for GPT-4o (visit o penai platform to sign up).
  • A Twilio SendGrid account and API key for sending emails (alerts and surveys).
  • Python 3.13+ installed. (You’ll use FastAPI for the server).
  • Ngrok (or Pyngrok) to expose your local server for Twilio webhooks.
  • Familiarity with Python and APIs.

Project Setup

First, clone the source code repository. You can find the full code on GitHub. In your terminal, run:

git clone https://github.com/lizpart/Smart-IT-HelpDesk.git
cd Smart-IT-HelpDesk

Then create & activate a virtual environment. Run: 

python -m venv smart-helpdesk
smart-helpdesk\Scripts\activate 
#On Mac:  
python3 -m venv smart-helpdesk
source smart-helpdesk/bin/activate

Next, install the required Python packages. Make sure you're in the directory where the requirements.txt file is located. Run the command below:

pip install -r requirements.txt

This pulls in:

  • FastAPI and uvicorn for the web server.
  • twilio helper library for WhatsApp messaging.
  • sendgrid for email API.
  • openai for calling GPT-4o (through LangChain).
  • qdrant-client and langchain-openai for vector search.
  • python-dotenv to load .env files.
  • pyngrok to manage an ngrok tunnel from Python.

Create a .env file in the project root to hold your configuration. Using a .env ensures you keep credentials out of source code. FastAPI, via Pydantic’s BaseSettings, will automatically load these variables. Setting env_file = ".env" in the Settings class ensures these values are available in settings. Replace these values with actual credentials: 

TWILIO_ACCOUNT_SID=ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
TWILIO_AUTH_TOKEN=your_auth_token_here
TWILIO_PHONE_NUMBER=+1123456789        # Twilio Sandbox number
OPENAI_API_KEY=sk-XXXXXXXXXXXXXXXXXXXXXXX
NGROK_AUTH_TOKEN=your_ngrok_auth_token
SENDGRID_API_KEY=SG.XXXXXXXXXXXXXXXXXXXXXX
FROM_EMAIL=you@yourdomain.com
NOTIFICATION_EMAIL=support-team@company.com
USER_EMAILS={"1234567890": "alice@example.com", "0987654321": "bob@example.com"}

Your application uses these environment variables to keep secrets and configuration out of your code. TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN together authenticate and authorize your app to send SMS or WhatsApp messages through your Twilio account. These can be found on your Twilio dashboard.

TWILIO_PHONE_NUMBER is the E.164‑formatted number you send from (whether sandbox or production). When sending from the Twilio WhatsApp Sandbox, this value will be shown on the WhatsApp Sandbox Dashboard under From. OPENAI_API_KEY lets your code call OpenAI’s APIs for text, embeddings, or other AI services. NGROK_AUTH_TOKEN connects your local server to a public URL so you can receive webhook callbacks during development. SENDGRID_API_KEY secures your access to SendGrid’s email‑sending service.FROM_EMAIL and NOTIFICATION_EMAIL define the sender address for outgoing emails and the recipient for system alerts respectively. Finally, USER_EMAILS holds a JSON mapping of user phone numbers to their email addresses so you can link incoming messages to the correct user records. With all these understood, proceed to creating your application.

Build the Application

To begin, dive into the Python application app.py. This FastAPI app orchestrates the chatbot’s logic. This tutorial will now review the main parts of the code section by section to understand the application at a low level.If you want to try out the application yourself, jump down to the header to run your server.

Configuration and Settings 

import os
from dotenv import load_dotenv
from pydantic_settings import BaseSettings
# Load environment variables
load_dotenv()
class Settings(BaseSettings):
    TWILIO_ACCOUNT_SID: str = os.getenv("TWILIO_ACCOUNT_SID", "")
    TWILIO_AUTH_TOKEN: str = os.getenv("TWILIO_AUTH_TOKEN", "")
    TWILIO_PHONE_NUMBER: str = "+14155238886"  # Twilio sandbox number
    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    NGROK_AUTH_TOKEN: str = os.getenv("NGROK_AUTH_TOKEN", "")
    SENDGRID_API_KEY: str = os.getenv("SENDGRID_API_KEY", "")
    NOTIFICATION_EMAIL: str = os.getenv("NOTIFICATION_EMAIL", "")
    FROM_EMAIL: str = os.getenv("FROM_EMAIL", "")
    USER_EMAILS: dict = json.loads(os.getenv("USER_EMAILS", "{}"))
    class Config:
        env_file = ".env"
        extra = "allow"

The load_dotenv() reads variables from the .env file into the environment. The class Settings(BaseSettings) defines your configuration schema. It then uses Pydantic’s BaseSettings so you can override these with environment variables. Each attribute (e.g., TWILIO_ACCOUNT_SID) pulls from os.getenv. If not set, defaults to an empty string.

The inner Config class tells Pydantic to load .env and allow extra fields.

Finally, instantiate settings: settings = Settings(). All keys from .env are now in settings.

Initialization: Ngrok and FastAPI App 

from pyngrok import ngrok, conf as pyngrok_conf
from fastapi import FastAPI
# Configure ngrok
pyngrok_conf.get_default().auth_token = settings.NGROK_AUTH_TOKEN
# Initialize FastAPI app
app = FastAPI()

This code imports pyngrok and sets its auth token to can start tunnels programmatically.

You then create the FastAPI app instance (app = FastAPI()), which will serve your webhook endpoints. Next, move onto memory management.

Class Conversation Memory

This section handles recall. 

import sqlite3
from threading import Lock
from datetime import datetime, timedelta
import uuid
import logging
from typing import Optional, List, Dict
# Also uses: settings (from your Settings class)
class ConversationMemory:
    """Handles conversation history storage and retrieval"""
    def __init__(self, db_path: str = "conversations.db"):
        self.db_path = db_path
        self.lock = Lock()
        self._initialize_database()
    def _initialize_database(self):
        """Initialize SQLite database for conversation storage"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    CREATE TABLE IF NOT EXISTS conversations (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        user_id TEXT NOT NULL,
                        message_type TEXT NOT NULL,  -- 'user' or 'assistant'
                        content TEXT NOT NULL,
                        media_url TEXT,
                        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                        session_id TEXT NOT NULL
                    )
                ''')
                # Create index for faster queries
                cursor.execute('''
                    CREATE INDEX IF NOT EXISTS idx_user_timestamp 
                    ON conversations(user_id, timestamp DESC)
                ''')
                conn.commit()
                logger.info("Conversation database initialized successfully")
        except Exception as e:
            logger.error(f"Error initializing conversation database: {e}")
            raise
    def _get_session_id(self, user_id: str) -> str:
        """Get or create session ID for user based on recent activity"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                # Check for recent conversation (within timeout period)
                timeout_timestamp = datetime.now() - timedelta(hours=settings.CONVERSATION_TIMEOUT_HOURS)
                cursor.execute('''
                    SELECT session_id FROM conversations 
                    WHERE user_id = ? AND timestamp > ? 
                    ORDER BY timestamp DESC LIMIT 1
                ''', (user_id, timeout_timestamp))
                result = cursor.fetchone()
                if result:
                    return result[0]
                else:
                    # Create new session
                    return str(uuid.uuid4())
        except Exception as e:
            logger.error(f"Error getting session ID: {e}")
            return str(uuid.uuid4())
    def add_message(self, user_id: str, message_type: str, content: str, media_url: Optional[str] = None):
        """Add a message to conversation history"""
        try:
            with self.lock:
                session_id = self._get_session_id(user_id)
                with sqlite3.connect(self.db_path) as conn:
                    cursor = conn.cursor()
                    cursor.execute('''
                        INSERT INTO conversations (user_id, message_type, content, media_url, session_id)
                        VALUES (?, ?, ?, ?, ?)
                    ''', (user_id, message_type, content, media_url, session_id))
                    conn.commit()
                    # Clean up old messages to maintain conversation limit
                    self._cleanup_old_messages(cursor, user_id, session_id)
                    conn.commit()
        except Exception as e:
            logger.error(f"Error adding message to conversation history: {e}")
    def _cleanup_old_messages(self, cursor, user_id: str, session_id: str):
        """Remove old messages beyond the conversation limit"""
        try:
            # Get message count for current session
            cursor.execute('''
                SELECT COUNT(*) FROM conversations 
                WHERE user_id = ? AND session_id = ?
            ''', (user_id, session_id))
            count = cursor.fetchone()[0]
            if count > settings.MAX_CONVERSATION_HISTORY:
                # Remove oldest messages beyond limit
                excess_count = count - settings.MAX_CONVERSATION_HISTORY
                cursor.execute('''
                    DELETE FROM conversations 
                    WHERE user_id = ? AND session_id = ?
                    AND id IN (
                        SELECT id FROM conversations 
                        WHERE user_id = ? AND session_id = ?
                        ORDER BY timestamp ASC 
                        LIMIT ?
                    )
                ''', (user_id, session_id, user_id, session_id, excess_count))
        except Exception as e:
            logger.error(f"Error cleaning up old messages: {e}")
    def get_conversation_history(self, user_id: str) -> List[Dict]:
        """Get recent conversation history for user"""
        try:
            session_id = self._get_session_id(user_id)
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    SELECT message_type, content, media_url, timestamp 
                    FROM conversations 
                    WHERE user_id = ? AND session_id = ?
                    ORDER BY timestamp ASC
                    LIMIT ?
                ''', (user_id, session_id, settings.MAX_CONVERSATION_HISTORY))
                results = cursor.fetchall()
                history = []
                for row in results:
                    message_type, content, media_url, timestamp = row
                    history.append({
                        'role': message_type,
                        'content': content,
                        'media_url': media_url,
                        'timestamp': timestamp
                    })
                return history
        except Exception as e:
            logger.error(f"Error retrieving conversation history: {e}")
            return []
    def clear_user_history(self, user_id: str):
        """Clear conversation history for a specific user"""
        try:
            with self.lock:
                with sqlite3.connect(self.db_path) as conn:
                    cursor = conn.cursor()
                    cursor.execute('DELETE FROM conversations WHERE user_id = ?', (user_id,))
                    conn.commit()
                    logger.info(f"Cleared conversation history for user: {user_id}")
        except Exception as e:
            logger.error(f"Error clearing user history: {e}")
    def cleanup_expired_conversations(self):
        """Remove conversations older than timeout period"""
        try:
            with self.lock:
                timeout_timestamp = datetime.now() - timedelta(hours=settings.CONVERSATION_TIMEOUT_HOURS * 2)  # Double timeout for cleanup
                with sqlite3.connect(self.db_path) as conn:
                    cursor = conn.cursor()
                    cursor.execute('DELETE FROM conversations WHERE timestamp < ?', (timeout_timestamp,))
                    deleted_count = cursor.rowcount
                    conn.commit()
                    if deleted_count > 0:
                        logger.info(f"Cleaned up {deleted_count} expired conversation messages")
        except Exception as e:
            logger.error(f"Error cleaning up expired conversations: {e}")

The ConversationMemory class is designed to manage the conversation history between users and the assistant in a seamless, ongoing manner. When a user sends a message, the class first determines whether the message should be grouped with previous messages in an existing session or if a new session should be started, based on how much time has passed since the last interaction. It then stores each message, whether from the user or the assistant, along with relevant metadata such as the message type, content, media URL, timestamp, and session ID in a SQLite database.

As the conversation continues, the class retrieves the most recent messages for a user, up to a configurable limit, to provide context for the assistant’s responses. This allows the assistant to reference earlier parts of the conversation and maintain continuity, making interactions feel more natural and informed. To ensure the database remains efficient and doesn’t grow indefinitely, the class automatically deletes older messages that exceed the set history limit for each session and periodically cleans up conversations that have expired based on a timeout setting.

Additionally, the class provides methods to clear all conversation history for a specific user, which can be triggered by user commands or admin actions, and to perform manual or scheduled cleanup of expired conversations. This continuous management of conversation data enables the assistant to deliver context-aware, multi-turn interactions while keeping resource usage under control.

KnowledgeBase Class

This section demonstrates how to build a simple retrieval‑augmented knowledge store using LangChain’s OpenAI embeddings and a Qdrant vector database. 

from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import CharacterTextSplitter
from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams
class KnowledgeBase:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            openai_api_key=settings.OPENAI_API_KEY,
            model="text-embedding-3-small"  # Using latest embedding model
        )
        self.collection_name = "technical_docs"
        self.vector_size = 1536
        # Initialize Qdrant client with better error handling
        try:
            self.qdrant_client = QdrantClient(path="./qdrant_storage")
            self.initialize_knowledge_base()
        except Exception as e:
            logger.error(f"Failed to initialize Qdrant client: {e}")
            # Fallback to in-memory client
            self.qdrant_client = QdrantClient(":memory:")
            logger.info("Using in-memory Qdrant client as fallback")
            self.initialize_knowledge_base()
    def initialize_knowledge_base(self):
        try:
            collections = self.qdrant_client.get_collections().collections
            collection_exists = any(col.name == self.collection_name for col in collections)
            if not collection_exists:
                self.qdrant_client.create_collection(
                    collection_name=self.collection_name,
                    vectors_config=VectorParams(size=self.vector_size, distance=Distance.COSINE)
                )
                self._create_knowledge_base()
                logger.info("Created new knowledge base")
            else:
                logger.info("Loaded existing knowledge base")
        except Exception as e:
            logger.error(f"Error initializing knowledge base: {str(e)}")
            # Create a minimal knowledge base for testing
            self._create_minimal_knowledge_base()
    def _create_knowledge_base(self):
        try:
            # Check if technical_docs directory exists
            if not os.path.exists("technical_docs/"):
                logger.warning("technical_docs directory not found, creating minimal knowledge base")
                self._create_minimal_knowledge_base()
                return
            loader = DirectoryLoader("technical_docs/", glob="**/*.txt", loader_cls=TextLoader)
            documents = loader.load()
            if not documents:
                logger.warning("No documents found in technical_docs/, creating minimal knowledge base")
                self._create_minimal_knowledge_base()
                return
            text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
            texts = text_splitter.split_documents(documents)
            for i, doc in enumerate(texts):
                try:
                    embedding = self.embeddings.embed_query(doc.page_content)
                    self.qdrant_client.upsert(
                        collection_name=self.collection_name,
                        points=[models.PointStruct(
                            id=i, 
                            vector=embedding, 
                            payload={"text": doc.page_content}
                        )]
                    )
                except Exception as e:
                    logger.error(f"Error processing document {i}: {e}")
                    continue
        except Exception as e:
            logger.error(f"Error creating knowledge base: {str(e)}")
            self._create_minimal_knowledge_base()
    def _create_minimal_knowledge_base(self):
        """Create a minimal knowledge base for testing purposes"""
        try:
            sample_docs = [
                "For network connectivity issues, check cable connections and restart the router.",
                "If equipment is overheating, ensure proper ventilation and clean air filters.",
                "For software errors, try restarting the application or checking for updates.",
                "Power supply issues often require checking fuses and voltage levels.",
                "Regular maintenance includes cleaning, lubrication, and calibration checks."
            ]
            for i, doc_text in enumerate(sample_docs):
                try:
                    embedding = self.embeddings.embed_query(doc_text)
                    self.qdrant_client.upsert(
                        collection_name=self.collection_name,
                        points=[models.PointStruct(
                            id=i, 
                            vector=embedding, 
                            payload={"text": doc_text}
                        )]
                    )
                except Exception as e:
                    logger.error(f"Error creating minimal knowledge base entry {i}: {e}")
            logger.info("Created minimal knowledge base for testing")
        except Exception as e:
            logger.error(f"Error creating minimal knowledge base: {e}")
    def get_relevant_context(self, query, k=3):
        try:
            query_embedding = self.embeddings.embed_query(query)
            search_result = self.qdrant_client.search(
                collection_name=self.collection_name,
                query_vector=query_embedding,
                limit=k
            )
            context = "\n".join([hit.payload["text"] for hit in search_result])
            return context
        except Exception as e:
            logger.error(f"Error retrieving context: {str(e)}")
            return "Basic troubleshooting: Check connections, restart system, and verify power supply."

The self.embeddings = OpenAIEmbeddings(...) creates a LangChain OpenAIEmbeddings instance (using your GPT API) to turn text into vector embeddings. The collection_name and vector_size, tells your application to store embeddings in a Qdrant collection named technical_docs, with dimensionality 1536 (the size of OpenAI’s text embeddings). self.qdrant_client = QdrantClient(path="./qdrant_storage") then initializes a local Qdrant client that stores data in ./qdrant_storage. The initialize_knowledge_base() checks if the collection exists. If not, it creates it with cosine similarity. The _create_knowledge_base() then loads all .txt files under technical_docs/, splits them into chunks, embeds each chunk with OpenAI, and upserts them into Qdrant with their text as payload. This populates the vector DB. The get_relevant_context(query, k=3) embeds the query, performs a vector search in Qdrant, and returns the top k matching document texts joined together. This gives us context snippets to feed to the API.

Together, this class ensures your technical documentation and troubleshooting guides are indexed for semantic search. Using Qdrant’s graph-accelerated search means you don’t have to compute distances against every document every time; instead, Qdrant finds the closest matches quickly.

DiagnosticAssistant Class

Building upon the KnowledgeBase class, which efficiently stores and retrieves semantically relevant information, you now introduce the DiagnosticAssistant class. This class coordinates the services to run diagnostics. 

from twilio.rest import Client
import openai
import uuid
class DiagnosticAssistant:
    def __init__(self):
        self.twilio_client = Client(settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN)
        self.openai_client = openai.Client(api_key=settings.OPENAI_API_KEY)
        self.knowledge_base = KnowledgeBase()
        self.ngrok_tunnel = None
        self.email_service = EmailService(settings.SENDGRID_API_KEY, settings.FROM_EMAIL, settings.NOTIFICATION_EMAIL)
        self.user_emails = settings.USER_EMAILS

The self.twilio_client = Client(...) initializes the Twilio REST client to send messages. The self.openai_client acts as the client for the OpenAI API (GPT-4o). self.knowledge_base instantiates the KnowledgeBase above, so your vector DB is ready. The self.email_service creates an EmailService (defined in email_service.py) for sending maintenance emails and surveys. self.user_emails then loads the phone-to-email mapping. 

def get_user_email(self, whatsapp_number: str) -> Optional[str]:
        """Get user email from WhatsApp number"""
        clean_number = whatsapp_number.replace('whatsapp:', '')
        return self.user_emails.get(clean_number)

The get_user_email() strips the whatsapp prefix and looks up the user’s email in our dict. If none is found, you won’t send emails to that user. Next, you need to write the code to build the conversation. 

def _build_conversation_context(self, user_id: str, current_message: str) -> List[Dict]:
        """Build conversation context including history and current message"""
        messages = [
            {
                "role": "system",
                "content": """You are a technical support assistant for industrial/technical equipment. 
                You have access to conversation history and should reference previous messages when relevant.
                Provide clear, concise solutions based on the provided context and conversation history.
                If a user refers to a previous issue or asks follow-up questions, use the conversation history to provide contextual responses."""
            }
        ]
        # Get conversation history
        history = self.conversation_memory.get_conversation_history(user_id)
        # Add conversation history to messages
        for msg in history:
            if msg['role'] == 'user':
                content = msg['content']
                if msg['media_url']:
                    content += " [User sent an image]"
                messages.append({"role": "user", "content": content})
            elif msg['role'] == 'assistant':
                messages.append({"role": "assistant", "content": msg['content']})
        # Add current message
        messages.append({"role": "user", "content": current_message})
        return messages

The _build_conversation_context function constructs a list of message dictionaries that represent the conversation context for the AI assistant. It starts by adding a system message that instructs the assistant on its role and how to use conversation history. Next, it retrieves the user's recent conversation history from the ConversationMemory class. For each message in the history, it appends a dictionary to the messages list, distinguishing between user and assistant messages. If a user message includes an image, it notes this in the content. Finally, it adds the current user message to the end of the list. The resulting messages list provides a structured context, including both past exchanges and the current input, which is then used as input for the AI model to generate a relevant, context-aware response. Next, you need to write the code to send the WhatsApp message. 

async def send_whatsapp_message(self, to: str, message: str):
        """Send WhatsApp message using Twilio"""
        try:
            from_whatsapp = f"whatsapp:{settings.TWILIO_PHONE_NUMBER}"
            to_whatsapp = to if to.startswith("whatsapp:") else f"whatsapp:{to}"
            logger.info(f"Sending message from {from_whatsapp} to {to_whatsapp}")
            message_chunks = self.chunk_message(message)
            for chunk in message_chunks:
                self.twilio_client.messages.create(
                    body=chunk,
                    from_=from_whatsapp,
                    to=to_whatsapp
                )
                if len(message_chunks) > 1:
                    time.sleep(1)
            logger.info("Message sent successfully")
        except Exception as e:
            logger.error(f"Error sending WhatsApp message: {e}")
            raise HTTPException(status_code=500, detail=f"Error sending message: {str(e)}")

This send_whatsapp_message() formats the “from” and “to” numbers with the whatsapp prefix. To avoid hitting Twilio’s message length limit, it calls chunk_message() to split long responses into 1500-char chunks. It then loops through each chunk and sends it via the Twilio API. The implementation pauses briefly between chunks so Twilio can process multiple messages. If an error occurs, the code logs it and returns a 500 error to the webhook. 

def chunk_message(self, message: str, max_length: int = 1500) -> list:
        if len(message) <= max_length:
            return [message]
        chunks = []
        current_chunk = ""
        sentences = message.replace("\n", ". ").split(". ")
        for sentence in sentences:
            if len(current_chunk) + len(sentence) + 2 <= max_length:
                current_chunk += sentence + ". "
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence + ". "
        if current_chunk:
            chunks.append(current_chunk.strip())
        if len(chunks) > 1:
            chunks = [f"({i+1}/{len(chunks)}) {chunk}" for i, chunk in enumerate(chunks)]
        return chunks

The chunk_message() helper to break a long string into numbered message pieces under Twilio’s limit. It splits on sentence boundaries and adds “(1/3)” style prefixes if more than one chunk is needed. 

async def check_hardware_issue(self, text: str) -> bool:
        """Check if the issue requires hardware maintenance"""
        try:
            completion = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",  # Using more cost-effective model for classification
                messages=[
                    {
                        "role": "system",
                        "content": """You are a technical support assistant for industrial/technical equipment.
                        Only respond with 'true' if the issue requires hardware maintenance or physical intervention
                        on technical equipment. Respond with 'false' for:
                        - Non-technical queries
                        - Personal issues
                        - Medical issues
                        - Questions about pets or animals
                        - General inquiries
                        Only respond with 'true' or 'false'."""
                    },
                    {
                        "role": "user",
                        "content": text
                    }
                ],
                max_tokens=10,
                temperature=0  # For consistent classification
            )
            response = completion.choices[0].message.content.lower().strip()
            return response == "true"
        except Exception as e:
            logger.error(f"Error checking hardware issue: {str(e)}")
            return False

The check_hardware_issue() uses GPT-4o as a classifier. This sends a very strict system prompt asking only for “true” or “false” answers. GPT-4o analyzes the user’s text and tells you if it’s a hardware/maintenance issue. This lets the application automatically detect when to escalate to maintenance. Next, you extend your pipeline beyond plain text to ingest and analyze user‑submitted images as well. This enables your help desk to process screenshots of errors or photos of physical issues for richer, multimodal troubleshooting. 

async def process_image(self, image_url: str, sender: str) -> str:
        try:
            response = requests.get(
                image_url,
                auth=(settings.TWILIO_ACCOUNT_SID, settings.TWILIO_AUTH_TOKEN),
                timeout=30
            )
            response.raise_for_status()
            image_data = base64.b64encode(response.content).decode('utf-8')
            data_url = f"data:image/jpeg;base64,{image_data}"
            # Get conversation history for context
            history = self.conversation_memory.get_conversation_history(sender)
            # Build messages with conversation context
            messages = [
                {
                    "role": "system",
                    "content": "You are a technical support assistant. Analyze images and provide solutions based on conversation history when relevant."
                }
            ]
            # Add relevant conversation history
            for msg in history[-3:]:  # Last 3 messages for context
                if msg['role'] == 'user':
                    content = msg['content']
                    if msg['media_url']:
                        content += " [Previous image sent]"
                    messages.append({"role": "user", "content": content})
                elif msg['role'] == 'assistant':
                    messages.append({"role": "assistant", "content": msg['content']})
            # Add current image analysis request
            messages.append({
                "role": "user",
                "content": [
                    {"type": "text", "text": "Describe this technical issue or error message, considering our previous conversation:"},
                    {"type": "image_url", "image_url": {"url": data_url}}
                ]
            })
            vision_completion = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",  # Updated model name
                messages=messages,
                max_tokens=300
            )
            image_description = vision_completion.choices[0].message.content
            # Store user's image message in conversation history
            self.conversation_memory.add_message(sender, "user", f"[Image sent] {image_description}", image_url)
            # Get relevant context from knowledge base
            context = self.knowledge_base.get_relevant_context(image_description)
            # Build contextual response using conversation history
            context_messages = self._build_conversation_context(sender, f"Analyze this image and provide a solution: {image_description}")
            # Add knowledge base context to the system message
            context_messages[0]["content"] += f"\n\nRelevant technical documentation:\n{context}"
            completion = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=context_messages,
                max_tokens=500
            )
            response_text = f"📷 Analysis complete!\n\nIssue: {image_description}\n\n{completion.choices[0].message.content}"
            # Store assistant's response in conversation history
            self.conversation_memory.add_message(sender, "assistant", response_text)
            return response_text
        except Exception as e:
            logger.error(f"Error processing image: {str(e)}")
            return "Sorry, I couldn't process the image. Please try again or describe the issue in text."

In process_image(), when a user sends an image, your code downloads it via Twilio’s authenticated URL, encodes it in base64, and submits it to the vision-capable GPT-4 endpoint gpt-4o-mini, with a prompt asking for a description. Once you receive the text description of the technical issue, you query Qdrant for related context and issue a second GPT-4o request to generate a solution. The final response then combines the described issue with GPT‑4o’s suggested fix.

Text Processing

This section implements the core logic that powers the technical support assistant. This handler is responsible for processing incoming WhatsApp messages, determining whether they are valid technical support queries, and generating appropriate responses. It combines language model capabilities, context retrieval from a vector database, and email notifications for hardware issues and customer feedback. See the code below. 

async def process_text(self, text: str, sender: str) -> str:
        try:
            # Handle special commands
            if text.lower().strip() in ['/clear', '/reset', 'clear history', 'reset conversation']:
                self.conversation_memory.clear_user_history(sender)
                return "✅ Conversation history cleared. How can I help you with your technical issue?"
            # First check if it's a technical query
            technical_check = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {
                        "role": "system",
                        "content": "Determine if this is a technical support query related to equipment or systems. Respond only with 'true' or 'false'."
                    },
                    {
                        "role": "user",
                        "content": text
                    }
                ],
                max_tokens=10,
                temperature=0
            )
            is_technical = technical_check.choices[0].message.content.lower().strip() == "true"
            if not is_technical:
                response = "I am a technical support assistant. I can only help with technical issues related to equipment and systems. For other types of questions, please consult the appropriate specialist."
                # Don't store non-technical queries in conversation history
                return response
            # Store user message in conversation history
            self.conversation_memory.add_message(sender, "user", text)
            # Get relevant context from knowledge base
            context = self.knowledge_base.get_relevant_context(text)
            # Build conversation context with history
            messages = self._build_conversation_context(sender, text)
            # Add knowledge base context to the system message
            messages[0]["content"] += f"\n\nRelevant technical documentation:\n{context}"
            # Get AI response with conversation context
            completion = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                max_tokens=500
            )
            response = completion.choices[0].message.content
            # Store assistant response in conversation history
            self.conversation_memory.add_message(sender, "assistant", response)
            # Check for hardware issues and send notifications
            user_email = self.get_user_email(sender)
            is_hardware_issue = await self.check_hardware_issue(text)
            if is_hardware_issue and settings.NOTIFICATION_EMAIL and self.email_service:
                try:
                    # Include conversation context in maintenance notification
                    history = self.conversation_memory.get_conversation_history(sender)
                    conversation_context = "\n".join([
                        f"{msg['role'].title()}: {msg['content']}" 
                        for msg in history[-5:]  # Last 5 messages for context
                    ])
                    await self.email_service.send_maintenance_notification(
                        settings.NOTIFICATION_EMAIL,
                        text,
                        priority="High" if "urgent" in text.lower() else "Medium",
                        hardware_details={
                            "issue_description": text,
                            "user_contact": sender,
                            "user_email": user_email,
                            "conversation_context": conversation_context
                        }
                    )
                    response += "\n\nI've notified our maintenance team about this hardware issue. They will review the conversation context and address it as soon as possible."
                except Exception as e:
                    logger.error(f"Failed to send maintenance notification: {e}")
            # Send satisfaction survey if we have user's email and email service is available
            if user_email and self.email_service:
                try:
                    interaction_id = str(uuid.uuid4())
                    await self.email_service.send_satisfaction_survey(user_email, interaction_id)
                    logger.info(f"Sent satisfaction survey to {user_email} for interaction {interaction_id}")
                except Exception as e:
                    logger.error(f"Failed to send satisfaction survey: {e}")
            else:
                logger.info(f"No email found for WhatsApp number {sender}, skipping satisfaction survey")
            return response
        except Exception as e:
            logger.error(f"Error processing text: {str(e)}")
            return "Sorry, I couldn't process your message. Please try again or contact support directly."

process_text() : This is the core message handler. First, you ask GPT-4o if the text is a technical support query. If not, you reply with a default “I can only help with technical issues” message. This helps to conserve your LLM resource as well as tie it to domain specifics. If it is technical, you get relevant context from Qdrant, then send the context + user query to GPT-4o for a solution. You then check if it’s a hardware issue is_hardware_issue. If true, you call send_maintenance_notification(), see EmailService below, to email the maintenance team. You add a note in the WhatsApp response that maintenance has been alerted. If the sender has a registered email, you generate a random interaction_id and call send_satisfaction_survey() to email them a survey after the support is done. Finally, you return GPT-4o’s solution text back to WhatsApp.

This method ties together all parts: GPT-4o classification and answer, Qdrant context retrieval, and triggering emails via SendGrid when appropriate.

Ngrok Tunneling

Next, expose your local endpoints publicly with ngrok so Twilio can reach and interact with your application. 

def start_ngrok(self):
        """Start ngrok tunnel"""
        try:
            self.ngrok_tunnel = ngrok.connect(8000)
            tunnel_url = self.ngrok_tunnel.public_url
            logger.info(f"ngrok tunnel established at: {tunnel_url}")
            return tunnel_url
        except Exception as e:
            logger.error(f"Error starting ngrok: {str(e)}")
            raise
    def cleanup(self):
        """Cleanup ngrok resources"""
        try:
            if self.ngrok_tunnel:
                ngrok.disconnect(self.ngrok_tunnel.public_url)
            logger.info("Cleaned up ngrok tunnel")
        except Exception as e:
            logger.error(f"Error during cleanup: {str(e)}")

start_ngrok() Opens a ngrok tunnel on port 8000 where your FastAPI will run. Twilio webhooks must point to this public URL so they reach your local server. cleanup() Disconnects the ngrok tunnel on shutdown.

At the end of app.py, you set up the FastAPI endpoints and the main loop: 

assistant = DiagnosticAssistant()
@app.post("/webhook")
async def whatsapp_webhook(request: Request, background_tasks: BackgroundTasks):
    # Parse incoming Twilio webhook form data
    form_data = await request.form()
    sender = form_data.get("From", "")
    if sender.startswith("whatsapp:"):
        sender = sender[9:]
    media_url = form_data.get("MediaUrl0")
    message_body = form_data.get("Body", "")
    # Determine if the user sent an image or text
    if media_url:
        response = await assistant.process_image(media_url)
    elif message_body:
        response = await assistant.process_text(message_body, sender)
    else:
        response = "Please send text or an image for support."
    await assistant.send_whatsapp_message(sender, response)
    return {"status": "success", "message": "Message processed"}

/webhook This FastAPI POST endpoint is where Twilio sends incoming WhatsApp messages. We extract:

From: who sent it (e.g., "whatsapp:+1234567890").

MediaUrl0: if an image was attached.

Body: the text message.

Then you strip the whatsapp: prefix from the sender’s number, then check if there’s an image or text. Based on that, you call process_image() or process_text().

Finally, you reply by calling send_whatsapp_message() to send back to the same sender.

Twilio expects a quick HTTP response, so you return JSON after scheduling the response message. 

@app.get("/")
async def root():
    return {
        "message": "Diagnostic Assistant is running",
        "version": "2.0.0",
        "status": "online"
    }

A simple root endpoint to verify the server is live. 

def main():
    print("Starting Diagnostic Assistant...")
    public_url = assistant.start_ngrok()
    print(f"Webhook URL: {public_url}/webhook")
    print("Set this as your Twilio sandbox webhook and send a test message.")
    uvicorn.run(app, host="0.0.0.0", port=8000)
    assistant.cleanup()
if __name__ == "__main__":
    main()

In main() when you run python app.py, this starts ngrok, prints the public URL, and runs the FastAPI server. It also reminds you to set the Twilio webhook to <ngrok_url>/webhook.

On shutdown, it cleans up the ngrok tunnel gracefully.

At this point, every line of app.py has been explained. You’ve covered configuration, the knowledge base, message handling, and how emails are triggered for maintenance alerts and surveys. Now let's move to the email service.

How the Email Service Integrates with Your AI Workflow

This helper module handles sending HTML emails via SendGrid. It defines an EmailService class with methods for maintenance alerts and satisfaction surveys. 

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, To, Email, Content, ClickTracking, TrackingSettings
from datetime import datetime
import logging
from typing import List, Dict, Optional
import json
import html
logger = logging.getLogger(__name__)
class EmailService:
    def __init__(self, api_key: str, from_email: str, notification_email: str):
        if not api_key:
            raise ValueError("SendGrid API key is required")
        if not from_email:
            raise ValueError("From email is required")
        if not notification_email:
            raise ValueError("Notification email is required")
        self.client = SendGridAPIClient(api_key)
        self.from_email = from_email
        self.notification_email = notification_email

You create a SendGrid API client using the SENDGRID_API_KEY. from_email is your verified sender; notification_email is where you’ll send maintenance alerts.

Next, is the maintenance notification. 

async def send_maintenance_notification(
        self, 
        to_email: str, 
        issue_description: str,
        priority: str = "Medium",
        hardware_details: Dict = None
    ):
        try:
            subject = f"🔧 Maintenance Required: {priority} Priority Issue Detected"
            # Format user contact information
            user_contact = hardware_details.get('user_contact', 'Not provided') if hardware_details else 'Not provided'
            user_email = hardware_details.get('user_email', 'Not provided') if hardware_details else 'Not provided'
            # Create a more structured HTML content
            content = f"""
            <html>
            <body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
                <div style="background-color: #f8f9fa; border-left: 4px solid #dc3545; padding: 15px; margin-bottom: 20px;">
                    <h2 style="color: #dc3545; margin: 0;">⚠️ Urgent Maintenance Required</h2>
                </div>
                <div style="background-color: #fff; border: 1px solid #dee2e6; border-radius: 4px; padding: 20px; margin-bottom: 20px;">
                    <h3 style="color: #495057; margin-top: 0;">Issue Description</h3>
                    <p style="margin-bottom: 20px; background-color: #f8f9fa; padding: 10px; border-radius: 4px;">{html.escape(issue_description)}</p>
                    <h3 style="color: #495057;">Priority Level</h3>
                    <p style="font-weight: bold; color: {'#dc3545' if priority == 'High' else '#ffc107' if priority == 'Medium' else '#28a745'}; font-size: 1.2em;">
                        🚨 {priority.upper()} PRIORITY
                    </p>
                    <h3 style="color: #495057;">Contact Information</h3>
                    <div style="background-color: #e9ecef; padding: 15px; border-radius: 4px;">
                        <p style="margin: 5px 0;"><strong>📱 WhatsApp Number:</strong> {html.escape(user_contact)}</p>
                        <p style="margin: 5px 0;"><strong>📧 User Email:</strong> {html.escape(user_email)}</p>
                    </div>
                    <h3 style="color: #495057;">Additional Details</h3>
                    <pre style="background-color: #f8f9fa; padding: 15px; border-radius: 4px; overflow-x: auto; font-size: 0.9em; border: 1px solid #dee2e6;">
{html.escape(json.dumps(hardware_details, indent=2) if hardware_details else 'No additional details provided')}
                    </pre>
                </div>
                <div style="background-color: #f8f9fa; padding: 15px; border-radius: 4px; font-size: 0.9em; color: #6c757d; text-align: center;">
                    <p style="margin: 0;">
                        🕒 <strong>Time Reported:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}
                    </p>
                    <p style="margin: 5px 0 0 0; font-style: italic;">
                        This notification was generated automatically by the Diagnostic Assistant system.
                    </p>
                </div>
            </body>
            </html>
            """
            message = Mail(
                from_email=Email(self.from_email),
                to_emails=[To(to_email)],
                subject=subject,
                html_content=Content("text/html", content)
            )
            response = self.client.send(message)
            logger.info(f"Maintenance notification sent successfully. Status code: {response.status_code}")
            return response.status_code
        except Exception as e:
            logger.error(f"Error sending maintenance notification: {str(e)}")
            raise

The send_maintenance_notification() sends a styled HTML email to alert maintenance staff.

You then use SendGrid’s Mail helper to construct the email. The subject indicates the issue and priority. The HTML includes the issue description (escaped for safety), priority (colored red for High), contact info (WhatsApp number and email of the user), and any extra hardware details in a <pre> block. You include a timestamp. Then you send the email via self.client.send(message), logging the response status. This follows SendGrid’s Python examples, which require a JSON body. You’re using the helper library to build it for us. Proceed to send the survey. 

async def send_satisfaction_survey(
        self, 
        to_email: str, 
        interaction_id: str, 
        whatsapp_number: Optional[str] = None
    ):
        try:
            subject = "📝 How was your experience? We'd love your feedback!"
            # Create survey URL parameters for better tracking
            survey_base_params = f"subject=Service%20Rating%20-%20{interaction_id}&body="
            # Create an HTML template with mailto links for rating
            survey_content = f"""
            <html>
            <body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
                <div style="background-color: #f8f9fa; border-left: 4px solid #007bff; padding: 15px; margin-bottom: 20px;">
                    <h2 style="color: #007bff; margin: 0;">🌟 Your Feedback Matters!</h2>
                </div>
                <div style="background-color: #fff; border: 1px solid #dee2e6; border-radius: 8px; padding: 25px; margin-bottom: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
                    <p style="font-size: 1.1em; margin-bottom: 20px;">We recently helped you with a technical issue and would love to know about your experience.</p>
                    <div style="background-color: #e3f2fd; padding: 15px; border-radius: 6px; margin-bottom: 25px; border-left: 4px solid #2196f3;">
                        <p style="margin: 0; font-weight: bold; color: #1976d2;">Reference Information:</p>
                        <ul style="list-style: none; padding: 10px 0 0 0; margin: 0;">
                            <li style="margin: 5px 0;">🔍 <strong>Interaction ID:</strong> {interaction_id}</li>
                            {f'<li style="margin: 5px 0;">📱 <strong>WhatsApp Number:</strong> {whatsapp_number}</li>' if whatsapp_number else ''}
                        </ul>
                    </div>
                    <p style="font-size: 1.1em; margin-bottom: 20px; text-align: center;"><strong>Please rate our service by clicking one of the options below:</strong></p>
                    <div style="display: flex; flex-direction: column; gap: 12px; max-width: 400px; margin: 0 auto;">
                        <a href="mailto:{self.notification_email}?{survey_base_params}Rating:%205%20stars%0AInteraction%20ID:%20{interaction_id}%0AComments:%20Please%20add%20any%20additional%20feedback%20here" 
                           style="background: linear-gradient(135deg, #28a745, #34ce57); color: white; padding: 15px 20px; text-decoration: none; border-radius: 8px; text-align: center; font-weight: bold; transition: all 0.3s ease; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
                           ⭐⭐⭐⭐⭐ Excellent - Exceeded Expectations
                        </a>
                        <a href="mailto:{self.notification_email}?{survey_base_params}Rating:%204%20stars%0AInteraction%20ID:%20{interaction_id}%0AComments:%20Please%20add%20any%20additional%20feedback%20here" 
                           style="background: linear-gradient(135deg, #17a2b8, #20c997); color: white; padding: 15px 20px; text-decoration: none; border-radius: 8px; text-align: center; font-weight: bold; transition: all 0.3s ease; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
                           ⭐⭐⭐⭐ Very Good - Met Expectations
                        </a>
                        <a href="mailto:{self.notification_email}?{survey_base_params}Rating:%203%20stars%0AInteraction%20ID:%20{interaction_id}%0AComments:%20Please%20add%20any%20additional%20feedback%20here" 
                           style="background: linear-gradient(135deg, #ffc107, #ffdb4d); color: #212529; padding: 15px 20px; text-decoration: none; border-radius: 8px; text-align: center; font-weight: bold; transition: all 0.3s ease; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
                           ⭐⭐⭐ Good - Satisfactory Service
                        </a>
                        <a href="mailto:{self.notification_email}?{survey_base_params}Rating:%202%20stars%0AInteraction%20ID:%20{interaction_id}%0AComments:%20Please%20add%20any%20additional%20feedback%20here" 
                           style="background: linear-gradient(135deg, #fd7e14, #ff922b); color: white; padding: 15px 20px; text-decoration: none; border-radius: 8px; text-align: center; font-weight: bold; transition: all 0.3s ease; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
                           ⭐⭐ Fair - Room for Improvement
                        </a>
                        <a href="mailto:{self.notification_email}?{survey_base_params}Rating:%201%20star%0AInteraction%20ID:%20{interaction_id}%0AComments:%20Please%20add%20any%20additional%20feedback%20here" 
                           style="background: linear-gradient(135deg, #dc3545, #e85d75); color: white; padding: 15px 20px; text-decoration: none; border-radius: 8px; text-align: center; font-weight: bold; transition: all 0.3s ease; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
                           ⭐ Poor - Needs Significant Improvement
                        </a>
                    </div>
                    <div style="margin-top: 25px; padding: 15px; background-color: #f8f9fa; border-radius: 6px; text-align: center;">
                        <p style="margin: 0; font-size: 0.95em; color: #6c757d;">
                            💡 <strong>Tip:</strong> Click any rating above to open your email client with a pre-filled message. 
                            Feel free to add additional comments in the email body!
                        </p>
                    </div>
                </div>
                <div style="background-color: #f8f9fa; padding: 20px; border-radius: 8px; font-size: 0.9em; color: #6c757d; text-align: center;">
                    <p style="margin: 0 0 10px 0; font-weight: bold;">Thank you for helping us improve our service! 🙏</p>
                    <p style="margin: 0; font-style: italic;">
                        This survey was sent automatically. Your feedback helps us provide better technical support.
                    </p>
                </div>
            </body>
            </html>
            """
            message = Mail(
                from_email=Email(self.from_email),
                to_emails=[To(to_email)],
                subject=subject,
                html_content=Content("text/html", survey_content)
            )
            # Disable click tracking as we're using mailto links
            tracking_settings = TrackingSettings()
            tracking_settings.click_tracking = ClickTracking(enable=False, enable_text=False)
            message.tracking_settings = tracking_settings
            response = self.client.send(message)
            logger.info(f"Satisfaction survey sent successfully to {to_email}. Status code: {response.status_code}")
            return response.status_code
        except Exception as e:
            logger.error(f"Error sending satisfaction survey: {str(e)}")
            raise

In this final section, the send_satisfaction_survey() builds an email asking the user to rate the service. It includes clickable buttons that are actually mailto: links to the notification email with a pre-filled subject and body containing the rating and interaction ID. For example, clicking “⭐⭐⭐⭐ Very Good” opens an email to support-team@company.com with subject “Service Rating - {interaction_id}” and body “Rating: 4 stars, Interaction ID and Comments”. You turn off SendGrid’s click tracking to allow these mailto links to work properly. Finally, you send the email and log the status.

With all these implementations in place, time to test the application.

Run the Local Server

To start the application, simply run: 

python app.py

This will:

  1. Launch an ngrok tunnel and print the public URL.
  2. Start the FastAPI server on port 8000.

You should see output like: 

Starting Diagnostic Assistant...
ngrok tunnel established at: https://abcdef.ngrok.io
Webhook URL: https://abcdef.ngrok.io/webhook

Configure Twilio WhatsApp Sandbox

Log into your Twilio Console, navigate to Messaging → Try it Out → Send a WhatsApp Message, and follow the steps to activate your sandbox. In particular:

  • Join the Sandbox: On your phone’s WhatsApp, send the message join <your-keyword> to the sandbox number +1 234 567 8900. Twilio will reply, confirming you’ve joined.
  • Configure Webhook URL: You will need to tell Twilio where to send incoming message webhooks. You’ll do this after running your application, since it will produce a ngrok link. For more illustrations, see the screenshot below.
WhatsApp Sandbox

The Twilio WhatsApp Sandbox is a pre-configured environment that lets you prototype immediately without having a fully approved Twilio number. Once you’ve joined the sandbox, any messages you send to the Twilio number will be forwarded to your webhook, and you can reply programmatically. After you are connected to the sandbox, your From number from Whatsapp should be displayed under Send a business-Initiated message. This number will need to be the same as the Twilio number in your .env file.

With the server running, you can now test the bot:

  • Send a message from your WhatsApp (that joined the sandbox) to your Twilio sandbox number.
  • The bot should reply after a moment. For example, try asking a simple technical question or even send an image of an error.
  • If you say something clearly non-technical, it should politely decline.
  • If you describe a hardware issue (e.g. “My printer is making a grinding noise”), the application will solve it using GPT-4o and also trigger an email alert.

Your maintenance team will receive a SendGrid email if it’s a hardware issue, thanks to the send_maintenance_notification() call. After the conversation, if your number was in USER_EMAILS, you’ll get a satisfaction survey email. See the screenshots below on how the application works on sending different issues.

Send an image to see what happens.

Twilio message analyzing printer connection error with code 0x0000011b

Now a text to see how it responds.

A chat message discussing PC screen issues, followed by three troubleshooting steps from Twilio.

Below is a continuation. It proceeds to bring a human in the loop when needed, by sending a notification email to the administrator. See the screenshot below.

A message detailing steps to troubleshoot operating system problems on a computer in Safe Mode.

It just doesn't stop there; see the screenshot containing the email triggered and sent to the technical support team.

Email with subject Maintenance Required about a pc blinking blue, marked as medium priority.

Finally, an email is sent for a survey to check the users' satisfaction. See the screenshot below.

Email asking for user feedback on technical support with different rating options

Troubleshooting Steps

Verify 'From' Address:

  • Double-check that the From parameter is set to whatsapp:+141XXXXXXXX.
  • This is the default number for the Twilio WhatsApp Sandbox.

Activate and Join the Sandbox:

  • Navigate to the Twilio WhatsApp Sandbox page.
  • Follow the instructions to send the join code from your WhatsApp number to the sandbox number.

Use Live Credentials:

  • Ensure that your API requests use your live Twilio Account SID and Auth Token.
  • Avoid using test credentials, as they don't support WhatsApp messaging.

Check Twilio Console for Approved Numbers:

  • Visit the Twilio WhatsApp Senders page to confirm which numbers are approved for WhatsApp messaging.
  • Only approved numbers can be used as the From address in your API requests.

Check your Email configuration

  • Ensure the FROM_EMAIL is SendGrid verified to be able to send emails.
  • If you do not receive emails in your inbox, check spam or promotions for notification emails.

That’s how to build a Smart IT Help Desk with AI, Twilio WhatsApp, Qdrant, and SendGrid

This tutorial gave you a complete, end-to-end example of how to stand up a locally hosted Smart IT Help Desk. It receives WhatsApp messages via Twilio’s Sandbox for WhatsApp, semantically enriches them with Qdrant vector search; reasons over context with GPT‑4o, and automatically escalates hardware issues and sends satisfaction surveys via SendGrid. You saw how to configure FastAPI, expose your webhook with ngrok, index technical docs into Qdrant, and wire together every component in app.py and email_service.pywith every line of code explained.

For further enhancements, consider adding:

  • Robust Data Architecture:

Migrate vector embeddings from local Qdrant to Qdrant Cloud for auto‑scaling, high availability, zero‑downtime upgrades, monitoring, and backups

Introduce a relational database (PostgreSQL/MySQL) with tables for users, tickets, messages, and survey_responses—enabling ACID guarantees, structured queries, and storage of user feedback for analytics and model retraining

  • Feedback‑Driven Learning Loop with a Self‑Improving Application:

Capture survey_ratings (1–5 stars) and comments in your SQL survey_responses table; periodically extract low‑rating cases to refine prompts, update Qdrant context, or fine‑tune models—so the system learns from its mistakes.

By systematically logging survey responses, analyzing low‑rating interactions, and updating both your semantic index and LLM prompts, you create a closed‑loop process that helps the help desk get smarter over time. Making it an adaptive system that learns from its mistakes and continuously refines its own performance.

These additions can further streamline your customer care, providing deeper insights and improving customer service efficiency.

Jacob Muganda builds AI applications and demystifies complex topics through concise tutorials on Python and AI integrations. He’s an experienced AI engineer adept at crafting and deploying cutting‑edge algorithms to solve real‑world challenges. He won the EFH Innovation Sprint Hackathon for contextualizing LLM reasoning to scientific papers and fintech.

Related Posts

Related Resources

A newspaper article

Twilio Docs

From APIs to SDKs to sample apps

API reference documentation, SDKs, helper libraries, quickstarts, and tutorials for your language and platform.
An Open book

Resource Center

The latest ebooks, industry reports, and webinars

Learn from customer engagement experts to improve your own communication.
User group reactions

Ahoy

Twilio's developer community hub

Best practices, code samples, and inspiration to build communications and digital engagement experiences.