feat: support inbound vonage calls (#480)

* feat: support inbound vonage calls

* fix: drift check

* feat: add warning with missing signature secret

* docs: vonage inbound steps

* chore: upgrade pipecat submodule

api/services/telephony/providers/vonage/__init__.py

@@ -21,6 +21,7 @@ def _config_loader(value: Dict[str, Any]) -> Dict[str, Any]:
         "private_key": value.get("private_key"),
         "api_key": value.get("api_key"),
         "api_secret": value.get("api_secret"),
+        "signature_secret": value.get("signature_secret"),
         "from_numbers": value.get("from_numbers", []),
     }
 
@@ -49,6 +50,13 @@ _UI_METADATA = ProviderUIMetadata(
             type="password",
             sensitive=True,
         ),
+        ProviderUIField(
+            name="signature_secret",
+            label="Signature Secret",
+            type="password",
+            sensitive=True,
+            description="Vonage signature secret for signed webhook verification",
+        ),
         ProviderUIField(
             name="from_numbers",
             label="Phone Numbers",

api/services/telephony/providers/vonage/config.py

@@ -1,6 +1,6 @@
 """Vonage telephony configuration schemas."""
 
-from typing import List, Literal
+from typing import List, Literal, Optional
 
 from pydantic import BaseModel, Field
 
@@ -13,6 +13,10 @@ class VonageConfigurationRequest(BaseModel):
     api_secret: str = Field(..., description="Vonage API Secret")
     application_id: str = Field(..., description="Vonage Application ID")
     private_key: str = Field(..., description="Private key for JWT generation")
+    signature_secret: Optional[str] = Field(
+        None,
+        description="Vonage signature secret used to verify signed webhooks",
+    )
     from_numbers: List[str] = Field(
         default_factory=list,
         description="List of Vonage phone numbers (without + prefix)",
@@ -27,4 +31,5 @@ class VonageConfigurationResponse(BaseModel):
     api_key: str  # Masked
     api_secret: str  # Masked
     private_key: str  # Masked
+    signature_secret: Optional[str] = None  # Masked
     from_numbers: List[str]

api/services/telephony/providers/vonage/provider.py

@@ -2,6 +2,7 @@
 Vonage (Nexmo) implementation of the TelephonyProvider interface.
 """
 
+import hashlib
 import json
 import random
 import time
@@ -44,12 +45,14 @@ class VonageProvider(TelephonyProvider):
                 - api_secret: Vonage API Secret
                 - application_id: Vonage Application ID
                 - private_key: Private key for JWT generation
+                - signature_secret: Signature secret for signed webhooks
                 - from_numbers: List of phone numbers to use
         """
         self.api_key = config.get("api_key")
         self.api_secret = config.get("api_secret")
         self.application_id = config.get("application_id")
         self.private_key = config.get("private_key")
+        self.signature_secret = config.get("signature_secret")
         self.from_numbers = config.get("from_numbers", [])
 
         # Handle both single number (string) and multiple numbers (list)
@@ -186,17 +189,18 @@ class VonageProvider(TelephonyProvider):
         Verify Vonage webhook signature for security.
         Vonage uses JWT for webhook signatures.
         """
-        if not self.api_secret:
-            logger.error("No API secret available for webhook signature verification")
+        if not self.signature_secret:
+            logger.error(
+                "No signature secret available for Vonage webhook verification"
+            )
             return False
 
         try:
-            # Vonage sends JWT in Authorization header. Verify the JWT signature
-            decoded = jwt.decode(
+            jwt.decode(
                 signature,
-                self.api_secret,
+                self.signature_secret,
                 algorithms=["HS256"],
-                options={"verify_signature": True},
+                options={"verify_signature": True, "verify_aud": False},
             )
             return True
         except jwt.InvalidTokenError:
@@ -295,9 +299,13 @@ class VonageProvider(TelephonyProvider):
             "ringing": TelephonyCallStatus.RINGING,
             "answered": TelephonyCallStatus.ANSWERED,
             "complete": TelephonyCallStatus.COMPLETED,
+            "completed": TelephonyCallStatus.COMPLETED,
+            "disconnected": TelephonyCallStatus.COMPLETED,
             "failed": TelephonyCallStatus.FAILED,
             "busy": TelephonyCallStatus.BUSY,
             "timeout": TelephonyCallStatus.NO_ANSWER,
+            "unanswered": TelephonyCallStatus.NO_ANSWER,
+            "cancelled": TelephonyCallStatus.NO_ANSWER,
             "rejected": TelephonyCallStatus.BUSY,
         }
 
@@ -349,6 +357,8 @@ class VonageProvider(TelephonyProvider):
                 if workflow_run.gathered_context
                 else None
             )
+            if not call_uuid and workflow_run.gathered_context:
+                call_uuid = workflow_run.gathered_context.get("call_id")
 
             if not call_uuid:
                 logger.error(
@@ -400,26 +410,126 @@ class VonageProvider(TelephonyProvider):
         """
         Determine if this provider can handle the incoming webhook.
         """
-        return False
+        claims = cls._decode_unverified_signed_claims(headers)
+        if claims.get("api_key") or claims.get("application_id"):
+            return True
+
+        return bool(
+            webhook_data.get("uuid")
+            and webhook_data.get("conversation_uuid")
+            and webhook_data.get("from")
+            and webhook_data.get("to")
+        )
 
     @staticmethod
-    def parse_inbound_webhook(webhook_data: Dict[str, Any]) -> NormalizedInboundData:
+    def parse_inbound_webhook(
+        webhook_data: Dict[str, Any], headers: Optional[Dict[str, str]] = None
+    ) -> NormalizedInboundData:
         """
         Parse Vonage-specific inbound webhook data into normalized format.
         """
+        claims = VonageProvider._decode_unverified_signed_claims(headers or {})
+        direction = webhook_data.get("direction") or "inbound"
+        status = webhook_data.get("status") or "started"
+
         return NormalizedInboundData(
             provider=VonageProvider.PROVIDER_NAME,
             call_id=webhook_data.get("uuid", ""),
             from_number=webhook_data.get("from", ""),
             to_number=webhook_data.get("to", ""),
-            direction=webhook_data.get("direction", ""),
-            call_status=webhook_data.get("status", ""),
-            account_id=webhook_data.get("account_id"),
+            direction=direction,
+            call_status=status,
+            account_id=claims.get("api_key") or webhook_data.get("account_id"),
             from_country=None,
             to_country=None,
             raw_data=webhook_data,
         )
 
+    @staticmethod
+    def _header(headers: Dict[str, str], name: str) -> Optional[str]:
+        for key, value in headers.items():
+            if key.lower() == name.lower():
+                return value
+        return None
+
+    @classmethod
+    def _bearer_token(cls, headers: Dict[str, str]) -> Optional[str]:
+        auth_header = cls._header(headers, "authorization")
+        if not auth_header:
+            return None
+        parts = auth_header.split(None, 1)
+        if len(parts) != 2 or parts[0].lower() != "bearer":
+            return None
+        return parts[1].strip()
+
+    @classmethod
+    def _decode_unverified_signed_claims(
+        cls, headers: Dict[str, str]
+    ) -> Dict[str, Any]:
+        token = cls._bearer_token(headers)
+        if not token:
+            return {}
+        try:
+            claims = jwt.decode(
+                token,
+                options={
+                    "verify_signature": False,
+                    "verify_aud": False,
+                    "verify_exp": False,
+                },
+            )
+        except jwt.InvalidTokenError:
+            return {}
+        return claims if isinstance(claims, dict) else {}
+
+    def _verify_signed_claims(
+        self, headers: Dict[str, str], body: str = ""
+    ) -> Optional[Dict[str, Any]]:
+        token = self._bearer_token(headers)
+        if not token:
+            logger.warning("Missing Vonage Authorization bearer token")
+            return None
+        if not self.signature_secret:
+            logger.error("Missing Vonage signature_secret for signed webhook")
+            return None
+
+        try:
+            claims = jwt.decode(
+                token,
+                self.signature_secret,
+                algorithms=["HS256"],
+                options={"verify_signature": True, "verify_aud": False},
+            )
+        except jwt.InvalidTokenError as exc:
+            logger.warning(f"Invalid Vonage signed webhook JWT: {exc}")
+            return None
+
+        if claims.get("iss") != "Vonage":
+            logger.warning("Vonage signed webhook JWT has unexpected issuer")
+            return None
+
+        if self.api_key and claims.get("api_key") != self.api_key:
+            logger.warning("Vonage signed webhook api_key does not match config")
+            return None
+
+        claim_application_id = claims.get("application_id")
+        if (
+            self.application_id
+            and claim_application_id
+            and claim_application_id != self.application_id
+        ):
+            logger.warning("Vonage signed webhook application_id does not match config")
+            return None
+
+        payload_hash = claims.get("payload_hash")
+        if payload_hash:
+            actual_hash = hashlib.sha256(body.encode("utf-8")).hexdigest()
+            if actual_hash != payload_hash:
+                logger.warning("Vonage signed webhook payload hash mismatch")
+                return None
+
+        return claims
+
     @staticmethod
     def validate_account_id(config_data: dict, webhook_account_id: str) -> bool:
         """Validate Vonage account_id from webhook matches configuration"""
@@ -437,9 +547,10 @@ class VonageProvider(TelephonyProvider):
         body: str = "",
     ) -> bool:
         """
-        Vonage inbound signature verification - minimalist implementation.
+        Verify Vonage signed webhook JWT and optional payload hash.
         """
-        return True
+        claims = self._verify_signed_claims(headers, body)
+        return claims is not None
 
     async def configure_inbound(
         self, address: str, webhook_url: Optional[str]
@@ -486,6 +597,15 @@ class VonageProvider(TelephonyProvider):
                 ),
             )
 
+        if not self.signature_secret:
+            return ProviderSyncResult(
+                ok=False,
+                message=(
+                    "Vonage signature_secret is required because inbound calls "
+                    "use signed webhook verification"
+                ),
+            )
+
         app_endpoint = f"{self.base_url}/v2/applications/{self.application_id}"
         auth = aiohttp.BasicAuth(self.api_key, self.api_secret)
 
@@ -510,12 +630,18 @@ class VonageProvider(TelephonyProvider):
         capabilities = app_data.get("capabilities") or {}
         voice = capabilities.get("voice") or {}
         webhooks = voice.get("webhooks") or {}
+        backend_endpoint, _ = await get_backend_endpoints()
 
         webhooks["answer_url"] = {
             "address": webhook_url,
             "http_method": "POST",
         }
+        webhooks["event_url"] = {
+            "address": f"{backend_endpoint}/api/v1/telephony/vonage/events",
+            "http_method": "POST",
+        }
         voice["webhooks"] = webhooks
+        voice["signed_callbacks"] = True
         capabilities["voice"] = voice
 
         update_body = {
@@ -561,13 +687,24 @@ class VonageProvider(TelephonyProvider):
         """
         Generate NCCO response for inbound Vonage webhook.
         """
-        # Minimalist NCCO response for interface compliance
         ncco_response = [
             {
-                "action": "talk",
-                "text": "Vonage inbound calls are not currently supported.",
-            },
-            {"action": "hangup"},
+                "action": "connect",
+                "eventUrl": [
+                    f"{backend_endpoint}/api/v1/telephony/vonage/events/{workflow_run_id}"
+                ],
+                "endpoint": [
+                    {
+                        "type": "websocket",
+                        "uri": websocket_url,
+                        "content-type": "audio/l16;rate=16000",
+                        "headers": {
+                            "workflow_run_id": str(workflow_run_id),
+                            "call_uuid": normalized_data.call_id,
+                        },
+                    }
+                ],
+            }
         ]
 
         return Response(

api/services/telephony/providers/vonage/routes.py

@@ -7,16 +7,12 @@ provider registry — see ProviderSpec.router.
 import json
 from typing import Optional
 
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, HTTPException, Request
 from loguru import logger
 from pipecat.utils.run_context import set_current_run_id
 
 from api.db import db_client
 from api.services.telephony.factory import get_telephony_provider_for_run
-from api.services.telephony.status_processor import (
-    StatusCallbackRequest,
-    _process_status_update,
-)
 
 router = APIRouter()
 
@@ -45,28 +41,33 @@ async def handle_ncco_webhook(
     return json.loads(response_content)
 
 
-@router.post("/vonage/events/{workflow_run_id}")
-async def handle_vonage_events(
-    request: Request,
-    workflow_run_id: int,
-):
-    """Handle Vonage-specific event webhooks.
+async def _read_json_body(request: Request) -> tuple[dict, str]:
+    body_bytes = await request.body()
+    try:
+        raw_body = body_bytes.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise HTTPException(
+            status_code=400, detail="Webhook body is not valid UTF-8"
+        ) from exc
+    try:
+        return json.loads(raw_body or "{}"), raw_body
+    except json.JSONDecodeError as exc:
+        raise HTTPException(status_code=400, detail="Webhook body is not JSON") from exc
 
-    Vonage sends all call events to a single endpoint.
-    Events include: started, ringing, answered, complete, failed, etc.
-    """
+
+async def _handle_vonage_event_request(request: Request, workflow_run_id: int):
     set_current_run_id(workflow_run_id)
-    # Parse the event data
-    event_data = await request.json()
-    logger.info(f"[run {workflow_run_id}] Received Vonage event: {event_data}")
+    event_data, raw_body = await _read_json_body(request)
+    logger.info(
+        f"[run {workflow_run_id}] Received Vonage event "
+        f"uuid={event_data.get('uuid')} status={event_data.get('status')}"
+    )
 
-    # Get workflow run for processing
     workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id)
     if not workflow_run:
         logger.error(f"[run {workflow_run_id}] Workflow run not found")
         return {"status": "error", "message": "Workflow run not found"}
 
-    # Get workflow and provider
     workflow = await db_client.get_workflow_by_id(workflow_run.workflow_id)
     if not workflow:
         logger.error(f"[run {workflow_run_id}] Workflow not found")
@@ -75,11 +76,18 @@ async def handle_vonage_events(
     provider = await get_telephony_provider_for_run(
         workflow_run, workflow.organization_id
     )
+    signature_valid = await provider.verify_inbound_signature(
+        str(request.url), event_data, dict(request.headers), raw_body
+    )
+    if not signature_valid:
+        raise HTTPException(status_code=401, detail="Invalid webhook signature")
+
+    from api.services.telephony.status_processor import (
+        StatusCallbackRequest,
+        _process_status_update,
+    )
 
-    # Parse the event data into generic format
     parsed_data = provider.parse_status_callback(event_data)
-
-    # Create StatusCallbackRequest from parsed data
     status_update = StatusCallbackRequest(
         call_id=parsed_data["call_id"],
         status=parsed_data["status"],
@@ -90,8 +98,35 @@ async def handle_vonage_events(
         extra=parsed_data.get("extra", {}),
     )
 
-    # Process the status update
     await _process_status_update(workflow_run_id, status_update)
-
-    # Return 204 No Content as expected by Vonage
+    return {"status": "ok"}
+
+
+@router.post("/vonage/events/{workflow_run_id}")
+async def handle_vonage_events(
+    request: Request,
+    workflow_run_id: int,
+):
+    """Handle Vonage-specific event webhooks.
+
+    Vonage sends all call events to a single endpoint.
+    Events include: started, ringing, answered, complete, failed, etc.
+    """
+    return await _handle_vonage_event_request(request, workflow_run_id)
+
+
+@router.post("/vonage/events")
+async def handle_vonage_events_without_run(request: Request):
+    """Handle application-level events by resolving the run from call UUID."""
+    event_data, _ = await _read_json_body(request)
+    call_id = event_data.get("uuid")
+    if call_id:
+        workflow_run = await db_client.get_workflow_run_by_call_id(call_id)
+        if workflow_run:
+            return await _handle_vonage_event_request(request, workflow_run.id)
+
+    logger.info(
+        "Received unmatched Vonage application event "
+        f"uuid={event_data.get('uuid')} status={event_data.get('status')}"
+    )
     return {"status": "ok"}