PraisonAIUI/examples/python/voice-agent/integrations/voice/processor.py

"""Process voice provider webhook events."""

from __future__ import annotations

import asyncio
import json
import logging
from typing import Any

from integrations.voice.agent_runner import build_assistant_request_response, execute_voice_tool
from integrations.voice.config import VoiceSettings
from integrations.voice.live import transcript_hub
from integrations.voice.store import VoiceCallStore, webhook_event_key

logger = logging.getLogger(__name__)


def _message(payload: dict[str, Any]) -> dict[str, Any]:
    msg = payload.get("message")
    return msg if isinstance(msg, dict) else payload


def _call_id(payload: dict[str, Any]) -> str | None:
    msg = _message(payload)
    call = msg.get("call")
    if isinstance(call, dict) and call.get("id"):
        return str(call["id"])
    return None


def _customer_number(payload: dict[str, Any]) -> str | None:
    msg = _message(payload)
    call = msg.get("call") if isinstance(msg.get("call"), dict) else {}
    customer = call.get("customer") if isinstance(call, dict) else {}
    if isinstance(customer, dict) and customer.get("number"):
        return str(customer["number"])
    return None


def _tool_parameters(item: dict[str, Any]) -> dict[str, Any]:
    """Extract tool params from provider payloads (parameters or arguments)."""
    for key in ("parameters", "arguments"):
        value = item.get(key)
        if isinstance(value, dict):
            return value
    fn = item.get("function")
    if isinstance(fn, dict):
        for key in ("parameters", "arguments"):
            value = fn.get(key)
            if isinstance(value, dict):
                return value
    return {}


def _spoken_tool_result(raw: str) -> str:
    """Return plain spoken text for the voice pipeline (no newlines)."""
    try:
        data = json.loads(raw)
        if isinstance(data, dict):
            spoken = data.get("spoken") or data.get("message")
            if spoken:
                return str(spoken).replace("\n", " ").strip()
    except json.JSONDecodeError:
        pass
    return str(raw).replace("\n", " ").strip()


def process_voice_webhook(
    event_type: str, payload: dict[str, Any], *, settings: VoiceSettings
) -> dict[str, Any] | None:
    """Handle a verified provider webhook POST. Returns response body when required."""
    event_key = webhook_event_key(event_type, payload)
    if event_type not in ("tool-calls", "assistant-request", "transfer-destination-request", "knowledge-base-request"):
        if not VoiceCallStore.mark_event_processed(event_key, event_type):
            logger.info("Skipping duplicate voice event %s", event_type)
            return None

    msg = _message(payload)
    call_id = _call_id(payload)

    if event_type == "assistant-request":
        return build_assistant_request_response(settings)

    if event_type == "tool-calls" and call_id:
        tool_calls = msg.get("toolCallList") or []
        if not tool_calls and isinstance(msg.get("toolWithToolCallList"), list):
            for entry in msg["toolWithToolCallList"]:
                if not isinstance(entry, dict):
                    continue
                tool_call = entry.get("toolCall")
                if isinstance(tool_call, dict):
                    tool_calls.append(
                        {
                            "id": tool_call.get("id"),
                            "name": entry.get("name") or (tool_call.get("function") or {}).get("name"),
                            "parameters": _tool_parameters(tool_call),
                            "arguments": _tool_parameters(tool_call),
                        }
                    )
        results = []
        for item in tool_calls:
            if not isinstance(item, dict):
                continue
            tool_call_id = str(item.get("id") or "")
            name = str(item.get("name") or "")
            if not name:
                fn = item.get("function")
                if isinstance(fn, dict):
                    name = str(fn.get("name") or "")
            params = _tool_parameters(item)
            result = _spoken_tool_result(execute_voice_tool(name, params))
            results.append({"toolCallId": tool_call_id, "name": name, "result": result})
        VoiceCallStore.upsert_call(call_id, status="in-progress", customer_number=_customer_number(payload))
        return {"results": results}

    if event_type == "status-update" and call_id:
        status = str(msg.get("status") or "unknown")
        VoiceCallStore.upsert_call(call_id, status=status, customer_number=_customer_number(payload))
        return None

    if event_type == "transcript" and call_id:
        role = str(msg.get("role") or "unknown")
        text = str(msg.get("transcript") or "").strip()
        kind = str(msg.get("transcriptType") or "final")
        if text:
            line = f"{role}: {text}"
            if kind == "final":
                VoiceCallStore.append_transcript_line(call_id, line)
            else:
                VoiceCallStore.upsert_call(
                    call_id,
                    status="in-progress",
                    customer_number=_customer_number(payload),
                    metadata={"partial_transcript": line},
                )
            transcript_hub.publish(
                call_id,
                {"type": "transcript", "role": role, "text": text, "transcriptType": kind},
            )
            from integrations.voice.chat_bridge import enqueue_transcript

            enqueue_transcript(call_id, role, text, final=kind == "final")
        return None

    if event_type == "end-of-call-report" and call_id:
        artifact = msg.get("artifact") if isinstance(msg.get("artifact"), dict) else {}
        transcript = artifact.get("transcript") if isinstance(artifact, dict) else None
        summary = str(msg.get("summary") or "")
        ended = str(msg.get("endedReason") or "ended")
        VoiceCallStore.upsert_call(
            call_id,
            status=f"ended ({ended})",
            customer_number=_customer_number(payload),
            transcript=str(transcript) if transcript else None,
            summary=summary or None,
            metadata={"artifact": artifact},
        )
        from integrations.voice.call_finalize import finalize_call

        asyncio.run(finalize_call(call_id, status=f"ended ({ended})"))
        return None

    if event_type == "conversation-update" and call_id:
        messages = msg.get("messages") or []
        lines = []
        for item in messages:
            if not isinstance(item, dict):
                continue
            role = item.get("role") or "unknown"
            text = item.get("message") or item.get("content") or ""
            if text:
                lines.append(f"{role}: {text}")
        if lines:
            VoiceCallStore.upsert_call(
                call_id, transcript="\n".join(lines), customer_number=_customer_number(payload)
            )
        return None

    logger.debug("Unhandled voice event %s: %s", event_type, json.dumps(payload)[:300])
    return None