3cx-telefonie-middleware/app/call_context.py

from __future__ import annotations

import asyncio
import json
import logging
from typing import Any

from app.kontor_mcp import KontorMCPClient

logger = logging.getLogger(__name__)


class CallContextService:

    def __init__(self, mcp_client: KontorMCPClient):
        self.mcp = mcp_client

        # E.164 -> context
        self._cache: dict[str, dict[str, Any]] = {}

        # E.164 -> currently running lookup
        self._tasks: dict[str, asyncio.Task] = {}

        self._lock = asyncio.Lock()

    async def enrich_phone(
        self,
        e164: str,
    ) -> dict[str, Any]:

        if not e164:
            return self._empty("invalid")

        async with self._lock:

            cached = self._cache.get(e164)

            if cached is not None:
                return cached

            task = self._tasks.get(e164)

            if task is None:
                task = asyncio.create_task(
                    self._lookup(e164)
                )
                self._tasks[e164] = task

        try:
            return await task

        finally:
            async with self._lock:
                if self._tasks.get(e164) is task:
                    self._tasks.pop(e164, None)

    async def _lookup(
        self,
        e164: str,
    ) -> dict[str, Any]:

        try:

            raw = await self.mcp.get_customer_context(
                phone=e164
            )

            result = self._extract_result(raw)

            if result is None:
                context = self._empty("unavailable")
            else:
                context = self._normalize_context(
                    result
                )

            async with self._lock:
                self._cache[e164] = context

            return context

        except Exception as exc:

            logger.warning(
                "MCP customer context failed for %s: %s",
                e164,
                exc,
            )

            return {
                "status": "unavailable",
                "matched_on": [],
                "customer": None,
                "open_orders": [],
                "recent_orders": [],
                "hints": [],
                "source": {
                    "system": "kontor-mcp",
                    "error": type(exc).__name__,
                },
            }

    @staticmethod
    def _extract_result(
        raw: Any,
    ) -> dict[str, Any] | None:

        if not isinstance(raw, dict):
            return None

        if raw.get("isError") is True:
            return None

        structured = raw.get(
            "structuredContent"
        )

        if isinstance(structured, dict):

            result = structured.get(
                "result"
            )

            if isinstance(result, dict):
                return result

            if isinstance(result, str):

                try:
                    parsed = json.loads(result)

                    if isinstance(parsed, dict):
                        return parsed

                except json.JSONDecodeError:
                    pass

        content = raw.get("content")

        if isinstance(content, list):

            for item in content:

                if not isinstance(item, dict):
                    continue

                text = item.get("text")

                if not isinstance(text, str):
                    continue

                try:
                    parsed = json.loads(text)

                    if isinstance(parsed, dict):
                        return parsed

                except json.JSONDecodeError:
                    continue

        return None

    @staticmethod
    def _normalize_context(
        result: dict[str, Any],
    ) -> dict[str, Any]:

        customer = result.get(
            "customer"
        )

        matched_on = result.get(
            "matched_on"
        )

        if not matched_on and isinstance(
            customer,
            dict,
        ):
            matched_on = customer.get(
                "matched_on",
                [],
            )

        return {
            "status": result.get(
                "status",
                "unknown",
            ),

            "matched_on": matched_on or [],

            "customer": customer,

            "open_orders": result.get(
                "open_orders",
                [],
            ),

            "recent_orders": result.get(
                "recent_orders",
                [],
            ),

            "hints": result.get(
                "hints",
                [],
            ),

            "source": {
                "system": "kontor-mcp",
                "request_id": result.get(
                    "request_id"
                ),
            },
        }

    @staticmethod
    def _empty(
        status: str,
    ) -> dict[str, Any]:

        return {
            "status": status,
            "matched_on": [],
            "customer": None,
            "open_orders": [],
            "recent_orders": [],
            "hints": [],
            "source": {
                "system": "kontor-mcp",
            },
        }