normalise wss urls from Ngrok

Author: Nash0x7E2

plugins/twilio/vision_agents/plugins/twilio/models.py

@@ -10,6 +10,8 @@
 from twilio.request_validator import RequestValidator
 from twilio.twiml.voice_response import VoiceResponse, Connect
 
+from .utils import normalize_websocket_url
+
 logger = logging.getLogger(__name__)
 
 
@@ -18,11 +20,13 @@ def create_media_stream_twiml(websocket_url: str) -> str:
     Create TwiML that starts a media stream to the given websocket URL.
 
     Args:
-        websocket_url: The websocket URL to stream audio to (e.g. wss://example.com/media/123)
+        websocket_url: The websocket URL to stream audio to (e.g. wss://example.com/media/123).
+                       https:// URLs are automatically converted to wss://.
 
     Returns:
         TwiML string.
     """
+    websocket_url = normalize_websocket_url(websocket_url)
     response = VoiceResponse()
     connect = Connect()
     connect.stream(url=websocket_url)
@@ -40,19 +44,21 @@ def create_media_stream_response(websocket_url: str) -> Response:
     Returns:
         FastAPI Response with TwiML content.
     """
-    return Response(content=create_media_stream_twiml(websocket_url), media_type="text/xml")
+    return Response(
+        content=create_media_stream_twiml(websocket_url), media_type="text/xml"
+    )
 
 
 class TwilioSignatureVerifier:
     """
     Verifies Twilio webhook signatures to ensure requests are authentic.
-    
+
     Uses the TWILIO_AUTH_TOKEN environment variable to validate
     the X-Twilio-Signature header against the request URL and body.
-    
+
     Example:
         verifier = TwilioSignatureVerifier()
-        
+
         @app.post("/twilio/voice")
         async def webhook(
             request: Request,
@@ -62,17 +68,17 @@ async def webhook(
             # Request is verified, safe to process
             ...
     """
-    
+
     def __init__(self, auth_token: Optional[str] = None):
         """
         Initialize the verifier.
-        
+
         Args:
             auth_token: Twilio Auth Token. If not provided, reads from
                        TWILIO_AUTH_TOKEN environment variable.
         """
         self._auth_token = auth_token
-    
+
     @property
     def auth_token(self) -> str:
         """Get the auth token, falling back to environment variable."""
@@ -82,34 +88,34 @@ def auth_token(self) -> str:
                 "TWILIO_AUTH_TOKEN environment variable is required for signature verification"
             )
         return token
-    
+
     async def __call__(self, request: Request) -> None:
         """
         FastAPI dependency that verifies Twilio signature.
-        
+
         Raises:
             HTTPException: If signature is missing or invalid.
         """
         from fastapi import HTTPException
-        
+
         signature = request.headers.get("X-Twilio-Signature")
         if not signature:
             logger.warning("Missing X-Twilio-Signature header")
             raise HTTPException(status_code=403, detail="Missing Twilio signature")
-        
+
         # Get the full URL (Twilio uses the URL for signature validation)
         url = str(request.url)
-        
+
         # Get form data as dict
         form = await request.form()
         params = {key: value for key, value in form.items()}
-        
+
         # Validate the signature
         validator = RequestValidator(self.auth_token)
         if not validator.validate(url, params, signature):
             logger.warning(f"Invalid Twilio signature for {url}")
             raise HTTPException(status_code=403, detail="Invalid Twilio signature")
-        
+
         logger.debug(f"Twilio signature verified for {url}")
 
 
@@ -120,28 +126,39 @@ async def __call__(self, request: Request) -> None:
 class CallWebhookInput(BaseModel):
     """
     Twilio voice webhook form data.
-    
+
     This model represents the form data sent by Twilio when a call is received.
     Use with FastAPI's Form() for automatic parsing.
-    
+
     Example:
         @app.post("/twilio/voice")
         async def webhook(data: CallWebhookInput = Depends(CallWebhookInput.as_form)):
             print(f"Call from {data.caller} to {data.to}")
     """
-    
+
     # Call identification
-    call_sid: str = Field(alias="CallSid", description="Unique identifier for this call")
+    call_sid: str = Field(
+        alias="CallSid", description="Unique identifier for this call"
+    )
     account_sid: str = Field(alias="AccountSid", description="Twilio account SID")
     api_version: str = Field(alias="ApiVersion", default="2010-04-01")
-    
+
     # Call status
-    call_status: str = Field(alias="CallStatus", description="Current call status (ringing, in-progress, etc.)")
-    direction: str = Field(alias="Direction", description="Call direction (inbound or outbound)")
-    
+    call_status: str = Field(
+        alias="CallStatus",
+        description="Current call status (ringing, in-progress, etc.)",
+    )
+    direction: str = Field(
+        alias="Direction", description="Call direction (inbound or outbound)"
+    )
+
     # From (caller) information
-    from_number: str = Field(alias="From", description="Caller's phone number (E.164 format)")
-    caller: str = Field(alias="Caller", description="Caller's phone number (same as From)")
+    from_number: str = Field(
+        alias="From", description="Caller's phone number (E.164 format)"
+    )
+    caller: str = Field(
+        alias="Caller", description="Caller's phone number (same as From)"
+    )
     caller_city: Optional[str] = Field(alias="CallerCity", default=None)
     caller_state: Optional[str] = Field(alias="CallerState", default=None)
     caller_zip: Optional[str] = Field(alias="CallerZip", default=None)
@@ -150,7 +167,7 @@ async def webhook(data: CallWebhookInput = Depends(CallWebhookInput.as_form)):
     from_state: Optional[str] = Field(alias="FromState", default=None)
     from_zip: Optional[str] = Field(alias="FromZip", default=None)
     from_country: Optional[str] = Field(alias="FromCountry", default=None)
-    
+
     # To (called) information
     to: str = Field(alias="To", description="Called phone number (E.164 format)")
     called: str = Field(alias="Called", description="Called phone number (same as To)")
@@ -162,15 +179,17 @@ async def webhook(data: CallWebhookInput = Depends(CallWebhookInput.as_form)):
     to_state: Optional[str] = Field(alias="ToState", default=None)
     to_zip: Optional[str] = Field(alias="ToZip", default=None)
     to_country: Optional[str] = Field(alias="ToCountry", default=None)
-    
+
     # STIR/SHAKEN verification
-    stir_verstat: Optional[str] = Field(alias="StirVerstat", default=None, description="STIR/SHAKEN verification status")
-    
+    stir_verstat: Optional[str] = Field(
+        alias="StirVerstat", default=None, description="STIR/SHAKEN verification status"
+    )
+
     # Call token for additional security
     call_token: Optional[str] = Field(alias="CallToken", default=None)
-    
+
     model_config = {"populate_by_name": True}
-    
+
     @classmethod
     def as_form(
         cls,
@@ -204,7 +223,7 @@ def as_form(
     ) -> "CallWebhookInput":
         """
         Create CallWebhookInput from FastAPI Form fields.
-        
+
         Usage:
             @app.post("/twilio/voice")
             async def webhook(data: CallWebhookInput = Depends(CallWebhookInput.as_form)):
@@ -239,4 +258,3 @@ async def webhook(data: CallWebhookInput = Depends(CallWebhookInput.as_form)):
             StirVerstat=StirVerstat,
             CallToken=CallToken,
         )
-

plugins/twilio/vision_agents/plugins/twilio/utils.py

@@ -0,0 +1,12 @@
+def normalize_websocket_url(url: str) -> str:
+    """Normalize a URL to use the wss:// scheme for Twilio media streams."""
+    url = url.strip()
+    if url.startswith("wss://https://"):
+        url = "wss://" + url[14:]
+    elif url.startswith("wss://http://"):
+        url = "wss://" + url[13:]
+    elif url.startswith("https://"):
+        url = "wss://" + url[8:]
+    elif url.startswith("http://"):
+        url = "wss://" + url[8:]
+    return url