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", }, }