# ============================== # IMPORTS # ============================== from agents import function_tool, SQLiteSession from supabase import create_client from runtime_context import get_thread_id from routes.notification_routes import notify_escalation import os from dotenv import load_dotenv # ============================== # LOAD ENV # ============================== load_dotenv() SUPABASE_URL = os.getenv("SUPABASE_URL") SUPABASE_KEY = os.getenv("SUPABASE_KEY") supabase = create_client(SUPABASE_URL, SUPABASE_KEY) # ============================== # SESSION # ============================== def get_session(thread_id: str): db_path = os.path.join(os.path.dirname(__file__), "conversations.db") return SQLiteSession(thread_id, db_path) # ============================== # FORMAT CHAT HISTORY # ============================== def format_chat_history(items): formatted = [] for item in items: try: if not isinstance(item, dict): continue role = item.get("role") if role not in ["user", "assistant"]: continue content = item.get("content", "") text = "" if isinstance(content, list): for c in content: if isinstance(c, dict) and "text" in c: text += c["text"] + " " else: text = str(content) text = text.strip() if text: formatted.append(f"{role.upper()}: {text}") except Exception: continue return "\n".join(formatted) # ============================================================ # HELPER → GET CUSTOMER + CONVERSATION # ============================================================ def get_customer_and_conversation(phone: str): # get customer customer = supabase.table("customers").select("*").eq("phone", phone).limit(1).execute() if not customer.data: new_customer = supabase.table("customers").insert({"phone": phone}).execute() customer_id = new_customer.data[0]["id"] else: customer_id = customer.data[0]["id"] # get conversation conv = supabase.table("conversations") \ .select("*") \ .eq("thread_id", phone) \ .limit(1) \ .execute() if not conv.data: new_conv = supabase.table("conversations").insert({ "customer_id": customer_id, "thread_id": phone, "phone": phone, "status": "active" }).execute() conversation_id = new_conv.data[0]["id"] else: conversation_id = conv.data[0]["id"] return customer_id, conversation_id def safe_notify_escalation( conversation_id: str, customer_id: str, customer_name: str, summary: str, ): try: notify_escalation( conversation_id=conversation_id, customer_id=customer_id, customer_name=customer_name, summary=summary, ) except Exception as e: print("ESCALATION NOTIFICATION ERROR:", str(e)) # ============================================================ # ESCALATION TOOL # ============================================================ @function_tool async def escalate_to_admin_tool( user_name: str, email: str, summary: str, last_user_message: str, priority: str, topics: str, short_note: str ) -> str: """ Escalates conversation to human support. priority MUST be one of: low | medium | high | urgent topics should be a short topic/category for dashboard. Example: "pricing", "appointment", "delivery issue", "customer complaint" short_note should be a short dashboard note about this customer. Example: "Customer asked about pricing and wants human follow-up." """ try: print("\nESCALATION TRIGGERED") # ============================== # VALIDATE PRIORITY # ============================== allowed_priorities = ["low", "medium", "high", "urgent"] if priority not in allowed_priorities: priority = "medium" # ============================== # THREAD ID = PHONE # ============================== phone = get_thread_id() or "unknown" thread_id = phone print("Phone:", phone, "Priority:", priority) # ============================== # GET CUSTOMER + CONVERSATION IDS # ============================== customer_id, conversation_id = get_customer_and_conversation(phone) # ============================== # LOAD CHAT HISTORY # ============================== session = get_session(thread_id) messages = await session.get_items() messages = messages[-20:] chat_history = format_chat_history(messages) if last_user_message: chat_history += f"\nUSER: {last_user_message}" # ============================== # HUMAN FINAL MESSAGE # ============================== final_ai_message = ( "Got it, let me double check this for you.\n" "I’m passing this to my team and we’ll get back to you shortly." ) chat_history += f"\nASSISTANT: {final_ai_message}" # ============================== # SAFE FALLBACK VALUES # ============================== user_name = user_name or "Unknown" email = email or "Not provided" topics = topics or "General" short_note = short_note or summary or "Customer needs human follow-up." summary = summary or short_note # ============================== # UPDATE CUSTOMER INFO # ============================== supabase.table("customers").update({ "name": user_name, "email": email, "topics": topics, "short_note": short_note, "updated_at": "now()" }).eq("id", customer_id).execute() # ============================== # INSERT ESCALATION # ============================== supabase.table("escalations").insert({ "customer_id": customer_id, "conversation_id": conversation_id, "name": user_name, "email": email, "phone": phone, "summary": summary, "chat_history": chat_history, "status": "pending", "priority": priority, "updated_at": "now()" }).execute() # ============================== # UPDATE CONVERSATION STATUS # ============================== supabase.table("conversations").update({ "status": "escalated", "priority": priority, "is_escalated": True, "updated_at": "now()" }).eq("id", conversation_id).execute() # ============================== # SEND ADMIN NOTIFICATION # ============================== safe_notify_escalation( conversation_id=conversation_id, customer_id=customer_id, customer_name=user_name if user_name != "Unknown" else phone, summary=summary, ) print("Escalation saved with priority:", priority) return final_ai_message except Exception as e: print("ESCALATION ERROR:", str(e)) return "Let me check this and get back to you shortly."