from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from app.utils.database import DatabaseManager from twilio.rest import Client import os import logging import httpx from bson import ObjectId from datetime import datetime, timezone from pydantic import BaseModel from typing import Optional, Dict, Any, List from zoneinfo import ZoneInfo from services.twilio_service import send_sms # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) app = FastAPI(title="Phone Number Service") # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize database and Twilio client db = DatabaseManager() phone_numbers_collection = None ai_assistants_collection = None businesses_collection = None twilio_client = Client( os.getenv("TWILIO_ACCOUNT_SID"), os.getenv("TWILIO_AUTH_TOKEN") ) @app.on_event("startup") async def startup_event(): global phone_numbers_collection, ai_assistants_collection, businesses_collection phone_numbers_collection = db.get_collection("phone_numbers") ai_assistants_collection = db.get_collection("ai_assistants") businesses_collection = db.get_collection("businesses") @app.on_event("shutdown") async def shutdown_event(): db.close() @app.get("/available-numbers") async def get_available_numbers(country="GB", limit=20): numbers = twilio_client.available_phone_numbers(country).local.list(limit=limit) return {"numbers": [{"phone_number": num.phone_number, "friendly_name": num.friendly_name} for num in numbers]} @app.post("/import-number") async def import_number(request: dict): try: phone_number = request["phone_number"] business_id = request["business_id"] agent_id = request["agent_id"] address_sid = request["address_sid"] sip_trunk_sid = request["sip_trunk_sid"] # Need this from the client #DO NOT REMOVE THIS bundle_sid = request["bundle_sid"] # Get regulatory bundle for the country country_code = phone_number[:3] # e.g., "+44" for UK #bundles = twilio_client.numbers.regulatory_compliance.bundles.list( iso_country=country_code[1:] # Remove the '+' for ISO format #) # Get SIP Trunk details sip_trunk = twilio_client.trunking.v1.trunks(sip_trunk_sid).fetch() # Purchase number from Twilio number = twilio_client.incoming_phone_numbers.create( phone_number=phone_number, address_sid=address_sid, bundle_sid=bundle_sid #DONT CHANGE THIS ) # Associate number with SIP Trunk # Add phone number to SIP Trunk twilio_client.trunking.v1.trunks(sip_trunk_sid).phone_numbers.create( phone_number_sid=number.sid ) # Create origination URL for SIP routing twilio_client.trunking.v1.trunks(sip_trunk_sid).origination_urls.create( sip_url=f"sip:{iso_country}@{sip_trunk.domain_name}", weight=10, priority=10, enabled=True, friendly_name=f"Number {number.phone_number}" ) # Fetch SIP Auth credentials from Twilio agent_doc = ai_assistants_collection.find_one({"_id": ObjectId(agent_id)}) if not agent_doc: raise HTTPException(status_code=404, detail="Agent not found") # Get business details for nickname business_doc = businesses_collection.find_one({"_id": ObjectId(business_id)}) business_name = business_doc.get("name", "Unknown Business") if business_doc else "Unknown Business" # Configure number on Retell async with httpx.AsyncClient() as client: retell_response = client.post( f"{os.getenv('RETELLAI_API_URL')}/import-phone-number", headers={ "Authorization": f"Bearer {os.getenv('RETELL_API_KEY')}", "Content-Type": "application/json" }, json={ "phone_number": number.phone_number, "termination_uri": f"{sip_trunk.domain_name}", "sip_trunk_auth_username": os.getenv("TWILIO_SIP_USERNAME"), # Optional auth params "sip_trunk_auth_password": os.getenv("TWILIO_SIP_PASSWORD"), # Optional auth params "inbound_agent_id": agent_doc["retell_agent_id"], "outbound_agent_id": agent_doc["retell_agent_id"], "nickname": f"{business_name} - {number.phone_number}" } ) retell_response = await retell_response if retell_response.status_code not in [200, 201]: raise HTTPException( status_code=retell_response.status_code, detail=f"Retell API error: {retell_response.text}" ) # Save to database phone_record = { "phone_number": number.phone_number, "twilio_sid": number.sid, "business_id": business_id, "agent_id": agent_id, "retell_agent_id": agent_doc["retell_agent_id"], "sip_trunk_sid": sip_trunk_sid, "status": "active" } result = phone_numbers_collection.insert_one(phone_record) return { "message": "Phone number imported successfully", "record_id": str(result.inserted_id) } except Exception as e: logger.error(f"Error in import_number: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/outbound-call") async def make_outbound_call(request: dict): try: from_number = request["from_number"] to_number = request["to_number"] # Optional metadata for the call metadata = request.get("metadata", {}) # Get customer name from request or use default customer_phone = request.get("customer_phone", "Customer") # Set up dynamic variables for Retell LLM # Get the outbound agent ID from environment variables outbound_agent_id = os.getenv("OUTBOUND_AGENT_RETELL_ID") if not outbound_agent_id: raise HTTPException( status_code=500, detail="Outbound agent ID not configured" ) # Make the outbound call using Retell API async with httpx.AsyncClient() as client: retell_response = await client.post( f"{os.getenv('RETELLAI_API_URL')}/v2/create-phone-call", headers={ "Authorization": f"Bearer {os.getenv('RETELL_API_KEY')}", "Content-Type": "application/json" }, json={ "retell_llm_dynamic_variables": { "customer_phone": customer_phone }, "from_number": from_number, "to_number": to_number, "override_agent_id": outbound_agent_id, "metadata": metadata } ) if retell_response.status_code not in [200, 201]: raise HTTPException( status_code=retell_response.status_code, detail=f"Retell API error: {retell_response.text}" ) call_data = retell_response.json() # Save call record to database call_record = { "call_id": call_data["call_id"], "from_number": from_number, "to_number": to_number, "agent_id": outbound_agent_id, "status": call_data["call_status"], "direction": "outbound", "metadata": metadata, "created_at": datetime.utcnow() } result =phone_numbers_collection.insert_one(call_record) return { "message": "Outbound call initiated successfully", "call_id": call_data["call_id"], "record_id": str(result.inserted_id) } except KeyError as e: raise HTTPException( status_code=400, detail=f"Missing required field: {str(e)}" ) except Exception as e: logger.error(f"Error in make_outbound_call: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) class RetellWebhookRequest(BaseModel): event_type: str call_id: str payload: Dict[str, Any] from_number: Optional[str] = None to_number: Optional[str] = None duration: Optional[int] = None call_status: Optional[str] = None speaker: Optional[str] = None text: Optional[str] = None @app.post("/webhook/retell") async def retell_webhook(request: RetellWebhookRequest): try: # Log the entire webhook payload logger.info("Received Retell webhook event:") logger.info(f"Event Type: {request.event_type}") logger.info(f"Call ID: {request.call_id}") logger.info("Full Payload:") logger.info(request.dict()) # Handle different event types if request.event_type == 'call.started': logger.info(f"Call started - From: {request.from_number} To: {request.to_number}") elif request.event_type == 'call.ended': logger.info(f"Call ended - Duration: {request.duration} seconds") logger.info(f"Call Status: {request.call_status}") elif request.event_type == 'transcription': logger.info(f"Transcription - Speaker: {request.speaker}") logger.info(f"Text: {request.text}") # Store the event in database for future reference event_record = { "event_type": request.event_type, "call_id": request.call_id, "timestamp": datetime.utcnow(), "payload": request.dict() } await phone_numbers_collection.insert_one(event_record) return {"status": "success"} except Exception as e: logger.error(f"Error processing webhook: {str(e)}") logger.error(f"Request payload: {request.dict()}") raise HTTPException(status_code=500, detail=str(e)) class CalWebhookRequest(BaseModel): type: str title: str organizer: Dict[str, Any] attendees: List[Dict[str, Any]] payload: Dict[str, Any] triggerEvent: str @app.post("/webhook/cal") #CAL.com async def cal_webhook(request: Request): try: # Parse the JSON body data = await request.json() logger.info("Received Cal.com webhook event:") logger.info(f"Event Type: {data.get('triggerEvent')}") logger.info("Full Payload:") logger.info(data) if data['triggerEvent'] == "BOOKING_CREATED": payload = data['payload'] # Get attendee (customer) details attendee = payload['attendees'][0] customer_name = attendee['name'] customer_phone = attendee.get('phone') # You might need to add phone field in Cal.com form # Parse and format the appointment time start_time_str = payload['startTime'] start_time = datetime.fromisoformat(start_time_str.replace('Z', '+00:00')) # Convert to customer's timezone customer_tz = ZoneInfo(attendee['timeZone']) local_time = start_time.astimezone(customer_tz) formatted_time = local_time.strftime("%A, %B %d at %I:%M %p") # Compose the SMS message message = ( f"Hi {customer_name}, your test drive appointment has been confirmed for {formatted_time}. " f"Jose will be your advisor when you visit the dealership. " f"The appointment will take place at {payload.get('location', 'our dealership')}. " f"If you need to reschedule, please contact us. We look forward to seeing you!" ) # Send SMS using Twilio service try: await send_sms( to_number=customer_phone, message=message, from_number=os.getenv('TWILIO_FROM_NUMBER') ) logger.info(f"Confirmation SMS sent to {customer_phone}") except Exception as e: logger.error(f"Failed to send SMS: {str(e)}") # Store the event in database event_record = { "event_type": data['triggerEvent'], "timestamp": datetime.utcnow(), "payload": data } phone_numbers_collection.insert_one(event_record) return {"status": "success"} except Exception as e: logger.error(f"Error processing Cal.com webhook: {str(e)}") logger.error(f"Request payload: {data if 'data' in locals() else 'No data'}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/webhook/post-booking") async def post_booking_webhook(request: Request): # Log the complete request for debugging logger.info("=-=-=-=-=-=-=-=Received post-booking webhook request:") logger.info(f"Headers: {dict(request.headers)}") body = await request.body() logger.info(f"Raw body: {body.decode()}") try: # Parse the JSON body data = await request.json() # Extract post booking operation details post_booking_op = data.get('name') if post_booking_op == 'post_booking_op': args = data.get('args', {}) appointment_ts = args.get('appointment_booked_ts') customer_name = args.get('customer_name') customer_email = args.get('customer_email') customer_phone = data.get('call', {}).get('to_number') direction=data.get('call', {}).get("direction") question_1 = args.get('question_1') answer_1 = args.get('answer_1') question_2 = args.get('question_2') answer_2 = args.get('answer_2') question_3 = args.get('question_3') answer_3 = args.get('answer_3') twilio_call_sid=data.get('call', {}).get("twilio_call_sid") logger.info(f"Received post-booking webhook for customer {customer_name}") logger.info(f"Appointment timestamp: {appointment_ts}") logger.info(f"Customer Name: {customer_name}") logger.info(f"Customer Email: {customer_email}") logger.info(f"Customer phone: {customer_phone}") logger.info(f"Question 1: {question_1}") logger.info(f"Answer 1: {answer_1}") logger.info(f"Question 2: {question_2}") logger.info(f"Answer 2: {answer_2}") logger.info(f"Question 3: {question_3}") logger.info(f"Answer 3: {answer_3}") logger.info(f"Twilio Call SID: {twilio_call_sid}") # Store the event in database event_record = { "event_type": "post_booking", "timestamp": datetime.utcnow(), "payload": data } phone_numbers_collection.insert_one(event_record) return {"status": "success"} except Exception as e: logger.error(f"Error processing post-booking webhook: {str(e)}") logger.error(f"Request payload: {data if 'data' in locals() else 'No data'}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/webhook/retell-events") async def retell_events_webhook(request: Request): logger.info("=-=-=-=-=-=-=-=Received retell-events webhook request=-=-=-=-=-=-=-=") try: # Parse the JSON body data = await request.json() # Log the entire webhook payload # logger.info("Received Retell event webhook:") # logger.info("Full Payload:") # logger.info(data) # Store the event in database event_record = { "event_type": "retell_event", "timestamp": datetime.utcnow(), "payload": data } await phone_numbers_collection.insert_one(event_record) return {"status": "success"} except Exception as e: logger.error(f"Error processing Retell event webhook: {str(e)}") logger.error(f"Request payload: {data if 'data' in locals() else 'No data'}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/webhook/directory") async def retell_dynamic_variables(request: Request): try: # Parse the JSON body data = await request.json() logger.info("Received Retell dynamic variables request:") logger.info(data) # Get call information from the request call_id = data.get("call_id") agent_id = data.get("agent_id") caller_number = data.get("from_number") # Example: Generate dynamic transfer variables # In a real implementation, you would fetch this from your database transfer_agents = [ { "name": "John ", "department": "Sales", "phone_number": "+923204046969", "availability": "available" }, { "name": "Sarah Johnson", "department": "Service", "phone_number": "+14155552345", "availability": "available" }, { "name": "Michael Brown", "department": "Finance", "phone_number": "+14155553456", "availability": "busy" } ] # Filter only available agents available_agents = [agent for agent in transfer_agents if agent["availability"] == "available"] # Create a single transfer instruction string transfer_instructions = "" for agent in available_agents: transfer_instructions += f"If the user wants to reach {agent['name']} or {agent['department']} department, transfer to {agent['phone_number']}; " # Remove trailing semicolon and space if present if transfer_instructions.endswith("; "): transfer_instructions = transfer_instructions[:-2] # Format the response with the single transfer instruction response = { "transfer_instructions": transfer_instructions } # Log the response #logger.info("Sending dynamic variables response:") #logger.info(response) # Store the original data in database (not the simplified response) event_record = { "event_type": "dynamic_variables_request", "call_id": call_id, "agent_id": agent_id, "timestamp": datetime.utcnow(), "request": data, "available_agents": available_agents # Store the full agent data } phone_numbers_collection.insert_one(event_record) return response except Exception as e: logger.error(f"Error processing dynamic variables request: {str(e)}") logger.error(f"Request payload: {data if 'data' in locals() else 'No data'}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/webhook/missed-call-reminder") async def call_reminder(request: Request): try: # Parse the JSON body logger.info("=-=-=-=-=-=-=-=Received call reminder request=-=-=-=-=-=-=-=") data = await request.json() logger.info("Received call reminder request:") logger.info(data) # Extract information from the request args = data.get("args", {}) customer_name = args.get("customer_name", "A customer") # Get customer phone based on call direction call_data = data.get("call", {}) direction = call_data.get("direction") if direction == "inbound": customer_phone = call_data.get("from_number", "+19294193939") else: # outbound or any other case customer_phone = call_data.get("to_number", "+19294193939") transfer_target_name = args.get("transfer_to_name", "the agent") transfer_target_phone = args.get("transfer_to_number", "Unknown") call_id = data.get("call", {}).get("call_id", "Unknown") logger.info(f"Received call reminder request for customer {customer_name} with phone number {customer_phone}") # Notification recipient notification_phone = "+923334046969" # Fixed notification number # Compose the SMS message #MISSED CALL ALERT: A customer (+19294193939) called and was transferred to the agent (Unknown), but the call was not answered. #Please call the customer back as soon as possible. Call ID: Unknown message = ( f"MISSED CALL ALERT: A customer {customer_name} ({customer_phone}) called and was transferred to " f"{transfer_target_name} ({transfer_target_phone}), but the call was not answered. " f"Please call the customer back as soon as possible. Dealerpulse.cloud Call ID: {call_id}" ) # Send SMS using Twilio service try: await send_sms( to_number=notification_phone, message=message, from_number=os.getenv('TWILIO_FROM_NUMBER') ) logger.info(f"Reminder SMS sent to {notification_phone}") # Also send SMS to the transfer target if available if transfer_target_phone and transfer_target_phone.startswith("+"): await send_sms( to_number=transfer_target_phone, message=f"You missed a call from {customer_name} ({customer_phone}). Please call them back as soon as possible.", from_number=os.getenv('TWILIO_FROM_NUMBER') ) logger.info(f"Reminder SMS sent to transfer target {transfer_target_phone}") except Exception as e: logger.error(f"Failed to send SMS: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to send SMS: {str(e)}") # Store the event in database event_record = { "event_type": "call_reminder", "call_id": call_id, "customer_name": customer_name, "customer_phone": customer_phone, "transfer_target_name": transfer_target_name, "transfer_target_phone": transfer_target_phone, "notification_sent_to": notification_phone, "timestamp": datetime.utcnow(), "request": data } phone_numbers_collection.insert_one(event_record) return { "status": "success", "message": "Call reminder notification sent" } except Exception as e: logger.error(f"Error processing call reminder: {str(e)}") logger.error(f"Request payload: {data if 'data' in locals() else 'No data'}") raise HTTPException(status_code=500, detail=str(e))