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, )