""" Twilio Campaign Outreach — FastAPI Application (HF Spaces Deployment) Exposes endpoints to initiate calls, serve TwiML, capture DTMF responses, and poll status. """ from fastapi import FastAPI, HTTPException, Form, Query, Response from pydantic import BaseModel from typing import Optional from campaign_service import ( generate_call_id, store_campaign, initiate_call, update_response, get_campaign, # SMS functions generate_sms_id, store_sms, send_sms, get_sms, ) app = FastAPI(title="Campaign Outreach Service", version="1.0.0") # --- Pydantic Schemas --- class CampaignInitiateRequest(BaseModel): message: str class CampaignInitiateResponse(BaseModel): call_id: str status: str phone: str message_preview: str class SMSRequest(BaseModel): message: str class SMSResponse(BaseModel): sms_id: str status: str phone: str message_preview: str # --- Endpoints --- @app.get("/health") def health(): return {"status": "ok", "service": "Campaign Outreach"} @app.post("/campaign/initiate/phonecall/{number}", response_model=CampaignInitiateResponse) def initiate_phonecall(number: str, body: CampaignInitiateRequest): """ Initiate an outbound voice call to the given phone number. The personalized message in the request body will be read aloud to the lead. The lead responds with DTMF: 0 = Accept meeting, 1 = Reject meeting. """ if not body.message or not body.message.strip(): raise HTTPException(status_code=400, detail="Message cannot be empty.") call_id = generate_call_id() # Store the campaign record store_campaign(call_id=call_id, phone=number, message=body.message.strip()) # Trigger the Twilio outbound call try: initiate_call(phone_number=number, call_id=call_id) except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=f"Twilio call failed: {e}") return CampaignInitiateResponse( call_id=call_id, status="initiated", phone=number, message_preview=body.message.strip()[:120] + ("..." if len(body.message.strip()) > 120 else ""), ) @app.get("/campaign/twiml/{call_id}") def twiml_webhook(call_id: str): """ Twilio webhook: called when the lead picks up the phone. Returns TwiML that reads the personalized message and gathers a DTMF keypress. """ record = get_campaign(call_id) if record is None: raise HTTPException(status_code=404, detail="Campaign not found.") message_text = record["message"] # Build TwiML XML response twiml = f""" {message_text} Press 0 to accept a meeting. Press 1 to decline. We didn't receive your input. Goodbye. """ return Response(content=twiml, media_type="application/xml") @app.post("/campaign/response") def response_webhook(call_id: str = Query(...), Digits: str = Form(default="")): """ Twilio webhook: called when the lead presses a digit. Records the lead's response (0 = Accept, 1 = Reject). """ record = get_campaign(call_id) if record is None: raise HTTPException(status_code=404, detail="Campaign not found.") # Update the store update_response(call_id=call_id, digits=Digits) # Generate thank-you TwiML if Digits == "0": reply = "Thank you! Our team will reach out shortly to confirm your meeting. Goodbye!" elif Digits == "1": reply = "Understood. Thank you for your time. Have a great day. Goodbye!" else: reply = "Invalid input received. Goodbye." twiml = f""" {reply} """ return Response(content=twiml, media_type="application/xml") @app.post("/campaign/call-status/{call_id}") def call_status_webhook(call_id: str): """ Twilio status callback: receives call lifecycle events (ringing, in-progress, completed, etc). Updates the campaign record if the call failed or was not answered. """ record = get_campaign(call_id) if record and record["status"] == "initiated": # If we get a status callback but no DTMF was captured, mark as no-answer pass # Twilio sends multiple status updates; we only care about final state return {"status": "received"} @app.get("/campaign/status/{call_id}") def campaign_status(call_id: str): """ Agent-facing endpoint: poll the outcome of an initiated campaign call. """ record = get_campaign(call_id) if record is None: raise HTTPException(status_code=404, detail=f"Campaign with call_id '{call_id}' not found.") return { "call_id": record["call_id"], "phone": record["phone"], "status": record["status"], "lead_response": record["lead_response"], "message_used": record["message"], "initiated_at": record["initiated_at"], "responded_at": record["responded_at"], } # ================================================================ # SMS ENDPOINTS # ================================================================ @app.post("/campaign/initiate/sms/{number}", response_model=SMSResponse) def initiate_sms(number: str, body: SMSRequest): """ Send a personalized SMS to the given phone number. The appointment confirmation footer is automatically appended. """ if not body.message or not body.message.strip(): raise HTTPException(status_code=400, detail="Message cannot be empty.") sms_id = generate_sms_id() # Store the SMS campaign record store_sms(sms_id=sms_id, phone=number, message=body.message.strip()) # Send via Twilio try: send_sms(phone_number=number, message=body.message.strip(), sms_id=sms_id) except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=f"SMS delivery failed: {e}") return SMSResponse( sms_id=sms_id, status="sent", phone=number, message_preview=body.message.strip()[:120] + ("..." if len(body.message.strip()) > 120 else ""), ) @app.get("/campaign/sms/status/{sms_id}") def sms_status(sms_id: str): """ Agent-facing endpoint: poll the outcome of an initiated SMS campaign. """ record = get_sms(sms_id) if record is None: raise HTTPException(status_code=404, detail=f"SMS campaign '{sms_id}' not found.") return { "sms_id": record["sms_id"], "phone": record["phone"], "status": record["status"], "message_sent": record["message"], "sent_at": record["sent_at"], } if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=7860)