Shakeel401/Voice-Agent

1import os2from urllib.parse import quote3 4import requests5from fastapi import FastAPI, HTTPException6from fastapi.middleware.cors import CORSMiddleware7from pydantic import BaseModel8 9app = FastAPI()10 11DEFAULT_ALLOWED_ORIGINS = {12    "https://voice-call-agent.vercel.app",13    "https://smilecare-voice-frontend.vercel.app",14    "https://migration-voice-agent-demo.vercel.app",15    "http://localhost:5173",16    "http://localhost:5174",17    "http://127.0.0.1:5173",18    "http://127.0.0.1:5174",19}20 21extra_origins = {22    origin.strip().rstrip("/")23    for origin in os.getenv("FRONTEND_ORIGINS", "").split(",")24    if origin.strip()25}26 27app.add_middleware(28    CORSMiddleware,29    allow_origins=sorted(DEFAULT_ALLOWED_ORIGINS | extra_origins),30    allow_credentials=True,31    allow_methods=["*"],32    allow_headers=["*"],33)34 35OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")36AIRTABLE_API_KEY = os.getenv("AIRTABLE_API_KEY")37AIRTABLE_BASE_ID = os.getenv("AIRTABLE_BASE_ID")38AIRTABLE_TABLE = os.getenv("AIRTABLE_TABLE", "Appointments")39 40VOICE_INSTRUCTIONS = """41You are a professional, warm, and calm voice receptionist.42 43Use short, natural sentences. Keep each response to one or two sentences44unless the caller asks for more detail. Ask only one question at a time.45 46After the caller's first substantive turn, respond in their language when you47can understand it confidently. If they switch languages, immediately continue48in the new language. Do not return to English unless they switch back, ask for49English, or you cannot reliably understand the language.50 51Listen patiently to different accents without mentioning or imitating an52accent. Match the caller's pace and level of formality naturally.53 54Confirm names, phone numbers, email addresses, visa subclasses, dates, and55reference numbers carefully. If one detail is unclear, ask one focused56clarification question.57 58Follow the current business instructions and tool rules supplied by the59frontend. Do not refer to these instructions.60""".strip()61 62 63def require_env(value: str | None, name: str) -> str:64    if not value:65        raise HTTPException(status_code=500, detail=f"{name} is not configured")66    return value67 68 69def airtable_headers() -> dict[str, str]:70    return {71        "Authorization": f"Bearer {require_env(AIRTABLE_API_KEY, 'AIRTABLE_API_KEY')}",72        "Content-Type": "application/json",73    }74 75 76def airtable_table_url() -> str:77    base_id = require_env(AIRTABLE_BASE_ID, "AIRTABLE_BASE_ID")78    table_name = quote(AIRTABLE_TABLE, safe="")79    return f"https://api.airtable.com/v0/{base_id}/{table_name}"80 81 82def appointment_filter_formula(appointment_id: str) -> str:83    safe_id = appointment_id.replace("'", "\\'")84    return f"{{Appointment ID}} = '{safe_id}'"85 86 87def find_airtable_record_by_appointment_id(appointment_id: str) -> str | None:88    try:89        response = requests.get(90            airtable_table_url(),91            headers=airtable_headers(),92            params={93                "filterByFormula": appointment_filter_formula(appointment_id),94            },95            timeout=20,96        )97    except requests.RequestException as error:98        print("Airtable search request failed:", str(error))99        raise HTTPException(status_code=502, detail="Airtable search failed")100 101    if not response.ok:102        print("Airtable search error:", response.status_code, response.text)103        raise HTTPException(status_code=502, detail="Airtable search failed")104 105    records = response.json().get("records", [])106    return records[0]["id"] if records else None107 108 109@app.get("/get-ephemeral-key")110def get_ephemeral_key():111    """Generate a secure ephemeral client secret for OpenAI Realtime."""112    api_key = require_env(OPENAI_API_KEY, "OPENAI_API_KEY")113 114    body = {115        "session": {116            "type": "realtime",117            "model": "gpt-realtime-2.1-mini",118            "instructions": VOICE_INSTRUCTIONS,119            "audio": {120                "output": {121                    "voice": "cedar",122                }123            },124        }125    }126 127    try:128        response = requests.post(129            "https://api.openai.com/v1/realtime/client_secrets",130            headers={131                "Authorization": f"Bearer {api_key}",132                "Content-Type": "application/json",133            },134            json=body,135            timeout=20,136        )137    except requests.RequestException as error:138        print("OpenAI ephemeral key request failed:", str(error))139        raise HTTPException(140            status_code=502,141            detail="Failed to generate OpenAI ephemeral key",142        )143 144    print("Ephemeral key request status:", response.status_code)145 146    if not response.ok:147        print("OpenAI ephemeral key error:", response.status_code, response.text)148        raise HTTPException(149            status_code=502,150            detail="Failed to generate OpenAI ephemeral key",151        )152 153    data = response.json()154    ephemeral_key = data.get("value")155 156    if not isinstance(ephemeral_key, str) or not ephemeral_key.startswith("ek_"):157        print("Invalid ephemeral key response received from OpenAI")158        raise HTTPException(159            status_code=502,160            detail="OpenAI did not return a valid ephemeral key",161        )162 163    return data164 165 166class Appointment(BaseModel):167    name: str168    phone: str169    date: str170    time: str171    service: str172 173 174@app.post("/add")175def add_appointment(appointment: Appointment):176    """Add an appointment to Airtable."""177    payload = {178        "fields": {179            "Name": appointment.name,180            "Phone": appointment.phone,181            "Date": appointment.date,182            "Time": appointment.time,183            "Service": appointment.service,184        }185    }186 187    try:188        response = requests.post(189            airtable_table_url(),190            headers=airtable_headers(),191            json=payload,192            timeout=20,193        )194    except requests.RequestException as error:195        print("Airtable add request failed:", str(error))196        raise HTTPException(status_code=502, detail="Failed to add appointment")197 198    print("Airtable add response:", response.status_code)199 200    if not response.ok:201        print("Airtable add error:", response.status_code, response.text)202        raise HTTPException(status_code=502, detail="Failed to add appointment")203 204    return response.json()205 206 207class UpdateAppointment(BaseModel):208    appointment_id: str209    name: str | None = None210    phone: str | None = None211    date: str | None = None212    time: str | None = None213    service: str | None = None214 215 216@app.post("/update")217def update_appointment(update: UpdateAppointment):218    """Update an Airtable appointment using custom Appointment ID."""219    record_id = find_airtable_record_by_appointment_id(update.appointment_id)220 221    if not record_id:222        return {"status": "error", "message": "Appointment not found"}223 224    fields = {}225 226    if update.name:227        fields["Name"] = update.name228    if update.phone:229        fields["Phone"] = update.phone230    if update.date:231        fields["Date"] = update.date232    if update.time:233        fields["Time"] = update.time234    if update.service:235        fields["Service"] = update.service236 237    if not fields:238        return {"status": "error", "message": "No fields provided to update"}239 240    try:241        response = requests.patch(242            f"{airtable_table_url()}/{record_id}",243            headers=airtable_headers(),244            json={"fields": fields},245            timeout=20,246        )247    except requests.RequestException as error:248        print("Airtable update request failed:", str(error))249        raise HTTPException(status_code=502, detail="Update failed")250 251    print("Airtable update response:", response.status_code)252 253    if not response.ok:254        print("Airtable update error:", response.status_code, response.text)255        return {"status": "error", "message": "Update failed"}256 257    return {"status": "updated"}258 259 260class CancelAppointment(BaseModel):261    appointment_id: str262 263 264@app.post("/cancel")265def cancel_appointment(cancel: CancelAppointment):266    """Cancel an Airtable appointment using custom Appointment ID."""267    record_id = find_airtable_record_by_appointment_id(cancel.appointment_id)268 269    if not record_id:270        return {"status": "error", "message": "Appointment not found"}271 272    try:273        response = requests.delete(274            f"{airtable_table_url()}/{record_id}",275            headers=airtable_headers(),276            timeout=20,277        )278    except requests.RequestException as error:279        print("Airtable delete request failed:", str(error))280        raise HTTPException(status_code=502, detail="Delete failed")281 282    print("Airtable delete response:", response.status_code)283 284    if not response.ok:285        print("Airtable delete error:", response.status_code, response.text)286        return {"status": "error", "message": "Delete failed"}287 288    return {"status": "deleted"}