call_context.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  1. from __future__ import annotations
  2. import asyncio
  3. import json
  4. import logging
  5. from typing import Any
  6. from app.kontor_mcp import KontorMCPClient
  7. logger = logging.getLogger(__name__)
  8. class CallContextService:
  9. def __init__(self, mcp_client: KontorMCPClient):
  10. self.mcp = mcp_client
  11. # E.164 -> context
  12. self._cache: dict[str, dict[str, Any]] = {}
  13. # E.164 -> currently running lookup
  14. self._tasks: dict[str, asyncio.Task] = {}
  15. self._lock = asyncio.Lock()
  16. async def enrich_phone(
  17. self,
  18. e164: str,
  19. ) -> dict[str, Any]:
  20. if not e164:
  21. return self._empty("invalid")
  22. async with self._lock:
  23. cached = self._cache.get(e164)
  24. if cached is not None:
  25. return cached
  26. task = self._tasks.get(e164)
  27. if task is None:
  28. task = asyncio.create_task(
  29. self._lookup(e164)
  30. )
  31. self._tasks[e164] = task
  32. try:
  33. return await task
  34. finally:
  35. async with self._lock:
  36. if self._tasks.get(e164) is task:
  37. self._tasks.pop(e164, None)
  38. async def _lookup(
  39. self,
  40. e164: str,
  41. ) -> dict[str, Any]:
  42. try:
  43. raw = await self.mcp.get_customer_context(
  44. phone=e164
  45. )
  46. result = self._extract_result(raw)
  47. if result is None:
  48. context = self._empty("unavailable")
  49. else:
  50. context = self._normalize_context(
  51. result
  52. )
  53. async with self._lock:
  54. self._cache[e164] = context
  55. return context
  56. except Exception as exc:
  57. logger.warning(
  58. "MCP customer context failed for %s: %s",
  59. e164,
  60. exc,
  61. )
  62. return {
  63. "status": "unavailable",
  64. "matched_on": [],
  65. "customer": None,
  66. "open_orders": [],
  67. "recent_orders": [],
  68. "hints": [],
  69. "source": {
  70. "system": "kontor-mcp",
  71. "error": type(exc).__name__,
  72. },
  73. }
  74. @staticmethod
  75. def _extract_result(
  76. raw: Any,
  77. ) -> dict[str, Any] | None:
  78. if not isinstance(raw, dict):
  79. return None
  80. if raw.get("isError") is True:
  81. return None
  82. structured = raw.get(
  83. "structuredContent"
  84. )
  85. if isinstance(structured, dict):
  86. result = structured.get(
  87. "result"
  88. )
  89. if isinstance(result, dict):
  90. return result
  91. if isinstance(result, str):
  92. try:
  93. parsed = json.loads(result)
  94. if isinstance(parsed, dict):
  95. return parsed
  96. except json.JSONDecodeError:
  97. pass
  98. content = raw.get("content")
  99. if isinstance(content, list):
  100. for item in content:
  101. if not isinstance(item, dict):
  102. continue
  103. text = item.get("text")
  104. if not isinstance(text, str):
  105. continue
  106. try:
  107. parsed = json.loads(text)
  108. if isinstance(parsed, dict):
  109. return parsed
  110. except json.JSONDecodeError:
  111. continue
  112. return None
  113. @staticmethod
  114. def _normalize_context(
  115. result: dict[str, Any],
  116. ) -> dict[str, Any]:
  117. customer = result.get(
  118. "customer"
  119. )
  120. matched_on = result.get(
  121. "matched_on"
  122. )
  123. if not matched_on and isinstance(
  124. customer,
  125. dict,
  126. ):
  127. matched_on = customer.get(
  128. "matched_on",
  129. [],
  130. )
  131. return {
  132. "status": result.get(
  133. "status",
  134. "unknown",
  135. ),
  136. "matched_on": matched_on or [],
  137. "customer": customer,
  138. "open_orders": result.get(
  139. "open_orders",
  140. [],
  141. ),
  142. "recent_orders": result.get(
  143. "recent_orders",
  144. [],
  145. ),
  146. "hints": result.get(
  147. "hints",
  148. [],
  149. ),
  150. "source": {
  151. "system": "kontor-mcp",
  152. "request_id": result.get(
  153. "request_id"
  154. ),
  155. },
  156. }
  157. @staticmethod
  158. def _empty(
  159. status: str,
  160. ) -> dict[str, Any]:
  161. return {
  162. "status": status,
  163. "matched_on": [],
  164. "customer": None,
  165. "open_orders": [],
  166. "recent_orders": [],
  167. "hints": [],
  168. "source": {
  169. "system": "kontor-mcp",
  170. },
  171. }