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

from __future__ import annotations

import json
import uuid
from typing import Any

import httpx


class KontorMCPError(RuntimeError):
    """Fehler bei der Kommunikation mit dem Kontor MCP Server."""


class KontorMCPClient:
    """
    MCP Streamable-HTTP Client für die 3CX-Telefonie-Middleware.

    Der Client kapselt die MCP-Kommunikation vollständig.
    Die Telefonie-Middleware kennt keine mssRest-Details.
    """

    def __init__(
        self,
        url: str,
        token: str,
        timeout: float = 10.0,
    ) -> None:
        self.url = url.rstrip("/")
        self.token = token
        self.timeout = timeout
        self.session_id: str | None = None
        self._request_id = 0

    def _next_id(self) -> int:
        self._request_id += 1
        return self._request_id

    def _headers(self) -> dict[str, str]:
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            "X-Request-ID": str(uuid.uuid4()),
        }

        if self.session_id:
            headers["Mcp-Session-Id"] = self.session_id

        return headers

    async def _post(
        self,
        payload: dict[str, Any],
    ) -> httpx.Response:
        try:
            async with httpx.AsyncClient(
                timeout=self.timeout,
                follow_redirects=True,
            ) as client:
                response = await client.post(
                    self.url,
                    headers=self._headers(),
                    json=payload,
                )
        except httpx.TimeoutException as exc:
            raise KontorMCPError(
                "Kontor MCP timeout"
            ) from exc
        except httpx.HTTPError as exc:
            raise KontorMCPError(
                f"Kontor MCP HTTP error: {exc}"
            ) from exc

        # MCP Streamable HTTP liefert die Session-ID beim Initialize.
        session_id = response.headers.get("mcp-session-id")
        if session_id:
            self.session_id = session_id

        if response.status_code >= 400:
            raise KontorMCPError(
                f"Kontor MCP HTTP {response.status_code}: "
                f"{response.text[:500]}"
            )

        return response

    @staticmethod
    def _extract_result(response: httpx.Response) -> Any:
        content_type = response.headers.get(
            "content-type",
            "",
        ).lower()

        if "application/json" in content_type:
            data = response.json()

            if "error" in data:
                error = data["error"]
                raise KontorMCPError(
                    f"MCP error {error.get('code')}: "
                    f"{error.get('message')}"
                )

            return data.get("result")

        # Streamable HTTP / SSE
        for line in response.text.splitlines():
            line = line.strip()

            if not line.startswith("data:"):
                continue

            raw = line[5:].strip()

            if not raw:
                continue

            try:
                data = json.loads(raw)
            except json.JSONDecodeError:
                continue

            if "error" in data:
                error = data["error"]
                raise KontorMCPError(
                    f"MCP error {error.get('code')}: "
                    f"{error.get('message')}"
                )

            if "result" in data:
                return data["result"]

        # Manche MCP-Server beantworten Notifications mit 202/leer.
        if not response.text.strip():
            return None

        raise KontorMCPError(
            "Kontor MCP lieferte keine verwertbare Antwort"
        )

    async def initialize(self) -> dict[str, Any]:
        """
        MCP Handshake.

        Wichtig:
        1. initialize ohne Session
        2. Session-ID aus Response übernehmen
        3. notifications/initialized mit Session senden
        """

        payload = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {
                    "name": "3cx-telefonie-middleware",
                    "version": "1.0.0",
                },
            },
        }

        response = await self._post(payload)

        if not self.session_id:
            raise KontorMCPError(
                "MCP Server hat keine Session-ID geliefert"
            )

        result = self._extract_result(response)

        # MCP Notification: keine id, keine Antwort erforderlich.
        notification = {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
        }

        await self._post(notification)

        return result or {}

    async def list_tools(self) -> list[dict[str, Any]]:
        result = await self._request(
            "tools/list",
            {},
        )

        if not result:
            return []

        return result.get("tools", [])

    async def _request(
        self,
        method: str,
        params: dict[str, Any] | None = None,
    ) -> Any:
        payload: dict[str, Any] = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": method,
        }

        if params is not None:
            payload["params"] = params

        response = await self._post(payload)

        return self._extract_result(response)

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
    ) -> Any:
        return await self._request(
            "tools/call",
            {
                "name": name,
                "arguments": arguments or {},
            },
        )

    async def find_customer_by_phone(
        self,
        phone: str,
    ) -> Any:
        """
        Kundensuche über mehrere deterministische Telefonformate.

        Kontor kann Telefonnummern je nach Datenbestand unterschiedlich
        gespeichert haben. Wir versuchen deshalb E.164, national,
        ohne führende 0 und ohne Plus/Formatierung.

        Es wird niemals fuzzy über Telefonnummern gesucht.
        """

        variants = self._phone_variants(phone)

        matches: list[tuple[str, Any]] = []

        for variant in variants:
            result = await self.call_tool(
                "find_customer_by_phone",
                {"phone": variant},
            )

            # Nur tatsächlich gefundene Treffer sammeln.
            if self._phone_result_found(result):
                matches.append((variant, result))

        if not matches:
            return {
                "status": "not_found",
                "input": phone,
                "variants_tried": variants,
            }

        # Doppelte Treffer desselben Kunden zusammenführen.
        unique: dict[str, tuple[str, Any]] = {}

        for variant, result in matches:
            customer_id = self._customer_identity(result)

            if customer_id is None:
                # Ohne ID nicht künstlich zusammenführen.
                unique[f"{variant}:{len(unique)}"] = (
                    variant,
                    result,
                )
            else:
                unique[str(customer_id)] = (
                    variant,
                    result,
                )

        if len(unique) > 1:
            return {
                "status": "ambiguous",
                "input": phone,
                "variants_tried": variants,
                "matches": [
                    {
                        "matched_input": variant,
                        "result": result,
                    }
                    for variant, result in unique.values()
                ],
            }

        variant, result = next(iter(unique.values()))

        # Bestehenden MCP-Response ergänzen, statt ihn zu verlieren.
        if isinstance(result, dict):
            result = dict(result)
            result["matched_input"] = variant
            result["input"] = phone
            result["phone_variants_tried"] = variants

        return result

    @staticmethod
    def _phone_variants(phone: str) -> list[str]:
        import re

        raw = str(phone or "").strip()

        if not raw:
            return []

        digits = re.sub(r"\D", "", raw)

        if not digits:
            return []

        variants: list[str] = []

        def add(value: str) -> None:
            if value and value not in variants:
                variants.append(value)

        # Bereits vorhandene Schreibweise zuerst.
        add(raw)

        # Deutsche Mobil-/Festnetznummern.
        if digits.startswith("49"):
            national = "0" + digits[2:]
            subscriber = digits[2:]

            add("+" + digits)
            add(digits)
            add(national)
            add(subscriber)

        elif digits.startswith("0"):
            subscriber = digits[1:]

            add(digits)
            add("+49" + subscriber)
            add("49" + subscriber)
            add(subscriber)

        else:
            # Für bereits international gespeicherte Nummern
            # ohne +.
            add("+49" + digits)
            add("49" + digits)
            add("0" + digits)
            add(digits)

        return variants

    @staticmethod
    def _phone_result_found(result: Any) -> bool:
        if not isinstance(result, dict):
            return False

        status = result.get("status")

        if status in {
            "exact",
            "high_confidence",
        }:
            return True

        if result.get("found") is True:
            return True

        # Manche MCP-Responses liefern den Kunden direkt.
        return bool(
            result.get("customer_id")
            or result.get("customer_number")
        )

    @staticmethod
    def _customer_identity(result: Any) -> Any:
        if not isinstance(result, dict):
            return None

        return (
            result.get("customer_id")
            or result.get("customer_number")
            or result.get("id")
        )

    async def find_customer_by_email(
        self,
        email: str,
    ) -> Any:
        return await self.call_tool(
            "find_customer_by_email",
            {"email": email},
        )

    async def find_customer_by_number(
        self,
        customer_number: str,
    ) -> Any:
        return await self.call_tool(
            "find_customer_by_number",
            {"customer_number": customer_number},
        )

    async def search_products(
        self,
        query: str,
    ) -> Any:
        return await self.call_tool(
            "search_products",
            {"query": query},
        )

    async def get_product(
        self,
        product_id: str,
    ) -> Any:
        return await self.call_tool(
            "get_product",
            {"product_id": product_id},
        )

    async def get_product_categories(self) -> Any:
        return await self.call_tool(
            "get_product_categories",
            {},
        )

    async def get_order(
        self,
        order_number: str,
    ) -> Any:
        return await self.call_tool(
            "get_order",
            {"order_number": order_number},
        )

    async def get_customer_context(
        self,
        *,
        phone: str | None = None,
        customer_number: str | None = None,
    ) -> Any:
        arguments: dict[str, Any] = {}

        if phone:
            arguments["phone"] = phone

        if customer_number:
            arguments["customer_number"] = customer_number

        return await self.call_tool(
            "get_customer_context",
            arguments,
        )

    async def get_order_context(
        self,
        order_number: str,
    ) -> Any:
        return await self.call_tool(
            "get_order_context",
            {"order_number": order_number},
        )

    async def get_product_context(
        self,
        query: str | None = None,
        product_id: str | None = None,
    ) -> Any:
        arguments: dict[str, Any] = {}

        if query:
            arguments["query"] = query

        if product_id:
            arguments["product_id"] = product_id

        return await self.call_tool(
            "get_product_context",
            arguments,
        )