diff --git a/.serena/project.local.yml b/.serena/project.local.yml new file mode 100644 index 00000000..36cd3def --- /dev/null +++ b/.serena/project.local.yml @@ -0,0 +1,5 @@ +# This file allows you to locally override settings in project.yml for development purposes. +# +# Use the same keys as in project.yml here. Any setting you specify will override the corresponding +# setting in project.yml, allowing you to customise the configuration for your local development environment +# without affecting the project configuration in project.yml (which is intended to be versioned). diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 34e98c05..dd10dcd5 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -91,6 +91,7 @@ api_router.include_router(prompt_partials.router) api_router.include_router(prompt_optimization.router) api_router.include_router(telephony.router) +api_router.include_router(telephony.plivo_webhook_router) api_router.include_router(vobiz_telephony.router) api_router.include_router(call_imports.router) api_router.include_router(call_import_schemas.router) diff --git a/app/api/v1/media.py b/app/api/v1/media.py index d35d99a3..56dcb536 100644 --- a/app/api/v1/media.py +++ b/app/api/v1/media.py @@ -6,5 +6,5 @@ media_router = APIRouter() media_router.include_router(vobiz_telephony.webhook_router) -media_router.include_router(vobiz_telephony.ws_router) +media_router.include_router(vobiz_telephony.carrier_ws_router) media_router.include_router(voice_agent.ws_router) diff --git a/app/api/v1/routes/call_imports.py b/app/api/v1/routes/call_imports.py index 70551508..f9a14ce7 100644 --- a/app/api/v1/routes/call_imports.py +++ b/app/api/v1/routes/call_imports.py @@ -1360,11 +1360,14 @@ def _validate_direct_url_import_ready( ) -def _validate_exotel_import_ready( +def _validate_credentialed_import_ready( + provider: str, parameters: List[CallImportSchemaParameter], parameter_mapping: Dict[str, Any], ) -> None: - """Ensure Exotel credentialed import has a mapped recording_url column.""" + """Ensure credentialed import has a mapped recording_url column.""" + provider_key = (provider or "").lower() + provider_label = provider_key.capitalize() rec_url_param = next( ( p @@ -1377,7 +1380,7 @@ def _validate_exotel_import_ready( raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( - "Exotel import requires a schema parameter of type " + f"{provider_label} import requires a schema parameter of type " "'recording_url'." ), ) @@ -1386,12 +1389,20 @@ def _validate_exotel_import_ready( raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( - "Exotel import requires the 'recording_url' parameter to " + f"{provider_label} import requires the 'recording_url' parameter to " "be mapped to a source column." ), ) +def _validate_exotel_import_ready( + parameters: List[CallImportSchemaParameter], + parameter_mapping: Dict[str, Any], +) -> None: + """Ensure Exotel credentialed import has a mapped recording_url column.""" + _validate_credentialed_import_ready("exotel", parameters, parameter_mapping) + + def _is_manual_audio_call_import(call_import: CallImport) -> bool: """True for batches created via manual audio upload (recordings already in S3).""" return (call_import.source_format or "").lower() == "audio" @@ -2150,9 +2161,11 @@ async def start_call_import( payload.provider or "", ) _validate_telephony_credentials_live(db, organization_id, integration) - if (integration.provider or "").lower() == "exotel": - _validate_exotel_import_ready( - parameters, dict(call_import.parameter_mapping or {}) + if (integration.provider or "").lower() in {"exotel", "plivo"}: + _validate_credentialed_import_ready( + integration.provider, + parameters, + dict(call_import.parameter_mapping or {}), ) else: _validate_direct_url_import_ready( @@ -2345,8 +2358,10 @@ async def upload_call_import_csv( integration = _resolve_telephony_integration( db, organization_id, telephony_integration_id, provider or "" ) - if (integration.provider or "").lower() == "exotel": - _validate_exotel_import_ready(parameters, cleaned_mapping) + if (integration.provider or "").lower() in {"exotel", "plivo"}: + _validate_credentialed_import_ready( + integration.provider, parameters, cleaned_mapping + ) else: _validate_direct_url_import_ready(parameters, cleaned_mapping) integration = None @@ -3527,14 +3542,15 @@ async def retry_failed_call_import_rows( _validate_telephony_credentials_live(db, organization_id, integration) call_import.provider = integration.provider call_import.telephony_integration_id = integration.id - if (integration.provider or "").lower() == "exotel": + if (integration.provider or "").lower() in {"exotel", "plivo"}: schema = _resolve_schema( db, organization_id, call_import.workspace_id, call_import.schema_id, ) - _validate_exotel_import_ready( + _validate_credentialed_import_ready( + integration.provider, list(schema.parameters), dict(call_import.parameter_mapping or {}), ) diff --git a/app/api/v1/routes/telephony.py b/app/api/v1/routes/telephony.py index 6d4f8b9a..be00d481 100644 --- a/app/api/v1/routes/telephony.py +++ b/app/api/v1/routes/telephony.py @@ -33,6 +33,7 @@ from app.services.telephony.platform_outbound_pool import outbound_pool_api_payload router = APIRouter(prefix="/telephony", tags=["Telephony"]) +plivo_webhook_router = APIRouter(prefix="/telephony/plivo", tags=["Plivo Telephony Webhooks"]) class TelephonyAvailableNumberResponse(BaseModel): @@ -557,25 +558,52 @@ async def _read_webhook_params(request: Request) -> Dict[str, Any]: return params -@router.post("/webhooks/answer") -async def telephony_answer_webhook(request: Request, db: Session = Depends(get_db)): +async def _handle_plivo_answer_webhook(request: Request, db: Session) -> Response: params = await _read_webhook_params(request) verify_plivo_webhook(request, params, "answer", db) xml = telephony_service.handle_answer_webhook(params, db) return Response(content=xml, media_type="application/xml") -@router.post("/webhooks/events") -async def telephony_events_webhook(request: Request, db: Session = Depends(get_db)): +async def _handle_plivo_events_webhook(request: Request, db: Session) -> Dict[str, str]: params = await _read_webhook_params(request) verify_plivo_webhook(request, params, "events", db) telephony_service.handle_event_webhook(params, db) return {"status": "ok"} -@router.post("/webhooks/masking") -async def telephony_masking_webhook(request: Request, db: Session = Depends(get_db)): +async def _handle_plivo_masking_webhook(request: Request, db: Session) -> Response: params = await _read_webhook_params(request) verify_plivo_webhook(request, params, "masking", db) xml = telephony_service.handle_masking_webhook(params, db) return Response(content=xml, media_type="application/xml") + + +@plivo_webhook_router.post("/webhooks/answer") +async def plivo_answer_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_answer_webhook(request, db) + + +@plivo_webhook_router.post("/webhooks/events") +async def plivo_events_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_events_webhook(request, db) + + +@plivo_webhook_router.post("/webhooks/masking") +async def plivo_masking_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_masking_webhook(request, db) + + +@router.post("/webhooks/answer") +async def telephony_answer_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_answer_webhook(request, db) + + +@router.post("/webhooks/events") +async def telephony_events_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_events_webhook(request, db) + + +@router.post("/webhooks/masking") +async def telephony_masking_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_masking_webhook(request, db) diff --git a/app/api/v1/routes/vobiz_telephony.py b/app/api/v1/routes/vobiz_telephony.py index a25d7167..f78d669b 100644 --- a/app/api/v1/routes/vobiz_telephony.py +++ b/app/api/v1/routes/vobiz_telephony.py @@ -19,7 +19,7 @@ from app.services.telephony.phone_routing import resolve_inbound_agent_for_number from app.services.telephony.plivo_client import normalize_e164 from app.services.telephony.vobiz_agent_context import ( - build_vobiz_ws_url, + build_carrier_ws_url, extract_webhook_params, resolve_vobiz_agent_context, vobiz_webhook_base_url, @@ -49,15 +49,19 @@ from app.services.telephony.vobiz_xml import reject_call, speak_and_hangup, stream_to_agent from app.services.voice_agent.bot_fast_api import run_bot from app.services.voice_agent.voice_bundle import run_voice_bundle_fastapi +from app.services.telephony.carrier_media_serializer import ( + build_carrier_frame_serializer, + telephony_integration_id_from_call_row, +) from efficientai.runner.utils import parse_telephony_websocket -from efficientai.serializers.vobiz import VobizFrameSerializer # Exposed at module scope so tests can patch `.delay` without importing Celery tasks. initiate_vobiz_outbound_call_task = None router = APIRouter(prefix="/telephony/vobiz", tags=["Vobiz Telephony"]) webhook_router = APIRouter(prefix="/telephony/vobiz", tags=["Vobiz Telephony Webhooks"]) -ws_router = APIRouter(prefix="/telephony/vobiz", tags=["Vobiz Telephony Media"]) +carrier_ws_router = APIRouter(prefix="/telephony/carrier", tags=["Carrier Telephony Media"]) +ws_router = carrier_ws_router class VobizOutboundCallRequest(BaseModel): @@ -164,7 +168,9 @@ def _resolve_agent_for_answer( if session and session.agent_id and session.organization_id: return UUID(session.agent_id), UUID(session.organization_id), call_ref - agent_id, organization_id = resolve_inbound_agent_for_number(db, params.get("to")) + agent_id, organization_id, _telephony_integration_id = resolve_inbound_agent_for_number( + db, params.get("to") + ) if not agent_id or not organization_id: return None, None, None return agent_id, organization_id, None @@ -478,7 +484,7 @@ async def vobiz_answer_webhook( if session_token and call_uuid: link_provider_call_id(db, call_ref=session_token, provider_call_id=call_uuid) - ws_url = build_vobiz_ws_url( + ws_url = build_carrier_ws_url( agent_id=str(agent_id), session=session_token, persona_id=persona_id, @@ -612,8 +618,8 @@ async def vobiz_recording_ready_webhook( return {"status": "ok"} -@ws_router.websocket("/ws") -async def vobiz_media_websocket(websocket: WebSocket): +@carrier_ws_router.websocket("/ws") +async def carrier_media_websocket(websocket: WebSocket): agent_id = websocket.query_params.get("agent_id") session_token = websocket.query_params.get("session") persona_id = websocket.query_params.get("persona_id") @@ -655,7 +661,7 @@ async def vobiz_media_websocket(websocket: WebSocket): # endregion if not call_short_id: logger.warning( - "No CallRecording for Vobiz session {}; live transcript and recording will not be linked", + "No CallRecording for carrier media session {}; live transcript and recording will not be linked", session_token, ) try: @@ -679,25 +685,13 @@ async def vobiz_media_websocket(websocket: WebSocket): persona_id=persona_id, scenario_id=scenario_id, ) - from app.services.telephony.vobiz_agent_context import resolve_vobiz_telephony_run_params - - run_params = resolve_vobiz_telephony_run_params( - db, - context=context, - call_direction=session.direction, - persona_id=persona_id, - scenario_id=scenario_id, - evaluator_id=session.evaluator_id, - ) - serializer = VobizFrameSerializer( + serializer = build_carrier_frame_serializer( + provider_platform=getattr(call_row, "provider_platform", None), stream_id=stream_id, call_id=call_id, - auth_id=settings.VOBIZ_AUTH_ID, - auth_token=settings.VOBIZ_AUTH_TOKEN, - params=VobizFrameSerializer.InputParams( - sample_rate=8000, - api_base=settings.VOBIZ_API_BASE, - ), + organization_id=UUID(session.organization_id), + db=db, + telephony_integration_id=telephony_integration_id_from_call_row(call_row), ) if context.use_voice_bundle_pipeline: @@ -717,6 +711,8 @@ async def vobiz_media_websocket(websocket: WebSocket): stt_api_key=context.stt_api_key, tts_api_key=context.tts_api_key, llm_api_key=context.llm_api_key, + llm_endpoint_url=context.llm_endpoint_url, + llm_base_url=context.llm_base_url, serializer=serializer, telephony_mode=True, call_short_id=call_short_id, @@ -751,12 +747,12 @@ async def vobiz_media_websocket(websocket: WebSocket): persona_speaks_via_tts=run_params.persona_speaks_via_tts, ) except ValueError as e: - logger.error("Vobiz media websocket setup failed: {}", e) + logger.error("Carrier media websocket setup failed: {}", e) await websocket.close(code=1011, reason=str(e)) except WebSocketDisconnect: - logger.info("Vobiz media websocket disconnected") + logger.info("Carrier media websocket disconnected") except Exception as e: - logger.error("Vobiz media websocket error: {}", e, exc_info=True) + logger.error("Carrier media websocket error: {}", e, exc_info=True) try: await websocket.close(code=1011, reason="Server error") except Exception: diff --git a/app/api/v1/routes/voice_agent.py b/app/api/v1/routes/voice_agent.py index 0dadd3da..512a5701 100644 --- a/app/api/v1/routes/voice_agent.py +++ b/app/api/v1/routes/voice_agent.py @@ -512,7 +512,7 @@ def resolve_azure_endpoint_for_provider(provider: ModelProvider) -> str | None: llm_base_url = ( resolve_voice_llm_base_url(db, organization_id, voice_bundle, llm_provider) - if llm_provider + if llm_provider and voice_bundle else None ) diff --git a/app/app_factory.py b/app/app_factory.py index c2af3e09..4a83d2a2 100644 --- a/app/app_factory.py +++ b/app/app_factory.py @@ -215,7 +215,7 @@ def create_app() -> FastAPI: ) if _service_mode() == "media": logger.info( - "Vobiz telephony edge: /telephony/vobiz/webhooks/* and /telephony/vobiz/ws" + "Carrier telephony edge: /telephony/vobiz/webhooks/* and /telephony/carrier/ws" ) if settings.SERVICE_MODE == "api": logger.info( diff --git a/app/services/telephony/call_recording_lifecycle.py b/app/services/telephony/call_recording_lifecycle.py index 0868dab3..ee6c150c 100644 --- a/app/services/telephony/call_recording_lifecycle.py +++ b/app/services/telephony/call_recording_lifecycle.py @@ -9,6 +9,7 @@ from typing import Any, Dict, Optional from uuid import UUID +from sqlalchemy import func, or_ from sqlalchemy.orm import Session from sqlalchemy.orm.attributes import flag_modified @@ -180,19 +181,27 @@ def _normalize_call_event(status: Optional[str]) -> str: return normalized +def _dialect_name(db: Session) -> str: + bind = db.get_bind() if hasattr(db, "get_bind") else getattr(db, "bind", None) + if bind is None: + return "postgresql" + return bind.dialect.name + + +def _call_data_text(db: Session, key: str): + """Return JSON text for call_data[key] on Postgres and SQLite.""" + if _dialect_name(db) == "sqlite": + return func.json_extract(CallRecording.call_data, f"$.{key}") + return func.json_extract_path_text(CallRecording.call_data, key) + + def _find_by_call_ref(db: Session, call_ref: str) -> Optional[CallRecording]: - rows = ( + return ( db.query(CallRecording) - .filter(CallRecording.provider_platform == "vobiz") + .filter(_call_data_text(db, "call_ref") == call_ref) .order_by(CallRecording.created_at.desc()) - .limit(200) - .all() + .first() ) - for row in rows: - data = row.call_data if isinstance(row.call_data, dict) else {} - if data.get("call_ref") == call_ref: - return row - return None def find_call_recording( @@ -209,18 +218,20 @@ def find_call_recording( ) if row: return row - rows = ( + json_id_match = or_( + *( + _call_data_text(db, key) == provider_call_id + for key in ("request_uuid", "message_uuid", "api_id", "call_uuid") + ) + ) + row = ( db.query(CallRecording) - .filter(CallRecording.provider_platform == "vobiz") + .filter(json_id_match) .order_by(CallRecording.created_at.desc()) - .limit(200) - .all() + .first() ) - for candidate in rows: - data = candidate.call_data if isinstance(candidate.call_data, dict) else {} - for key in ("request_uuid", "message_uuid", "api_id", "call_uuid"): - if data.get(key) == provider_call_id: - return candidate + if row: + return row if call_ref: return _find_by_call_ref(db, call_ref) return None @@ -237,6 +248,8 @@ def create_inbound_call_recording( provider_call_id: Optional[str] = None, evaluator_id: Optional[UUID] = None, evaluator_result_id: Optional[UUID] = None, + provider_platform: str = "vobiz", + telephony_integration_id: Optional[UUID] = None, ) -> CallRecording: existing = find_call_recording(db, call_ref=call_ref, provider_call_id=provider_call_id) if existing: @@ -254,6 +267,8 @@ def create_inbound_call_recording( } if evaluator_id is not None: call_data["evaluator_id"] = str(evaluator_id) + if telephony_integration_id is not None: + call_data["telephony_integration_id"] = str(telephony_integration_id) row = CallRecording( organization_id=organization_id, workspace_id=agent.workspace_id, @@ -263,7 +278,7 @@ def create_inbound_call_recording( call_event="call_started", call_data=call_data, provider_call_id=provider_call_id, - provider_platform="vobiz", + provider_platform=provider_platform, agent_id=agent.id, evaluator_result_id=evaluator_result_id, ) diff --git a/app/services/telephony/carrier_media_serializer.py b/app/services/telephony/carrier_media_serializer.py new file mode 100644 index 00000000..3585f36f --- /dev/null +++ b/app/services/telephony/carrier_media_serializer.py @@ -0,0 +1,84 @@ +"""Choose the carrier media serializer and hangup credentials.""" + +from __future__ import annotations + +from typing import Optional +from uuid import UUID + +from sqlalchemy.orm import Session + +from app.config import settings +from app.core.encryption import decrypt_api_key +from app.services.credentials.resolver import resolve_telephony_integration +from efficientai.serializers.plivo import PlivoFrameSerializer +from efficientai.serializers.vobiz import VobizFrameSerializer + + +def _plivo_call_control_credentials( + db: Session, + organization_id: UUID, + *, + telephony_integration_id: Optional[UUID] = None, +) -> tuple[str, str]: + integration = resolve_telephony_integration( + "plivo", + db, + organization_id, + credential_id=telephony_integration_id, + ) + if integration: + return ( + decrypt_api_key(integration.auth_id).strip(), + decrypt_api_key(integration.auth_token).strip(), + ) + return (settings.PLIVO_AUTH_ID or "").strip(), (settings.PLIVO_AUTH_TOKEN or "").strip() + + +def telephony_integration_id_from_call_row(call_row) -> Optional[UUID]: + if call_row is None: + return None + data = call_row.call_data if isinstance(call_row.call_data, dict) else {} + raw = data.get("telephony_integration_id") + if not raw: + return None + try: + return UUID(str(raw)) + except (TypeError, ValueError): + return None + + +def build_carrier_frame_serializer( + *, + provider_platform: Optional[str], + stream_id: str, + call_id: Optional[str], + organization_id: UUID, + db: Session, + telephony_integration_id: Optional[UUID] = None, +): + """Return the media serializer whose hangup API matches the live carrier.""" + platform = (provider_platform or "").strip().lower() + if platform == "plivo": + auth_id, auth_token = _plivo_call_control_credentials( + db, + organization_id, + telephony_integration_id=telephony_integration_id, + ) + return PlivoFrameSerializer( + stream_id=stream_id, + call_id=call_id, + auth_id=auth_id, + auth_token=auth_token, + params=PlivoFrameSerializer.InputParams(sample_rate=8000), + ) + + return VobizFrameSerializer( + stream_id=stream_id, + call_id=call_id, + auth_id=settings.VOBIZ_AUTH_ID, + auth_token=settings.VOBIZ_AUTH_TOKEN, + params=VobizFrameSerializer.InputParams( + sample_rate=8000, + api_base=settings.VOBIZ_API_BASE, + ), + ) diff --git a/app/services/telephony/inbound_stream_answer.py b/app/services/telephony/inbound_stream_answer.py new file mode 100644 index 00000000..8f674b51 --- /dev/null +++ b/app/services/telephony/inbound_stream_answer.py @@ -0,0 +1,137 @@ +"""Build WebSocket stream XML for inbound carrier answer webhooks.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional +from uuid import UUID + +from loguru import logger +from sqlalchemy.orm import Session + +from app.config import settings +from app.models.database import Agent +from app.services.telephony.call_recording_lifecycle import ( + create_inbound_call_recording, + link_provider_call_id, +) +from app.services.telephony.phone_routing import resolve_inbound_agent_for_number +from app.services.telephony.plivo_xml import reject_call, speak_and_hangup +from app.services.telephony.vobiz_agent_context import build_carrier_ws_url, vobiz_webhook_base_url +from app.services.telephony.vobiz_session import create_call_session +from app.services.telephony.vobiz_xml import stream_to_agent + + +def build_inbound_stream_answer_xml( + db: Session, + params: Dict[str, Any], + *, + provider_platform: str = "plivo", +) -> str: + """Return Plivo/Vobiz-compatible XML that streams inbound audio to the voice agent.""" + to_number = params.get("To") or params.get("to") + from_number = params.get("From") or params.get("from") + call_uuid = ( + params.get("CallUUID") + or params.get("CallSid") + or params.get("call_sid") + or params.get("Sid") + ) + + if not to_number: + return speak_and_hangup("Call could not be routed.") + + agent_id, organization_id, telephony_integration_id = resolve_inbound_agent_for_number( + db, to_number + ) + if not agent_id or not organization_id: + logger.warning( + "Inbound stream answer miss: to={} from={} call_uuid={}", + to_number, + from_number, + call_uuid, + ) + return reject_call("No active routing found for this number.") + + agent = db.query(Agent).filter(Agent.id == agent_id).first() + if not agent: + return reject_call("No active routing found for this number.") + + inbound_evaluator_id: Optional[UUID] = None + inbound_evaluator_result_id: Optional[UUID] = None + inbound_persona_id: Optional[UUID] = None + inbound_scenario_id: Optional[UUID] = None + if agent.workspace_id: + from app.services.evaluators.evaluator_inbound_service import ( + consume_inbound_evaluator_combination, + create_inbound_evaluator_result, + find_inbound_suite_for_agent, + ) + + suite = find_inbound_suite_for_agent( + db, agent, organization_id, agent.workspace_id + ) + if suite: + selected, _idx, _next_idx = consume_inbound_evaluator_combination(db, suite) + inbound_evaluator_id = selected.id + inbound_persona_id = selected.persona_id + inbound_scenario_id = selected.scenario_id + result_row = create_inbound_evaluator_result( + db, + organization_id, + agent.workspace_id, + selected, + ) + inbound_evaluator_result_id = result_row.id + + session = create_call_session( + agent_id=str(agent_id), + organization_id=str(organization_id), + direction="inbound", + from_number=from_number, + to_number=to_number, + persona_id=str(inbound_persona_id) if inbound_persona_id else None, + scenario_id=str(inbound_scenario_id) if inbound_scenario_id else None, + evaluator_id=str(inbound_evaluator_id) if inbound_evaluator_id else None, + ) + + create_inbound_call_recording( + db, + agent=agent, + organization_id=organization_id, + call_ref=session.call_ref, + from_number=from_number, + to_number=to_number, + provider_call_id=call_uuid, + evaluator_id=inbound_evaluator_id, + evaluator_result_id=inbound_evaluator_result_id, + provider_platform=provider_platform, + telephony_integration_id=telephony_integration_id, + ) + + if call_uuid: + link_provider_call_id(db, call_ref=session.call_ref, provider_call_id=call_uuid) + + ws_url = build_carrier_ws_url( + agent_id=str(agent_id), + session=session.call_ref, + persona_id=str(inbound_persona_id) if inbound_persona_id else None, + scenario_id=str(inbound_scenario_id) if inbound_scenario_id else None, + ) + + record_action_url = None + if settings.VOBIZ_CARRIER_SESSION_RECORDING and provider_platform == "vobiz": + record_action_url = ( + f"{vobiz_webhook_base_url()}{settings.API_V1_PREFIX}/telephony/vobiz/webhooks/recording-ready" + f"?call_ref={session.call_ref}" + ) + + logger.info( + "Inbound stream answer agent_id={} session={} to={} from={} call_uuid={}", + agent_id, + session.call_ref, + to_number, + from_number, + call_uuid, + ) + + return stream_to_agent(ws_url, record_action_url=record_action_url) diff --git a/app/services/telephony/number_import_service.py b/app/services/telephony/number_import_service.py index a5389527..7711af3a 100644 --- a/app/services/telephony/number_import_service.py +++ b/app/services/telephony/number_import_service.py @@ -13,6 +13,11 @@ from app.models.enums import TelephonyProvider from app.services.telephony.phone_routing import sync_agent_telephony_number_link from app.services.telephony.plivo_client import PlivoClient, expand_phone_candidates, normalize_e164 +from app.services.telephony.plivo_webhook_urls import ( + legacy_answer_webhook_url, + plivo_answer_webhook_url, + plivo_hangup_webhook_url, +) from app.services.telephony.telephony_service import telephony_service from app.services.telephony.vobiz_agent_context import vobiz_webhook_base_url from app.services.telephony.vobiz_client import VobizClient, build_vobiz_client_for_org @@ -42,21 +47,15 @@ def _vobiz_answer_webhook_url() -> str: def _plivo_answer_webhook_url() -> str: - base = (settings.PLIVO_WEBHOOK_BASE_URL or "").strip().rstrip("/") - if not base: - raise ValueError( - "Plivo webhook base URL is not configured. Set plivo.webhook_base_url in platform config." - ) - return f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/answer" + return plivo_answer_webhook_url() def _plivo_hangup_webhook_url() -> str: - base = (settings.PLIVO_WEBHOOK_BASE_URL or "").strip().rstrip("/") - return f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/events" + return plivo_hangup_webhook_url() def _exotel_voice_webhook_url() -> str: - return _plivo_answer_webhook_url() + return legacy_answer_webhook_url() def _normalize_remote_number(provider: str, item: Dict[str, Any]) -> Optional[str]: @@ -161,27 +160,111 @@ def _list_remote_numbers( raise ValueError(f"Unsupported provider: {provider}") -def _remote_metadata(provider: str, item: Dict[str, Any]) -> Dict[str, Any]: +def _normalize_country_iso2( + raw: Optional[str], + *, + e164: Optional[str] = None, +) -> Optional[str]: + """Normalize provider country values to ISO-3166 alpha-2 for storage.""" + value = (raw or "").strip() + if not value: + return _country_iso2_from_e164(e164) + + if len(value) == 2 and value.isalpha(): + return value.upper() + + mapped = _COUNTRY_NAME_TO_ISO2.get(value.lower()) + if mapped: + return mapped + + return _country_iso2_from_e164(e164) + + +def _country_iso2_from_e164(e164: Optional[str]) -> Optional[str]: + if not e164: + return None + digits = str(e164).strip().lstrip("+") + if not digits.isdigit(): + return None + for length in (3, 2, 1): + prefix = digits[:length] + iso2 = _CALLING_CODE_TO_ISO2.get(prefix) + if iso2: + return iso2 + return None + + +def _extract_application_id(value: Optional[str]) -> Optional[str]: + """Extract a Plivo-style application id from a URI or raw id.""" + if not value: + return None + text = str(value).strip().rstrip("/") + if "/Application/" in text: + tail = text.rsplit("/Application/", 1)[-1] + app_id = tail.split("/", 1)[0].strip() + return app_id or None + return text or None + + +_COUNTRY_NAME_TO_ISO2 = { + "india": "IN", + "united states": "US", + "united states of america": "US", + "usa": "US", + "united kingdom": "GB", + "uk": "GB", + "canada": "CA", + "australia": "AU", + "germany": "DE", + "france": "FR", + "singapore": "SG", +} + +_CALLING_CODE_TO_ISO2 = { + "91": "IN", + "1": "US", + "44": "GB", + "61": "AU", + "49": "DE", + "33": "FR", + "65": "SG", + "971": "AE", +} + + +def _remote_metadata( + provider: str, + item: Dict[str, Any], + *, + e164: Optional[str] = None, +) -> Dict[str, Any]: provider_key = provider.lower() if provider_key == TelephonyProvider.EXOTEL.value: + raw_country = item.get("Country") or item.get("country") return { "provider_number_id": item.get("Sid") or item.get("sid"), - "country": item.get("Country") or item.get("country"), + "country": raw_country, + "country_iso2": _normalize_country_iso2(raw_country, e164=e164), "region": item.get("Region") or item.get("region") or item.get("FriendlyName"), "capabilities": None, "status": item.get("Status") or item.get("status"), "application_id": item.get("VoiceUrl") or item.get("voice_url"), } + raw_country = item.get("country") or item.get("Country") + application_id = _extract_application_id( + item.get("application_id") + or item.get("app_id") + or item.get("Application") + or item.get("application") + ) return { "provider_number_id": item.get("id") or item.get("Sid") or item.get("sid"), - "country": item.get("country") or item.get("Country"), + "country": raw_country, + "country_iso2": _normalize_country_iso2(raw_country, e164=e164), "region": item.get("region") or item.get("Region"), "capabilities": item.get("capabilities"), "status": item.get("status") or item.get("Status"), - "application_id": item.get("application_id") - or item.get("app_id") - or item.get("Application") - or item.get("application"), + "application_id": application_id, } @@ -250,7 +333,7 @@ def list_available_numbers( if not e164: continue imported = imported_by_phone.get(e164) - meta = _remote_metadata(provider_key, item) + meta = _remote_metadata(provider_key, item, e164=e164) results.append( { "e164": e164, @@ -295,7 +378,7 @@ def import_numbers( e164 = _normalize_remote_number(provider_key, item) if e164: enriched = dict(item) - enriched.update(_remote_metadata(provider_key, item)) + enriched.update(_remote_metadata(provider_key, item, e164=e164)) remote_by_e164[e164] = enriched agent: Optional[Agent] = None @@ -374,7 +457,7 @@ def import_numbers( row.inbound_enabled = True row.outbound_enabled = True row.source = "imported" - row.country_iso2 = remote.get("country") or row.country_iso2 + row.country_iso2 = remote.get("country_iso2") or row.country_iso2 row.region = remote.get("region") or row.region row.capabilities = remote.get("capabilities") or row.capabilities row.provider_app_id = remote.get("application_id") or row.provider_app_id @@ -385,7 +468,7 @@ def import_numbers( organization_id=org_id, telephony_integration_id=integration_id, phone_number=e164, - country_iso2=remote.get("country"), + country_iso2=remote.get("country_iso2"), region=remote.get("region"), capabilities=remote.get("capabilities"), provider_app_id=remote.get("application_id"), diff --git a/app/services/telephony/phone_routing.py b/app/services/telephony/phone_routing.py index b84446ca..8b011899 100644 --- a/app/services/telephony/phone_routing.py +++ b/app/services/telephony/phone_routing.py @@ -195,11 +195,11 @@ def _resolve_agent_for_org( def resolve_inbound_agent_for_number( db: Session, to_number_raw: Optional[str], -) -> Tuple[Optional[UUID], Optional[UUID]]: - """Resolve (agent_id, organization_id) for an inbound called number.""" +) -> Tuple[Optional[UUID], Optional[UUID], Optional[UUID]]: + """Resolve (agent_id, organization_id, telephony_integration_id) for inbound To.""" candidates = expand_phone_candidates(to_number_raw) if not candidates: - return None, None + return None, None, None number_row = _find_inbound_number_row(db, candidates) if not number_row: @@ -208,7 +208,7 @@ def resolve_inbound_agent_for_number( to_number_raw, candidates, ) - return None, None + return None, None, None agent_id, organization_id = _resolve_agent_for_org( db, @@ -217,7 +217,7 @@ def resolve_inbound_agent_for_number( number_row, ) if agent_id and organization_id: - return agent_id, organization_id + return agent_id, organization_id, number_row.telephony_integration_id logger.warning( "Inbound number {} owned by org {} but no agent linked (candidates={})", @@ -225,7 +225,7 @@ def resolve_inbound_agent_for_number( number_row.organization_id, candidates, ) - return None, None + return None, None, None def sync_agent_telephony_number_link(db: Session, agent: Agent) -> None: diff --git a/app/services/telephony/plivo_client.py b/app/services/telephony/plivo_client.py index 8e750c18..499a2ee1 100644 --- a/app/services/telephony/plivo_client.py +++ b/app/services/telephony/plivo_client.py @@ -1,6 +1,6 @@ """Thin Plivo SDK wrapper for telephony operations.""" -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from loguru import logger try: @@ -85,12 +85,21 @@ def _add(value: Optional[str]) -> None: class PlivoClient: """Wrapper around plivo.RestClient that returns normalized dictionaries.""" - def __init__(self, auth_id: str, auth_token: str): + def __init__( + self, + auth_id: str, + auth_token: str, + *, + credential_fingerprint: Optional[str] = None, + ): if plivo is None: raise ValueError( "Plivo SDK is not installed. Install it with `pip install -e .` or `pip install plivo`." ) self.client = plivo.RestClient(auth_id=auth_id, auth_token=auth_token) + self._auth_id = auth_id + self._auth_token = auth_token + self._credential_fingerprint = credential_fingerprint @staticmethod def _to_dict(data: Any) -> Dict[str, Any]: @@ -211,6 +220,16 @@ def get_call_details(self, call_uuid: str) -> Dict[str, Any]: logger.exception("Failed to fetch call details") raise ValueError(f"Failed to fetch call details: {str(e)}") + def download_recording(self, recording_url: str) -> Tuple[bytes, str]: + """Download a recording from Plivo with HTTP Basic auth.""" + from app.services.telephony.recording_download import download_recording_url + + return download_recording_url( + recording_url, + auth=(self._auth_id, self._auth_token), + credential_fingerprint=self._credential_fingerprint, + ) + def start_voice_verification( self, recipient: str, app_uuid: str, callback_url: Optional[str] = None ) -> Dict[str, Any]: diff --git a/app/services/telephony/plivo_webhook_urls.py b/app/services/telephony/plivo_webhook_urls.py new file mode 100644 index 00000000..97c6d89d --- /dev/null +++ b/app/services/telephony/plivo_webhook_urls.py @@ -0,0 +1,35 @@ +"""Public HTTPS callback URLs for native Plivo telephony webhooks.""" + +from __future__ import annotations + +from app.config import settings + + +def plivo_webhook_base() -> str: + base = (settings.PLIVO_WEBHOOK_BASE_URL or "").strip().rstrip("/") + if not base: + raise ValueError( + "Plivo webhook base URL is not configured. Set plivo.webhook_base_url in platform config." + ) + return base + + +def plivo_answer_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/plivo/webhooks/answer" + + +def plivo_hangup_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/plivo/webhooks/events" + + +def plivo_masking_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/plivo/webhooks/masking" + + +def legacy_answer_webhook_url() -> str: + """Legacy generic path kept for Exotel and already-imported Plivo numbers.""" + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/webhooks/answer" + + +def legacy_events_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/webhooks/events" diff --git a/app/services/telephony/plivo_xml.py b/app/services/telephony/plivo_xml.py index deb01b7c..9fa44ae8 100644 --- a/app/services/telephony/plivo_xml.py +++ b/app/services/telephony/plivo_xml.py @@ -23,7 +23,7 @@ def dial_number(to_number: str, caller_id: str) -> str: """Build XML to dial a target number with caller ID.""" plivoxml = _get_plivoxml() response = plivoxml.ResponseElement() - dial = plivoxml.DialElement(callerId=caller_id) + dial = plivoxml.DialElement(caller_id=caller_id) dial.add(plivoxml.NumberElement(to_number)) response.add(dial) return response.to_string() diff --git a/app/services/telephony/recording_download.py b/app/services/telephony/recording_download.py index e4d40ec8..be4e2107 100644 --- a/app/services/telephony/recording_download.py +++ b/app/services/telephony/recording_download.py @@ -29,6 +29,13 @@ "cloudfront.net", ) +# Credentialed telephony fetches must not send Basic auth to shared storage hosts. +_CREDENTIALED_ALLOWED_HOST_SUFFIXES = ( + "exotel.com", + "plivo.com", + "vobiz.ai", +) + _BLOCKED_NETWORKS = ( ipaddress.ip_network("0.0.0.0/8"), ipaddress.ip_network("10.0.0.0/8"), @@ -107,7 +114,12 @@ def assert_recording_url_safe( user_supplied: bool, allowed_suffixes: Optional[List[str]] = None, ) -> None: - """Validate a recording URL before any outbound HTTP request.""" + """Validate a recording URL before any outbound HTTP request. + + Literal IP hosts are always rejected so credentialed fetches cannot send + Basic auth to a CSV-controlled address. ``user_supplied`` is kept for + call-site compatibility; redirect policy is enforced by the downloader. + """ parsed = urlparse(recording_url.strip()) if parsed.scheme not in {"http", "https"}: raise ExotelInvalidContentError( @@ -125,24 +137,27 @@ def assert_recording_url_safe( try: literal_ip = ipaddress.ip_address(hostname) + except ValueError: + literal_ip = None + + if literal_ip is not None: if _ip_is_blocked(literal_ip): raise ExotelInvalidContentError( "Recording URL targets a blocked network address" ) - if user_supplied: - raise ExotelInvalidContentError( - "User-supplied recording URLs must use allowlisted hostnames" - ) - except ValueError: - if not _hostname_allowed(hostname, suffixes): + raise ExotelInvalidContentError( + "Recording URLs must use allowlisted hostnames, not IP addresses" + ) + + if not _hostname_allowed(hostname, suffixes): + raise ExotelInvalidContentError( + f"Recording URL hostname is not allowlisted: {hostname}" + ) + for resolved_ip in _resolve_host_ips(hostname): + if _ip_is_blocked(resolved_ip): raise ExotelInvalidContentError( - f"Recording URL hostname is not allowlisted: {hostname}" + "Recording URL resolves to a blocked network address" ) - for resolved_ip in _resolve_host_ips(hostname): - if _ip_is_blocked(resolved_ip): - raise ExotelInvalidContentError( - "Recording URL resolves to a blocked network address" - ) def download_recording_url( @@ -166,13 +181,26 @@ def download_recording_url( "User-supplied recording URLs must not be fetched with credentials" ) - assert_recording_url_safe(recording_url, user_supplied=user_supplied) + if auth is not None: + allowed_suffixes = list(_CREDENTIALED_ALLOWED_HOST_SUFFIXES) + else: + allowed_suffixes = _allowed_host_suffixes() + + assert_recording_url_safe( + recording_url, + user_supplied=user_supplied, + allowed_suffixes=allowed_suffixes, + ) request_hooks: Optional[dict[str, List[Callable[..., None]]]] = None if not user_supplied: def _validate_redirect(request: httpx.Request) -> None: - assert_recording_url_safe(str(request.url), user_supplied=False) + assert_recording_url_safe( + str(request.url), + user_supplied=False, + allowed_suffixes=allowed_suffixes, + ) request_hooks = {"request": [_validate_redirect]} diff --git a/app/services/telephony/telephony_service.py b/app/services/telephony/telephony_service.py index 6374ab49..ef358c58 100644 --- a/app/services/telephony/telephony_service.py +++ b/app/services/telephony/telephony_service.py @@ -27,7 +27,13 @@ from app.services.credentials.resolver import clear_other_defaults from app.services.telephony.exotel_client import build_exotel_client_from_integration from app.services.telephony.plivo_client import PlivoClient, normalize_e164 -from app.services.telephony.plivo_xml import dial_number, reject_call, speak_and_hangup +from app.services.telephony.plivo_webhook_urls import ( + legacy_events_webhook_url, + plivo_answer_webhook_url, + plivo_hangup_webhook_url, +) +from app.services.telephony.inbound_stream_answer import build_inbound_stream_answer_xml +from app.services.telephony.plivo_xml import reject_call from app.services.telephony.vobiz_client import build_vobiz_client_for_org from app.services.telephony.vobiz_xml import ( speak_and_hangup as vobiz_speak_and_hangup, @@ -88,7 +94,15 @@ def get_provider_client( auth_token = decrypt_api_key(integration.auth_token) provider_key = (integration.provider or provider or "").lower() if provider_key == "plivo": - return PlivoClient(auth_id=auth_id, auth_token=auth_token) + from app.workers.concurrency.telephony_credential_rate_limit import ( + fingerprint_for_integration, + ) + + return PlivoClient( + auth_id=auth_id, + auth_token=auth_token, + credential_fingerprint=fingerprint_for_integration(integration), + ) if provider_key == "exotel": from app.workers.concurrency.telephony_credential_rate_limit import ( fingerprint_for_integration, @@ -492,9 +506,8 @@ def initiate_outbound_call( raise ValueError("No active telephony integration found for from_number") client = self.get_provider_client(org_id, db, provider=integration.provider) - base = settings.PLIVO_WEBHOOK_BASE_URL.rstrip("/") - answer_url = f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/answer" - hangup_url = f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/events" + answer_url = plivo_answer_webhook_url() + hangup_url = plivo_hangup_webhook_url() response = client.create_outbound_call( from_=from_number, to_=to_number, answer_url=answer_url, hangup_url=hangup_url ) @@ -522,6 +535,8 @@ def initiate_outbound_call( raise ValueError("No default workspace found for organization") workspace_id = default_workspace.id + call_data = dict(response or {}) + call_data["telephony_integration_id"] = str(integration.id) db.add( CallRecording( organization_id=org_id, @@ -530,7 +545,7 @@ def initiate_outbound_call( status=CallRecordingStatus.PENDING, source=CallRecordingSource.WEBHOOK, call_event="outbound_initiated", - call_data=response, + call_data=call_data, provider_call_id=call_uuid, provider_platform=integration.provider, agent_id=agent_id, @@ -551,8 +566,7 @@ def start_voice_otp( recipient = normalize_e164(phone_number) callback_url = None if settings.PLIVO_WEBHOOK_BASE_URL: - base = settings.PLIVO_WEBHOOK_BASE_URL.rstrip("/") - callback_url = f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/events" + callback_url = legacy_events_webhook_url() response = self.get_provider_client(org_id, db, provider=provider).start_voice_verification( recipient=recipient, app_uuid=app_uuid, callback_url=callback_url @@ -700,27 +714,17 @@ def handle_answer_webhook(self, params: Dict[str, Any], db: Session) -> str: or params.get("call_sid") or params.get("Sid") ) - logger.info("Telephony answer webhook call_uuid={} to={} from={}", call_uuid, to_number, from_number) - - if not to_number: - return speak_and_hangup("Call could not be routed.") - - to_number = normalize_e164(to_number) - number = db.query(TelephonyPhoneNumber).filter(TelephonyPhoneNumber.phone_number == to_number).first() - if not number: - return reject_call("This number is not configured.") - - if number.agent_id: - agent = db.query(Agent).filter(Agent.id == number.agent_id).first() - if agent and agent.phone_number: - try: - return dial_number(normalize_e164(agent.phone_number), to_number) - except ValueError: - logger.warning("Agent {} has non-E.164 phone number", agent.id) - - return speak_and_hangup("No active routing found for this number.") + logger.info( + "Telephony answer webhook call_uuid={} to={} from={}", + call_uuid, + to_number, + from_number, + ) + return build_inbound_stream_answer_xml(db, params, provider_platform="plivo") def handle_event_webhook(self, params: Dict[str, Any], db: Session) -> None: + from app.services.telephony.call_recording_lifecycle import update_call_from_vobiz_event + call_uuid = ( params.get("CallUUID") or params.get("RequestUUID") @@ -732,6 +736,15 @@ def handle_event_webhook(self, params: Dict[str, Any], db: Session) -> None: if not call_uuid: return + updated = update_call_from_vobiz_event( + db, + provider_call_id=call_uuid, + call_status=call_status, + payload=params, + ) + if updated: + return + row = db.query(CallRecording).filter(CallRecording.provider_call_id == call_uuid).first() if not row: return @@ -771,6 +784,8 @@ def handle_masking_webhook(self, params: Dict[str, Any], db: Session) -> str: return reject_call() target = session.party_b_number if from_number == session.party_a_number else session.party_a_number + from app.services.telephony.plivo_xml import dial_number + return dial_number(target, to_number) diff --git a/app/services/telephony/vobiz_agent_context.py b/app/services/telephony/vobiz_agent_context.py index 758ba28a..316285c4 100644 --- a/app/services/telephony/vobiz_agent_context.py +++ b/app/services/telephony/vobiz_agent_context.py @@ -37,6 +37,60 @@ class VobizAgentContext: stt_api_key: Optional[str] tts_api_key: Optional[str] llm_api_key: Optional[str] + llm_endpoint_url: Optional[str] = None + llm_base_url: Optional[str] = None + + +def _resolve_azure_endpoint_for_provider( + db: Session, + organization_id: UUID, + provider: ModelProvider, +) -> Optional[str]: + from app.services.ai.llm_service import _resolve_azure_endpoint_from_provider + + provider_value = provider.value if hasattr(provider, "value") else str(provider) + ai_provider_rec = db.query(AIProvider).filter( + AIProvider.organization_id == organization_id, + AIProvider.provider == provider_value, + AIProvider.is_active.is_(True), + ).first() + if not ai_provider_rec: + ai_provider_rec = db.query(AIProvider).filter( + AIProvider.organization_id == organization_id, + func.lower(AIProvider.provider) == provider_value.lower(), + AIProvider.is_active.is_(True), + ).first() + if not ai_provider_rec: + return None + return _resolve_azure_endpoint_from_provider(ai_provider_rec, None) + + +def _resolve_voice_llm_urls( + db: Session, + organization_id: UUID, + voice_bundle: Optional[VoiceBundle], +) -> tuple[Optional[str], Optional[str]]: + if not voice_bundle or not voice_bundle.llm_provider: + return None, None + + llm_provider = voice_bundle.llm_provider + provider_key = ( + llm_provider.value if hasattr(llm_provider, "value") else str(llm_provider) + ).lower() + llm_endpoint_url = ( + _resolve_azure_endpoint_for_provider(db, organization_id, llm_provider) + if provider_key == "azure" + else None + ) + from app.services.voice_agent.llm_voice_providers import resolve_voice_llm_base_url + + llm_base_url = resolve_voice_llm_base_url( + db, + organization_id, + voice_bundle, + llm_provider, + ) + return llm_endpoint_url, llm_base_url @dataclass @@ -220,6 +274,8 @@ def resolve_vobiz_agent_context( stt_api_key = None tts_api_key = None llm_api_key = None + llm_endpoint_url = None + llm_base_url = None if use_voice_bundle_pipeline and voice_bundle: if voice_bundle.stt_provider: @@ -228,6 +284,11 @@ def resolve_vobiz_agent_context( tts_api_key = _resolve_api_key_for_provider(db, organization_id, voice_bundle.tts_provider) if voice_bundle.llm_provider: llm_api_key = _resolve_api_key_for_provider(db, organization_id, voice_bundle.llm_provider) + llm_endpoint_url, llm_base_url = _resolve_voice_llm_urls( + db, + organization_id, + voice_bundle, + ) else: ai_provider = None if agent.ai_provider_id: @@ -270,6 +331,8 @@ def resolve_vobiz_agent_context( stt_api_key=stt_api_key, tts_api_key=tts_api_key, llm_api_key=llm_api_key, + llm_endpoint_url=llm_endpoint_url, + llm_base_url=llm_base_url, ) @@ -354,13 +417,14 @@ def vobiz_webhook_base_url() -> str: return base.rstrip("/") -def build_vobiz_ws_url( +def build_carrier_ws_url( *, agent_id: str, session: str, persona_id: Optional[str] = None, scenario_id: Optional[str] = None, ) -> str: + """Build the WebSocket URL for live carrier audio (Plivo, Vobiz, etc.).""" from app.config import settings from app.services.media_urls import carrier_media_ws_base_url from urllib.parse import quote @@ -374,7 +438,12 @@ def build_vobiz_ws_url( query += f"&persona_id={quote(persona_id)}" if scenario_id: query += f"&scenario_id={quote(scenario_id)}" - return f"{ws_base}{settings.API_V1_PREFIX}/telephony/vobiz/ws?{query}" + return f"{ws_base}{settings.API_V1_PREFIX}/telephony/carrier/ws?{query}" + + +def build_vobiz_ws_url(**kwargs) -> str: + """Deprecated alias for :func:`build_carrier_ws_url`.""" + return build_carrier_ws_url(**kwargs) def extract_webhook_params(payload: Dict[str, Any]) -> Dict[str, Any]: diff --git a/app/services/telephony/webhook_auth.py b/app/services/telephony/webhook_auth.py index 16a7c1ad..cc6808a8 100644 --- a/app/services/telephony/webhook_auth.py +++ b/app/services/telephony/webhook_auth.py @@ -135,6 +135,38 @@ def _auth_token_for_vobiz_org(db: Session, org_id: UUID) -> Optional[str]: return _platform_vobiz_auth_token() +def _platform_plivo_auth_token() -> Optional[str]: + token = (settings.PLIVO_AUTH_TOKEN or "").strip() + return token or None + + +def _resolve_plivo_auth_token_by_auth_id(auth_id: str, db: Session) -> Optional[str]: + candidate = (auth_id or "").strip() + if not candidate: + return None + + platform_id = (settings.PLIVO_AUTH_ID or "").strip() + if platform_id and candidate.lower() == platform_id.lower(): + return _platform_plivo_auth_token() + + rows = ( + db.query(TelephonyIntegration) + .filter( + TelephonyIntegration.provider == "plivo", + TelephonyIntegration.is_active.is_(True), + ) + .all() + ) + for row in rows: + try: + stored_id = decrypt_api_key(row.auth_id).strip() + except Exception: + continue + if stored_id.lower() == candidate.lower(): + return decrypt_api_key(row.auth_token).strip() + return None + + def _resolve_auth_token_for_call_event( params: Dict[str, Any], db: Session, @@ -181,7 +213,27 @@ def resolve_plivo_auth_token( phone_number = params.get("To") or params.get("to") return _resolve_auth_token_for_phone(phone_number, db) if webhook_kind == "events": - return _resolve_auth_token_for_call_event(params, db) + token = _resolve_auth_token_for_call_event(params, db) + if token: + return token + + parent_auth_id = ( + params.get("ParentAuthID") + or params.get("AuthID") + or params.get("auth_id") + ) + if parent_auth_id: + token = _resolve_plivo_auth_token_by_auth_id(str(parent_auth_id), db) + if token: + return token + + phone_number = ( + params.get("To") + or params.get("to") + or params.get("From") + or params.get("from") + ) + return _resolve_auth_token_for_phone(phone_number, db) return None diff --git a/app/services/voice_agent/llm_voice_providers.py b/app/services/voice_agent/llm_voice_providers.py index 59bec721..7cf8848b 100644 --- a/app/services/voice_agent/llm_voice_providers.py +++ b/app/services/voice_agent/llm_voice_providers.py @@ -1,339 +1,339 @@ -"""Live voice pipeline LLM provider registry and service factory. - -Mirrors the LLM-capable providers exposed in Voice Bundles / Integrations -(see ``app/services/judge_alignment/model_catalog.py``). -""" - -from __future__ import annotations - -import json -import os -from typing import Any, Callable, Dict, Optional - -from loguru import logger - -# Providers selectable for the LLM leg of STT+LLM+TTS voice bundles. -LLM_VOICE_PROVIDER_KEYS = frozenset( - { - "openai", - "anthropic", - "google", - "xai", - "fireworks", - "cohere", - "mistral", - "meta", - "together", - "perplexity", - "azure", - "aws", - "openrouter", - "custom", - "sarvam", - } -) - -_DEFAULT_LLM_MODELS: Dict[str, str] = { - "openai": "gpt-4.1", - "google": "gemini-2.5-flash", - "anthropic": "claude-sonnet-4.6", - "xai": "grok-3-beta", - "fireworks": "accounts/fireworks/models/llama-v3p1-8b-instruct", - "cohere": "command-r-plus-08-2024", - "mistral": "mistral-small-latest", - "meta": "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", - "together": "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", - "perplexity": "sonar", - "azure": "gpt-4.1", - "aws": "amazon.nova-lite-v1:0", - "openrouter": "openai/gpt-4o-2024-11-20", - "custom": "gpt-4o-mini", - "sarvam": "sarvam-30b", -} - -_ENV_KEYS: Dict[str, str] = { - "openai": "OPENAI_API_KEY", - "google": "GOOGLE_API_KEY", - "anthropic": "ANTHROPIC_API_KEY", - "xai": "XAI_API_KEY", - "fireworks": "FIREWORKS_API_KEY", - "cohere": "COHERE_API_KEY", - "mistral": "MISTRAL_API_KEY", - "meta": "TOGETHER_API_KEY", - "together": "TOGETHER_API_KEY", - "perplexity": "PERPLEXITY_API_KEY", - "azure": "AZURE_OPENAI_API_KEY", - "aws": "AWS_ACCESS_KEY_ID", - "openrouter": "OPENROUTER_API_KEY", - "custom": "OPENAI_API_KEY", - "sarvam": "SARVAM_API_KEY", -} - - -def normalize_llm_model(provider: str, model: str) -> str: - """Normalize catalog model ids for provider-specific APIs.""" - provider_key = (provider or "").strip().lower() - if not model: - return model - if provider_key == "fireworks" and not model.startswith("accounts/"): - return f"accounts/fireworks/models/{model}" - if provider_key == "azure": - from app.services.ai.llm_service import _azure_deployment_name - - return _azure_deployment_name(model) - return model - - -def default_llm_model(provider: str) -> str: - return _DEFAULT_LLM_MODELS.get((provider or "").strip().lower(), "gpt-4.1") - - -def llm_env_key(provider: str) -> str: - return _ENV_KEYS.get((provider or "").strip().lower(), "OPENAI_API_KEY") - - -def _parse_aws_credentials(api_key: str) -> Dict[str, Any]: - """Parse AWS credential JSON or fall back to access key + env secret.""" - try: - parsed = json.loads(api_key) - if isinstance(parsed, dict): - access = ( - parsed.get("aws_access_key_id") - or parsed.get("access_key_id") - or parsed.get("aws_access_key") - ) - secret = ( - parsed.get("aws_secret_access_key") - or parsed.get("secret_access_key") - or parsed.get("aws_secret_key") - ) - return { - "aws_access_key": access, - "aws_secret_key": secret, - "aws_session_token": parsed.get("aws_session_token") - or parsed.get("session_token"), - "aws_region": parsed.get("aws_region") - or parsed.get("region") - or os.getenv("AWS_REGION", "us-east-1"), - } - except (json.JSONDecodeError, TypeError): - pass - return { - "aws_access_key": api_key, - "aws_secret_key": os.getenv("AWS_SECRET_ACCESS_KEY"), - "aws_region": os.getenv("AWS_REGION", "us-east-1"), - } - - -def get_llm_provider_registry(get_service: Callable[[str], Any]) -> Dict[str, Dict[str, Any]]: - """Build the LLM provider registry used by ``run_voice_bundle_fastapi``.""" - - def _openai_factory(api_key, model, params=None, base_url=None): - kwargs: Dict[str, Any] = {"api_key": api_key, "model": normalize_llm_model("openai", model)} - if params: - kwargs["params"] = params - if base_url: - kwargs["base_url"] = base_url - return get_service("OpenAILLMService")(**kwargs) - - registry: Dict[str, Dict[str, Any]] = {} - - for provider in sorted(LLM_VOICE_PROVIDER_KEYS): - env_key = llm_env_key(provider) - default_model = default_llm_model(provider) - - if provider == "openai": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, _f=_openai_factory: _f( - api_key, model, params - ), - } - elif provider == "google": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("GoogleLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider == "anthropic": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("AnthropicLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider == "fireworks": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("FireworksLLMService")( - api_key=api_key, - model=normalize_llm_model("fireworks", model), - **({"params": params} if params else {}), - ), - } - elif provider == "xai": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("GrokLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider == "mistral": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("MistralLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider in ("together", "meta"): - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("TogetherLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider == "perplexity": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("PerplexityLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider == "openrouter": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenRouterLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - } - elif provider == "aws": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("AWSBedrockLLMService")( - model=model, - params=params, - **_parse_aws_credentials(api_key), - ), - } - elif provider == "cohere": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenAILLMService")( - api_key=api_key, - model=model, - base_url="https://api.cohere.com/compatibility/v1", - **({"params": params} if params else {}), - ), - } - elif provider == "sarvam": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenAILLMService")( - api_key=api_key, - model=model, - base_url="https://api.sarvam.ai/v1", - **({"params": params} if params else {}), - ), - } - elif provider == "custom": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - "factory": lambda api_key, model, params=None, base_url=None, _f=_openai_factory: _f( - api_key, model, params, base_url=base_url - ), - "supports_base_url": True, - } - elif provider == "azure": - registry[provider] = { - "env_key": env_key, - "default_model": default_model, - # Azure is instantiated with endpoint metadata in run_voice_bundle_fastapi. - "factory": lambda api_key, model, params=None, _f=_openai_factory: _f( - api_key, normalize_llm_model("azure", model), params - ), - } - - return registry - - -def resolve_voice_llm_base_url(db, organization_id, voice_bundle, llm_provider) -> Optional[str]: - """Resolve an OpenAI-compatible base URL for custom / gateway-routed LLM legs.""" - provider_key = ( - llm_provider.value if hasattr(llm_provider, "value") else str(llm_provider) - ).lower() - if provider_key != "custom": - return None - - from app.services.credentials import resolve_ai_provider - - ai_provider = resolve_ai_provider( - provider_key, - db, - organization_id, - credential_id=getattr(voice_bundle, "llm_credential_id", None), - ) - if not ai_provider: - return None - - base_url = getattr(ai_provider, "gateway_base_url", None) - if base_url and str(base_url).strip(): - return str(base_url).strip() - return None - - -def instantiate_llm_service( - provider: str, - *, - get_service: Callable[[str], Any], - api_key: str, - model: str, - params: Optional[Any] = None, - base_url: Optional[str] = None, -): - """Instantiate a streaming LLM service for the live voice pipeline.""" - registry = get_llm_provider_registry(get_service) - provider_key = (provider or "").strip().lower() - cfg = registry.get(provider_key) - if cfg is None: - supported = ", ".join(sorted(registry.keys())) - raise ValueError( - f"Unsupported LLM provider '{provider_key}'. Supported providers: {supported}" - ) - - factory = cfg["factory"] - if provider_key == "custom" or cfg.get("supports_base_url"): - return factory(api_key, model, params, base_url=base_url) - if base_url: - logger.debug( - "Ignoring llm base_url for provider '{}' (not OpenAI-compatible custom routing)", - provider_key, - ) - return factory(api_key, model, params) +"""Live voice pipeline LLM provider registry and service factory. + +Mirrors the LLM-capable providers exposed in Voice Bundles / Integrations +(see ``app/services/judge_alignment/model_catalog.py``). +""" + +from __future__ import annotations + +import json +import os +from typing import Any, Callable, Dict, Optional + +from loguru import logger + +# Providers selectable for the LLM leg of STT+LLM+TTS voice bundles. +LLM_VOICE_PROVIDER_KEYS = frozenset( + { + "openai", + "anthropic", + "google", + "xai", + "fireworks", + "cohere", + "mistral", + "meta", + "together", + "perplexity", + "azure", + "aws", + "openrouter", + "custom", + "sarvam", + } +) + +_DEFAULT_LLM_MODELS: Dict[str, str] = { + "openai": "gpt-4.1", + "google": "gemini-2.5-flash", + "anthropic": "claude-sonnet-4.6", + "xai": "grok-3-beta", + "fireworks": "accounts/fireworks/models/llama-v3p1-8b-instruct", + "cohere": "command-r-plus-08-2024", + "mistral": "mistral-small-latest", + "meta": "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", + "together": "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", + "perplexity": "sonar", + "azure": "gpt-4.1", + "aws": "amazon.nova-lite-v1:0", + "openrouter": "openai/gpt-4o-2024-11-20", + "custom": "gpt-4o-mini", + "sarvam": "sarvam-30b", +} + +_ENV_KEYS: Dict[str, str] = { + "openai": "OPENAI_API_KEY", + "google": "GOOGLE_API_KEY", + "anthropic": "ANTHROPIC_API_KEY", + "xai": "XAI_API_KEY", + "fireworks": "FIREWORKS_API_KEY", + "cohere": "COHERE_API_KEY", + "mistral": "MISTRAL_API_KEY", + "meta": "TOGETHER_API_KEY", + "together": "TOGETHER_API_KEY", + "perplexity": "PERPLEXITY_API_KEY", + "azure": "AZURE_OPENAI_API_KEY", + "aws": "AWS_ACCESS_KEY_ID", + "openrouter": "OPENROUTER_API_KEY", + "custom": "OPENAI_API_KEY", + "sarvam": "SARVAM_API_KEY", +} + + +def normalize_llm_model(provider: str, model: str) -> str: + """Normalize catalog model ids for provider-specific APIs.""" + provider_key = (provider or "").strip().lower() + if not model: + return model + if provider_key == "fireworks" and not model.startswith("accounts/"): + return f"accounts/fireworks/models/{model}" + if provider_key == "azure": + from app.services.ai.llm_service import _azure_deployment_name + + return _azure_deployment_name(model) + return model + + +def default_llm_model(provider: str) -> str: + return _DEFAULT_LLM_MODELS.get((provider or "").strip().lower(), "gpt-4.1") + + +def llm_env_key(provider: str) -> str: + return _ENV_KEYS.get((provider or "").strip().lower(), "OPENAI_API_KEY") + + +def _parse_aws_credentials(api_key: str) -> Dict[str, Any]: + """Parse AWS credential JSON or fall back to access key + env secret.""" + try: + parsed = json.loads(api_key) + if isinstance(parsed, dict): + access = ( + parsed.get("aws_access_key_id") + or parsed.get("access_key_id") + or parsed.get("aws_access_key") + ) + secret = ( + parsed.get("aws_secret_access_key") + or parsed.get("secret_access_key") + or parsed.get("aws_secret_key") + ) + return { + "aws_access_key": access, + "aws_secret_key": secret, + "aws_session_token": parsed.get("aws_session_token") + or parsed.get("session_token"), + "aws_region": parsed.get("aws_region") + or parsed.get("region") + or os.getenv("AWS_REGION", "us-east-1"), + } + except (json.JSONDecodeError, TypeError): + pass + return { + "aws_access_key": api_key, + "aws_secret_key": os.getenv("AWS_SECRET_ACCESS_KEY"), + "aws_region": os.getenv("AWS_REGION", "us-east-1"), + } + + +def get_llm_provider_registry(get_service: Callable[[str], Any]) -> Dict[str, Dict[str, Any]]: + """Build the LLM provider registry used by ``run_voice_bundle_fastapi``.""" + + def _openai_factory(api_key, model, params=None, base_url=None): + kwargs: Dict[str, Any] = {"api_key": api_key, "model": normalize_llm_model("openai", model)} + if params: + kwargs["params"] = params + if base_url: + kwargs["base_url"] = base_url + return get_service("OpenAILLMService")(**kwargs) + + registry: Dict[str, Dict[str, Any]] = {} + + for provider in sorted(LLM_VOICE_PROVIDER_KEYS): + env_key = llm_env_key(provider) + default_model = default_llm_model(provider) + + if provider == "openai": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, _f=_openai_factory: _f( + api_key, model, params + ), + } + elif provider == "google": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("GoogleLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "anthropic": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("AnthropicLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "fireworks": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("FireworksLLMService")( + api_key=api_key, + model=normalize_llm_model("fireworks", model), + **({"params": params} if params else {}), + ), + } + elif provider == "xai": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("GrokLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "mistral": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("MistralLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider in ("together", "meta"): + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("TogetherLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "perplexity": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("PerplexityLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "openrouter": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenRouterLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "aws": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("AWSBedrockLLMService")( + model=model, + params=params, + **_parse_aws_credentials(api_key), + ), + } + elif provider == "cohere": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenAILLMService")( + api_key=api_key, + model=model, + base_url="https://api.cohere.com/compatibility/v1", + **({"params": params} if params else {}), + ), + } + elif provider == "sarvam": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenAILLMService")( + api_key=api_key, + model=model, + base_url="https://api.sarvam.ai/v1", + **({"params": params} if params else {}), + ), + } + elif provider == "custom": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, base_url=None, _f=_openai_factory: _f( + api_key, model, params, base_url=base_url + ), + "supports_base_url": True, + } + elif provider == "azure": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + # Azure is instantiated with endpoint metadata in run_voice_bundle_fastapi. + "factory": lambda api_key, model, params=None, _f=_openai_factory: _f( + api_key, normalize_llm_model("azure", model), params + ), + } + + return registry + + +def resolve_voice_llm_base_url(db, organization_id, voice_bundle, llm_provider) -> Optional[str]: + """Resolve an OpenAI-compatible base URL for custom / gateway-routed LLM legs.""" + provider_key = ( + llm_provider.value if hasattr(llm_provider, "value") else str(llm_provider) + ).lower() + if provider_key != "custom": + return None + + from app.services.credentials import resolve_ai_provider + + ai_provider = resolve_ai_provider( + provider_key, + db, + organization_id, + credential_id=getattr(voice_bundle, "llm_credential_id", None), + ) + if not ai_provider: + return None + + base_url = getattr(ai_provider, "gateway_base_url", None) + if base_url and str(base_url).strip(): + return str(base_url).strip() + return None + + +def instantiate_llm_service( + provider: str, + *, + get_service: Callable[[str], Any], + api_key: str, + model: str, + params: Optional[Any] = None, + base_url: Optional[str] = None, +): + """Instantiate a streaming LLM service for the live voice pipeline.""" + registry = get_llm_provider_registry(get_service) + provider_key = (provider or "").strip().lower() + cfg = registry.get(provider_key) + if cfg is None: + supported = ", ".join(sorted(registry.keys())) + raise ValueError( + f"Unsupported LLM provider '{provider_key}'. Supported providers: {supported}" + ) + + factory = cfg["factory"] + if provider_key == "custom" or cfg.get("supports_base_url"): + return factory(api_key, model, params, base_url=base_url) + if base_url: + logger.debug( + "Ignoring llm base_url for provider '{}' (not OpenAI-compatible custom routing)", + provider_key, + ) + return factory(api_key, model, params) diff --git a/app/workers/concurrency/telephony_credential_rate_limit.py b/app/workers/concurrency/telephony_credential_rate_limit.py index c2c9c66e..e5617ffe 100644 --- a/app/workers/concurrency/telephony_credential_rate_limit.py +++ b/app/workers/concurrency/telephony_credential_rate_limit.py @@ -98,13 +98,16 @@ def fingerprint_for_integration(integration) -> str: ) +_CREDENTIALED_RECORDING_IMPORT_PROVIDERS = frozenset({"exotel", "plivo"}) + + def requires_authenticated_recording_fetch(call_import) -> bool: """True when call-import recording fetch uses provider HTTP auth.""" if call_import.telephony_integration_id is None and not ( call_import.provider or "" ).strip(): return False - return (call_import.provider or "").strip().lower() == "exotel" + return (call_import.provider or "").strip().lower() in _CREDENTIALED_RECORDING_IMPORT_PROVIDERS _PEEK_LUA = """ diff --git a/app/workers/tasks/process_call_import_row.py b/app/workers/tasks/process_call_import_row.py index d3299b3d..efc177d7 100644 --- a/app/workers/tasks/process_call_import_row.py +++ b/app/workers/tasks/process_call_import_row.py @@ -11,8 +11,11 @@ * **Exotel credentialed import**: download the CSV-supplied ``recording_url`` with HTTP Basic auth from the batch's pinned credentials. - * **Other credentialed providers** (e.g. Plivo): download the - CSV-supplied ``recording_url`` without auth (public links). + * **Plivo credentialed import**: download the CSV-supplied + ``recording_url`` with HTTP Basic auth from the batch's pinned + credentials. + * **Other credentialed providers**: download the CSV-supplied + ``recording_url`` without auth (public links). Transient download errors schedule a Celery retry; auth/4xx/oversize errors mark the row failed immediately. @@ -220,11 +223,14 @@ def _row_or_import_gone( return None +_CREDENTIALED_RECORDING_IMPORT_PROVIDERS = frozenset({"exotel", "plivo"}) + + def _use_credentialed_recording_download(call_import, client) -> bool: """True when CSV recording URLs should be fetched with provider auth.""" if client is None or not hasattr(client, "download_recording"): return False - return (call_import.provider or "").lower() == "exotel" + return (call_import.provider or "").lower() in _CREDENTIALED_RECORDING_IMPORT_PROVIDERS def _is_direct_url_import(call_import) -> bool: @@ -498,8 +504,8 @@ def process_call_import_row_task( else: # ------------------------------------------------------------------ # Credentialed import — download from the CSV-supplied URL only. - # Exotel URLs require HTTP Basic auth; other providers use public - # download (e.g. Plivo presigned links). + # Exotel and Plivo URLs require HTTP Basic auth; other providers + # use public download (e.g. presigned links). # ------------------------------------------------------------------ audio_bytes: Optional[bytes] = None content_type: Optional[str] = None diff --git a/docs/telephony-media.md b/docs/telephony-media.md index 1af5d4cf..72c5491d 100644 --- a/docs/telephony-media.md +++ b/docs/telephony-media.md @@ -15,7 +15,7 @@ Vobiz sees **one public host** (the telephony edge): - `POST/GET /api/v1/telephony/vobiz/webhooks/answer` - `POST /api/v1/telephony/vobiz/webhooks/events` - `POST /api/v1/telephony/vobiz/webhooks/recording-ready` -- `WSS /api/v1/telephony/vobiz/ws` +- `WSS /api/v1/telephony/carrier/ws` (shared live-audio socket for Plivo, Vobiz, and other Stream-compatible carriers) Configure **`vobiz.webhook_base_url`** to that public URL (e.g. `https://telephony.staging.example.com`). When `media_ws_base_url` / `MEDIA_WS_BASE_URL` is unset, carrier answer XML reuses the same host (`https` → `wss`) via `carrier_media_ws_base_url()`. diff --git a/frontend/src/pages/agents/AgentWorkspaceDetail.tsx b/frontend/src/pages/agents/AgentWorkspaceDetail.tsx index d8247847..1b12af9b 100644 --- a/frontend/src/pages/agents/AgentWorkspaceDetail.tsx +++ b/frontend/src/pages/agents/AgentWorkspaceDetail.tsx @@ -34,6 +34,25 @@ function normalizeCallMedium(value: string | undefined | null): 'phone_call' | ' return value === 'web_call' ? 'web_call' : 'phone_call' } +function agentToFormData(agent: NonNullable>>): FormData { + return { + name: agent.name, + phone_number: agent.phone_number || '', + language: agent.language, + description: agent.description || '', + test_agent_template: templateFromApi(agent.test_agent_template), + prompt_variables: agent.prompt_variables || {}, + silence_hangup_secs: agent.silence_hangup_secs ?? 15, + call_type: agent.call_type, + call_medium: normalizeCallMedium(agent.call_medium), + telephony_phone_number_id: agent.telephony_phone_number_id || '', + voice_bundle_id: agent.voice_bundle_id || '', + voice_ai_integration_id: agent.voice_ai_integration_id || '', + voice_ai_agent_id: agent.voice_ai_agent_id || '', + provider_prompt: agent.provider_prompt || '', + } +} + interface FormData { name: string phone_number: string @@ -142,27 +161,10 @@ export default function AgentWorkspaceDetail({ } useEffect(() => { - if (agent) { - setFormData({ - name: agent.name, - phone_number: agent.phone_number || '', - language: agent.language, - description: agent.description || '', - test_agent_template: agent.test_agent_template - ? templateFromApi(agent.test_agent_template) - : defaultTestAgentTemplate(), - prompt_variables: agent.prompt_variables || {}, - silence_hangup_secs: agent.silence_hangup_secs ?? 15, - call_type: agent.call_type, - call_medium: normalizeCallMedium(agent.call_medium), - telephony_phone_number_id: agent.telephony_phone_number_id || '', - voice_bundle_id: agent.voice_bundle_id || '', - voice_ai_integration_id: agent.voice_ai_integration_id || '', - voice_ai_agent_id: agent.voice_ai_agent_id || '', - provider_prompt: agent.provider_prompt || '', - }) + if (agent && !isEditMode) { + setFormData(agentToFormData(agent)) } - }, [agent]) + }, [agent, isEditMode]) const updateMutation = useMutation({ mutationFn: (data: FormData) => { @@ -197,6 +199,7 @@ export default function AgentWorkspaceDetail({ onSuccess: () => { queryClient.invalidateQueries({ queryKey: ['agent', agentRouteId] }) queryClient.invalidateQueries({ queryKey: ['agents'] }) + queryClient.invalidateQueries({ queryKey: ['telephony-numbers'] }) setIsEditMode(false) showToast('Agent updated successfully!', 'success') }, @@ -286,26 +289,16 @@ export default function AgentWorkspaceDetail({ updateMutation.mutate(formData) } + const handleEditClick = () => { + if (agent) { + setFormData(agentToFormData(agent)) + } + setIsEditMode(true) + } + const handleCancelEdit = () => { if (agent) { - setFormData({ - name: agent.name, - phone_number: agent.phone_number || '', - language: agent.language, - description: agent.description || '', - test_agent_template: agent.test_agent_template - ? templateFromApi(agent.test_agent_template) - : defaultTestAgentTemplate(), - prompt_variables: agent.prompt_variables || {}, - silence_hangup_secs: agent.silence_hangup_secs ?? 15, - call_type: agent.call_type, - call_medium: normalizeCallMedium(agent.call_medium), - telephony_phone_number_id: agent.telephony_phone_number_id || '', - voice_bundle_id: agent.voice_bundle_id || '', - voice_ai_integration_id: agent.voice_ai_integration_id || '', - voice_ai_agent_id: agent.voice_ai_agent_id || '', - provider_prompt: agent.provider_prompt || '', - }) + setFormData(agentToFormData(agent)) } setIsEditMode(false) } @@ -409,7 +402,7 @@ export default function AgentWorkspaceDetail({ agentId={agent.agent_id} isEditMode={isEditMode} isPending={updateMutation.isPending} - onEditClick={() => setIsEditMode(true)} + onEditClick={handleEditClick} onCancelEdit={handleCancelEdit} onSave={handleSave} /> diff --git a/frontend/src/pages/agents/AgentsWorkspace.tsx b/frontend/src/pages/agents/AgentsWorkspace.tsx index d242762e..48195acf 100644 --- a/frontend/src/pages/agents/AgentsWorkspace.tsx +++ b/frontend/src/pages/agents/AgentsWorkspace.tsx @@ -116,6 +116,7 @@ export default function AgentsWorkspace() { const handleCreateSuccess = () => { queryClient.invalidateQueries({ queryKey: ['agents'] }) + queryClient.invalidateQueries({ queryKey: ['telephony-numbers'] }) setShowCreateModal(false) } diff --git a/frontend/src/pages/agents/components/AgentEditForm.tsx b/frontend/src/pages/agents/components/AgentEditForm.tsx index 8f566c38..2dd07657 100644 --- a/frontend/src/pages/agents/components/AgentEditForm.tsx +++ b/frontend/src/pages/agents/components/AgentEditForm.tsx @@ -161,20 +161,16 @@ export default function AgentEditForm({ } if (!canUseProviderNumbers || telephonyNumbers.length === 0) { - setPhoneNumberInputMode('custom') - return - } - - const selectedExists = telephonyNumbers.some((n) => n.id === formData.telephony_phone_number_id) - if (!selectedExists && phoneNumberInputMode === 'provider') { - onChange({ ...formData, telephony_phone_number_id: '', phone_number: '' }) + if (phoneNumberInputMode === 'provider') { + setPhoneNumberInputMode('custom') + } } }, [ - formData, - onChange, + formData.call_medium, + formData.telephony_phone_number_id, phoneNumberInputMode, canUseProviderNumbers, - telephonyNumbers, + telephonyNumbers.length, ]) const generateFromProductionMutation = useMutation({ @@ -431,23 +427,27 @@ export default function AgentEditForm({ disabled={!canUseProviderNumbers || telephonyNumbers.length === 0} > - {telephonyNumbers.map((number) => ( - - ))} + {telephonyNumbers.map((number) => { + const assignedToOtherAgent = + !!number.agent_id && number.agent_id !== agentId + return ( + + ) + })} ) : (