[Python] im using ngrok with twilio and when i get a call it gives an error on key press

When I run the call file I have, it calls my own number. and says press any digit to run your code. When I press it, it says an application error has occurred. not sure what the cause is. I'm running this code with ngrok for faster development. the ngrok terminal is in a spearate and i run it with ngrok http 8000, i run the app in a separate terminal with a venv active using he code: uvicorn app.main:app --reload --port 8000. and i run the call file with python call_outbound.py ive fixed a unch of the errors that would come up but when i get the call from twilio on my phone, it says "this is a trail account .... press any key to execute your code", when i press any key, it says sorry an pplication error has occured and so on and hangs up, kinda stuck and appreciate any help.

my files is:

import os
import logging
from fastapi import FastAPI, Request, HTTPException, Form
from fastapi.responses import Response, PlainTextResponse
from fastapi.staticfiles import StaticFiles
from dotenv import load_dotenv
from twilio.rest import Client
from twilio.base.exceptions import TwilioRestException
from urllib.parse import urljoin

# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("app.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)

load_dotenv()

app = FastAPI(title="On-Hold Ads System")
app.mount("/static", StaticFiles(directory="static"), name="static")

# Configuration
TWILIO_TRIAL_MODE = True # Set to False for production

# Demo data
playlists = {
("demobiz", "demodept"): ["demo_ad.mp3", "demo_music.mp3"]
}
counters = {("demobiz", "demodept"): 0}

# Helper functions
def get_next_audio(biz: str, dept: str) -> str:
"""Get next audio file in rotation"""
try:
key = (biz, dept)
if key not in playlists:
raise KeyError(f"No playlist for {key}")
idx = counters.get(key, 0)
new_idx = (idx + 1) % len(playlists[key])
counters[key] = new_idx
next_file = playlists[key][new_idx]
logger.debug(f"Audio rotation: {key} idx={idx}->{new_idx} file={next_file}")
return next_file
except Exception as e:
logger.error(f"Error in get_next_audio: {str(e)}", exc_info=True)
raise

def send_sms(to: str, biz_name: str) -> bool:
"""Send SMS via Twilio"""
try:
logger.info(f"Attempting to send SMS to {to}")
client = Client(os.getenv("TWILIO_SID"), os.getenv("TWILIO_TOKEN"))
message = client.messages.create(
to=to,
from_=os.getenv("TWILIO_PHONE"),
body=f"Thanks for your interest in {biz_name}! We'll contact you soon."
)
logger.info(f"SMS sent successfully! SID: {message.sid}")
return True
except TwilioRestException as e:
logger.error(f"SMS failed to {to}: {str(e)}", exc_info=True)
return False

# Core endpoints
@app.api_route("/hold", methods=["GET", "POST"])
async def hold_handler(request: Request):
"""Initial endpoint that redirects to bypass trial message"""
try:
logger.info(f"Incoming request to /hold: {request.method} {request.url}")
params = request.query_params
biz = params.get("biz", "demobiz")
dept = params.get("dept", "demodept")
caller = params.get("caller", "unknown")

logger.debug(f"Params - biz: {biz}, dept: {dept}, caller: {caller}")

base_url = str(request.base_url).rstrip('/')
redirect_url = urljoin(base_url, f'/hold/real-start?biz={biz}&dept={dept}&caller={caller}')

twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Redirect method="POST">{redirect_url}</Redirect>
</Response>"""

logger.debug(f"Generated TwiML for /hold:\n{twiml}")
return Response(content=twiml, media_type="application/xml")

except Exception as e:
logger.error(f"Error in hold_handler: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")

@app.api_route("/hold/real-start", methods=["GET", "POST"])
async def real_start_handler(request: Request):
"""Main endpoint that handles the actual on-hold experience"""
try:
logger.info(f"Incoming request to /hold/real-start: {request.method} {request.url}")
params = request.query_params
biz = params.get("biz", "demobiz")
dept = params.get("dept", "demodept")
caller = params.get("caller", "unknown")

logger.debug(f"Params - biz: {biz}, dept: {dept}, caller: {caller}")

audio_file = get_next_audio(biz, dept)
logger.info(f"Serving {audio_file} for {caller}")

base_url = str(request.base_url).rstrip('/')
play_url = f"{base_url}/static/{audio_file}"
action_url = urljoin(base_url, f'/hold/keypress?biz={biz}&dept={dept}&caller={caller}')
redirect_url = urljoin(base_url, f'/hold/real-start?biz={biz}&dept={dept}&caller={caller}')

twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Gather input="dtmf" timeout="5" numDigits="1"
action="{action_url}"
method="POST">
<Play>{play_url}</Play>
</Gather>
<Redirect>{redirect_url}</Redirect>
</Response>"""

logger.debug(f"Generated TwiML for /hold/real-start:\n{twiml}")
return Response(content=twiml, media_type="application/xml")

except KeyError as e:
logger.error(f"Playlist error: {str(e)}", exc_info=True)
return Response(content=f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Say>Error in on-hold system. Please call back later.</Say>
<Hangup/>
</Response>""", media_type="application/xml")
except Exception as e:
logger.error(f"Error in real_start_handler: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")

@app.post("/hold/keypress")
async def keypress_handler(request: Request):
"""Handle DTMF keypresses"""
try:
logger.info("Incoming keypress request")
form_data = await request.form()
digit = form_data.get("Digits", "")
params = request.query_params

biz = params.get("biz", "demobiz")
dept = params.get("dept", "demodept")
caller = params.get("caller", "unknown")

logger.info(f"Keypress from {caller}: digit='{digit}'")
logger.debug(f"Full form data: {dict(form_data)}")
logger.debug(f"Query params: {dict(params)}")

base_url = str(request.base_url).rstrip('/')
redirect_url = urljoin(base_url, f'/hold/real-start?biz={biz}&dept={dept}&caller={caller}')

if digit == "5":
logger.info("Processing digit 5 (SMS request)")
if send_sms(caller, biz):
response = """<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Say>We're sending you an SMS with more information.</Say>
<Redirect>{redirect_url}</Redirect>
</Response>"""
else:
response = """<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Say>Sorry, we couldn't send the SMS right now.</Say>
<Redirect>{redirect_url}</Redirect>
</Response>"""
else:
logger.info(f"Processing digit {digit} (standard redirect)")
response = """<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Redirect>{redirect_url}</Redirect>
</Response>"""

formatted_response = response.format(redirect_url=redirect_url)
logger.debug(f"Generated keypress response:\n{formatted_response}")
return Response(content=formatted_response, media_type="application/xml")

except Exception as e:
logger.error(f"Error in keypress_handler: {str(e)}", exc_info=True)
return Response(content="""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Say>An error occurred. Goodbye.</Say>
<Hangup/>
</Response>""", media_type="application/xml")

@app.post("/call-status")
async def status_handler(request: Request):
"""Handle Twilio call status updates"""
try:
form_data = await request.form()
call_sid = form_data.get("CallSid", "unknown")
call_status = form_data.get("CallStatus", "unknown")

logger.info(f"Call status update - SID: {call_sid}, Status: {call_status}")
logger.debug(f"Full status update: {dict(form_data)}")

return Response(content="<Response></Response>", media_type="application/xml")
except Exception as e:
logger.error(f"Error in status_handler: {str(e)}", exc_info=True)
return Response(content="<Response></Response>", media_type="application/xml")

# Debug endpoints
@app.get("/test-sms")
async def test_sms_endpoint():
"""Test SMS functionality"""
try:
test_number = "+16041231234" # Your verified number
logger.info(f"Testing SMS to {test_number}")
if send_sms(test_number, "Test Business"):
return PlainTextResponse("SMS sent successfully!")
return PlainTextResponse("SMS failed", status_code=500)
except Exception as e:
logger.error(f"Error in test_sms_endpoint: {str(e)}", exc_info=True)
return PlainTextResponse("Test failed", status_code=500)

@app.get("/test-audio")
async def test_audio_endpoint():
"""Verify audio files are accessible"""
try:
logger.info("Testing audio files availability")
files = [f for f in os.listdir("static") if f.endswith(".mp3")]
return {
"available_files": files,
"playlist": playlists.get(("demobiz", "demodept"), [])
}
except Exception as e:
logger.error(f"Error in test_audio_endpoint: {str(e)}", exc_info=True)
return {"error": "Unable to check audio files"}

@app.get("/health")
async def health_check():
"""Simple health check"""
return {"status": "ok", "service": "on-hold-system"}


# call_outbound.py
import os
from twilio.rest import Client
from dotenv import load_dotenv

load_dotenv()

account_sid = os.getenv("TWILIO_SID")
auth_token = os.getenv("TWILIO_TOKEN")
from_number = os.getenv("TWILIO_PHONE")

client = Client(account_sid, auth_token)

# ⚠️
ngrok_url = "https://LETSSAYTHISISMYURL.ngrok-free.app" # Keep this updated

# Build the full webhook with query params
twiml_url = f"{ngrok_url}/hold?biz=demobiz&dept=demodept&caller=+16041231234"

print(f"Using TwiML URL: {twiml_url}") # Verify output looks correct

# Trigger the call
call = client.calls.create(
url=twiml_url,
to="+16041231234", # Your verified number
from_=from_number
)
# Add this function to validate URLs
from urllib.parse import urlparse

def validate_url(url):
try:
result = urlparse(url)
return all([result.scheme in ('http', 'https'), result.netloc])
except:
return False

# Then modify your call creation:
ngrok_url = "https://LETSSAYTHISISMYURL.ngrok-free.app"
twiml_url = f"{ngrok_url}/hold/real-start?biz=demobiz&dept=demodept&caller=+16041231234"

if not validate_url(twiml_url):
print(f"ERROR: Invalid URL - {twiml_url}")
exit(1)

try:
call = client.calls.create(
url=twiml_url,
to="+16041231234",
from_=from_number,
status_callback=f"{ngrok_url}/call-status", # Optional callback
status_callback_event=['initiated', 'ringing', 'answered', 'completed']
)
print(f"Call SID: {call.sid}")
except Exception as e:
print(f"Call failed: {e}")

print(f"Started call: {call.sid}")```


Not sure where the error is coming from.
The code and the idea associated are owned as per the request of the product owner., I would really appreciate you guys for help thanks

Continue reading...