import asyncio import logging import json import os import aiohttp from typing import Any import redis.asyncio as redis_async from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from .env_load import load_agent_runtime_dotenv from .business_id_agents import get_runtime_overrides_from_agents_table from .mcube_defaults import apply_voice_defaults_to_dict, get_default_mcube_call_config log = logging.getLogger("mcube.webhook") load_agent_runtime_dotenv() REDIS_URL = os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0") MCUBE_WEBHOOK_PATH = os.getenv("MCUBE_WEBHOOK_PATH", "/webhooks/mcube") MCUBE_OUTBOUND_PATH = os.getenv("MCUBE_OUTBOUND_PATH", "/api/mcube/outbound-call") # Public URLs (must be reachable by MCube). Example: # MCUBE_PUBLIC_BASE_URL=https://your-public-host # MCUBE_PUBLIC_WS_URL_BASE=wss://your-public-host MCUBE_PUBLIC_BASE_URL = os.getenv("MCUBE_PUBLIC_BASE_URL", "") MCUBE_PUBLIC_WS_URL_BASE = os.getenv("MCUBE_PUBLIC_WS_URL_BASE", "") MCUBE_WS_PATH_PREFIX = os.getenv("MCUBE_WS_PATH_PREFIX", "/bid/websocket") # Backend (Django) base URL used to fetch per-agent MCube config. # In Docker single-container: http://127.0.0.1:8000 AGENT_BACKEND_BASE_URL = os.getenv("AGENT_BACKEND_BASE_URL", "http://127.0.0.1:8000").rstrip("/") APP = FastAPI(title="MCube Webhook Receiver") _redis: redis_async.Redis | None = None @APP.on_event("startup") async def _startup() -> None: global _redis _redis = redis_async.from_url(REDIS_URL, decode_responses=False) # Probe connection quickly await _redis.ping() log.info("mcube webhook: connected to redis %s", REDIS_URL) @APP.get("/health") async def health() -> dict[str, Any]: return {"status": "ok"} @APP.post(MCUBE_WEBHOOK_PATH) async def mcube_webhook(request: Request) -> JSONResponse: try: payload: dict[str, Any] = await request.json() call_sid = payload.get("call_id") or payload.get("callId") or payload.get("callID") status = str(payload.get("status", "")).lower() duration = payload.get("duration", 0) answered_by = payload.get("answered_by", "human") if not call_sid: return JSONResponse({"ok": False, "error": "missing call_id"}, status_code=400) redis = _redis assert redis is not None idem_key = f"mcube_webhook_processed:{call_sid}:{status}" # Idempotency guard: skip duplicates for the same call+status. already = await redis.set(idem_key, 1, nx=True, ex=30) if not already: return JSONResponse({"ok": True, "skipped": True}) # Store latest status for use by other components. await redis.set( f"mcube_call_status:{call_sid}", status.encode("utf-8") if isinstance(status, str) else status, ex=3600 * 24, ) await redis.set( f"mcube_call_duration:{call_sid}", str(duration).encode("utf-8") if isinstance(duration, (int, float)) else str(duration).encode("utf-8"), ex=3600 * 24, ) await redis.set( f"mcube_call_answered_by:{call_sid}", str(answered_by).encode("utf-8"), ex=3600 * 24, ) # Convenience “ended” marker for terminal statuses. terminal = status in {"not_answered", "failed", "voicemail", "blocked", "completed", "no-answer", "busy"} if terminal: await redis.set(f"mcube_call_ended:{call_sid}", 1, ex=3600 * 6) return JSONResponse({"ok": True}) except Exception as e: log.exception("mcube webhook failed") return JSONResponse( {"ok": False, "error": "mcube_webhook_failed", "detail": str(e)}, status_code=500, ) def main() -> None: logging.basicConfig(level=logging.INFO) import uvicorn host = os.getenv("MCUBE_WEBHOOK_HOST", "0.0.0.0") port = int(os.getenv("MCUBE_WEBHOOK_PORT", "8002")) uvicorn.run(APP, host=host, port=port) @APP.post(MCUBE_OUTBOUND_PATH) async def outbound_call(request: Request) -> JSONResponse: """ Minimal endpoint to kick off an outbound MCube click-to-call. This integrates with MCube's Restmcube-api/outbound-calls endpoint which expects: - HTTP header: Authorization - JSON body keys: custnumber, exenumber, gid, refurl, refid (as per your doc) Body examples supported: - { "to": "+1555..." , "exenumber": "8700...", "gid": "1", "call_id": "optional" } - or use env defaults for exenumber/gid/auth. """ from uuid import uuid4 body = await request.json() # Optional: if provided, we use Django agent config (mcube_exenumber/mcube_gid) # instead of relying purely on request body/env defaults. agent_name = str( body.get("agent_name") or body.get("agentName") or body.get("agent") or "" ).strip() business_id = body.get("business_id") or body.get("businessId") or body.get("bid") bot_id = body.get("bot_id") or body.get("botId") async def _fetch_agent_mcube_config(name: str) -> dict[str, Any]: if not name: return {} url = f"{AGENT_BACKEND_BASE_URL}/api/agents/{name}/config/" timeout_s = float(os.getenv("AGENT_BACKEND_FETCH_TIMEOUT_S", "5.0")) try: async with aiohttp.ClientSession() as session: async with session.get(url, timeout=timeout_s) as resp: if resp.status != 200: log.warning( "mcube outbound: agent config fetch failed agent=%s status=%s", name, resp.status, ) return {} return await resp.json() except Exception: log.exception("mcube outbound: agent config fetch errored agent=%s", name) return {} async def _fetch_cluster_bot_mcube_config(business_id_val: Any, bot_id_val: Any) -> dict[str, Any]: if business_id_val in (None, "") or bot_id_val in (None, ""): return {} try: bid_int = int(business_id_val) bot_int = int(bot_id_val) except Exception: return {} url = f"{AGENT_BACKEND_BASE_URL}/api/agents/cluster/bots/{bid_int}/{bot_int}/mcube-config/" timeout_s = float(os.getenv("AGENT_BACKEND_FETCH_TIMEOUT_S", "5.0")) try: async with aiohttp.ClientSession() as session: async with session.get(url, timeout=timeout_s) as resp: if resp.status != 200: log.warning( "mcube outbound: cluster bot config fetch failed bid=%s bot_id=%s status=%s", bid_int, bot_int, resp.status, ) return {} return await resp.json() except Exception: log.exception("mcube outbound: cluster bot config fetch errored bid=%s bot_id=%s", bid_int, bot_int) return {} # If client provides refurl as an HTTP(S) "agent hint", extract agent_name from it. # Example: https:///agent/ refurl = str(body.get("refurl") or os.getenv("MCUBE_REFURL", "")).strip() if (not agent_name) and (refurl.startswith("http://") or refurl.startswith("https://")): parts = refurl.split("?", 1)[0].strip("/").split("/") try: idx = parts.index("agent") agent_name = parts[idx + 1].strip() except Exception: pass agent_cfg: dict[str, Any] = {} if agent_name: agent_cfg = await _fetch_agent_mcube_config(agent_name) cluster_bot_cfg: dict[str, Any] = await _fetch_cluster_bot_mcube_config(business_id, bot_id) # MCube docs use custnumber/exenumber. We accept `to` as an alias for custnumber. custnumber_in = (body.get("custnumber") or body.get("to") or "").strip() if not custnumber_in: return JSONResponse( {"ok": False, "error": "missing 'to' (or 'custnumber') value"}, status_code=400, ) # Internal correlation id we send as `refid` to MCube. # MCube uses refid (lead ID). We accept: # - call_id as alias for refid (used by WS bridge correlation) # - refid directly if provided refid = str(body.get("refid") or body.get("call_id") or uuid4()) # MCube required body fields. # Keep the leading '+' if the caller provided it (legacy live_calls uses +91...). custnumber = custnumber_in # Precedence: # 1) explicit request body # 2) Django agent config (if agent_name provided) # 3) env defaults exenumber_body = body.get("exenumber") gid_body = body.get("gid") exenumber_db = cluster_bot_cfg.get("mcube_exenumber") or agent_cfg.get("mcube_exenumber") gid_db = cluster_bot_cfg.get("mcube_gid") or agent_cfg.get("mcube_gid") exenumber = ( str(exenumber_body) if exenumber_body not in (None, "") else str(exenumber_db or os.getenv("MCUBE_EXENUMBER", "")).strip() ) gid = ( str(gid_body) if gid_body not in (None, "") else str(gid_db or os.getenv("MCUBE_GID", "1")).strip() ) # MCube expects `refurl` for websocket association. # We may receive an HTTP(S) hint here; later we'll convert to the WS url. # Auth token. # MCube examples often use HTTP_AUTHORIZATION in the JSON body; we support both. http_authorization = str( body.get("HTTP_AUTHORIZATION") or body.get("http_authorization") or body.get("httpAuthorization") or body.get("authorization") or body.get("Authorization") or request.headers.get("HTTP_AUTHORIZATION") or request.headers.get("Authorization") or os.getenv("HTTP_AUTHORIZATION", "") or os.getenv("MCUBE_HTTP_AUTHORIZATION", "") ).strip() # Persist per-call config, but we only know the websocket callId after MCube returns `called`. assert _redis is not None defaults = get_default_mcube_call_config() def _pick(*vals: Any) -> Any: for v in vals: if v is None: continue if isinstance(v, str) and v.strip() == "": continue return v return None # Precedence (low -> high): # env defaults -> cluster bot -> agent config -> request body call_config = dict(defaults) for k in call_config.keys(): call_config[k] = _pick(cluster_bot_cfg.get(k), agent_cfg.get(k), call_config.get(k)) # Per logged-in agent / tenant: `livekitvoicebot_cluster.business_id_agents` (+ JSON config). if business_id not in (None, ""): try: bid_int = int(business_id) except Exception: bid_int = None if bid_int is not None: bia = await asyncio.to_thread( get_runtime_overrides_from_agents_table, bid_int, agent_id=body.get("agent_id") or body.get("agentId"), user_id=body.get("user_id") or body.get("userId"), email=body.get("email") or body.get("agent_email"), name=body.get("agent_name") or body.get("name"), ) for bk, bv in bia.items(): if bv: call_config[bk] = bv # Pass through additional per-business/per-agent fields when present. # These are not required for the MCube runtime pipeline today, but we store them in Redis # so downstream components (or future features) can remain fully dynamic. for extra_key in ( "message_inbound", "message_outbound", "platform_settings", "conversation_behavior", ): ev = _pick( body.get(extra_key), cluster_bot_cfg.get(extra_key), agent_cfg.get(extra_key), bia.get(extra_key) if isinstance(bia, dict) else None, ) if ev is not None and (not isinstance(ev, str) or ev.strip() != ""): call_config[extra_key] = ev body_system_prompt = body.get("system_prompt") if isinstance(body_system_prompt, str): body_system_prompt = body_system_prompt.replace("\\n", "\n") call_config["system_prompt"] = _pick(body_system_prompt, call_config.get("system_prompt")) or call_config["system_prompt"] call_config["first_message"] = ( _pick( body.get("first_message"), body.get("agent_first_message"), call_config.get("first_message"), ) or "" ).strip() for k in [ "llm_model", "llm_provider", "stt_provider", "stt_language_code", "stt_model_id", "tts_provider", "tts_model", "tts_voice_id", "tts_encoding", "tts_chunk_ms", "tts_gain", "playback_pace_factor", "checkpoint_every", ]: call_config[k] = _pick(body.get(k), call_config.get(k)) or call_config[k] apply_voice_defaults_to_dict(call_config) # Correlation for ai_worker: load `business_id_agents` (cluster) by business + agent keys. for k in ("business_id", "bot_id", "agent_id", "user_id", "agent_name"): v = body.get(k) if v is not None and (not isinstance(v, str) or v.strip() != ""): call_config[k] = v if body.get("email") is not None and str(body.get("email") or "").strip() != "": call_config["agent_email"] = str(body.get("email")).strip() if body.get("agent_email") is not None and str(body.get("agent_email") or "").strip() != "": call_config["agent_email"] = str(body.get("agent_email")).strip() # If we're missing auth or exenumber, we can't initiate. But we still store under refid for local smoke tests. if not http_authorization or not exenumber: await _redis.set( f"mcube_call_config:{refid}", json.dumps(call_config).encode("utf-8"), ex=3600 * 6, ) return JSONResponse( { "ok": True, "call_id": refid, "mcube_call_sid": None, "status": "not_initiated", "warning": "MCube auth token and/or exenumber not set", "stored_config": True, } ) from .providers.mcube_provider import MCubeProvider provider = MCubeProvider(http_authorization=http_authorization) callback_url = ( (body.get("callback_url") or body.get("callbackUrl") or "").strip() or (f"{MCUBE_PUBLIC_BASE_URL}{MCUBE_WEBHOOK_PATH}" if MCUBE_PUBLIC_BASE_URL else "") ) websocket_url = ( (body.get("websocket_url") or body.get("websocketUrl") or "").strip() or ( f"{MCUBE_PUBLIC_WS_URL_BASE}{MCUBE_WS_PATH_PREFIX}/{refid}" if MCUBE_PUBLIC_WS_URL_BASE else "" ) ) # If refurl wasn't explicitly provided, default it to the websocket_url. # This matches the legacy `live_calls/homebook/services/make_calls.py` behavior, # where `refurl` points at the WS endpoint. # # If the client provides an HTTP(S) "refurl hint" (ex: `/agent/`), we compute the WS URL # because MCube expects ws/wss here. if websocket_url and (not refurl or refurl.startswith("http://") or refurl.startswith("https://")): refurl = websocket_url try: result = await provider.initiate_call( custnumber=custnumber, exenumber=exenumber, gid=gid, refurl=refurl, refid=refid, callback_url=callback_url or None, websocket_url=websocket_url or None, ) except Exception as e: # Make failures visible to the caller (instead of generic 500) so we can debug quickly. log.exception("mcube outbound: initiate_call failed call_id=%s", refid) return JSONResponse( {"ok": False, "error": "mcube_initiate_call_failed", "call_id": refid, "detail": str(e)}, status_code=502, ) # Store config keyed by MCube websocket callId (WS bridge uses it from URL). await _redis.set( f"mcube_call_config:{refid}", json.dumps(call_config).encode("utf-8"), ex=3600 * 6, ) # Also store under returned call sid (best-effort). await _redis.set( f"mcube_call_config:{result.call_sid}", json.dumps(call_config).encode("utf-8"), ex=3600 * 6, ) return JSONResponse( { "ok": True, "call_id": refid, "mcube_call_sid": result.call_sid, "status": result.status, "callback_url": callback_url, "websocket_url": websocket_url, "mcube_refurl": refurl, } ) if __name__ == "__main__": main() # import asyncio # import logging # import json # import os # import time # import aiohttp # from typing import Any # from urllib.parse import urlparse # import redis.asyncio as redis_async # from fastapi import FastAPI, Request # from fastapi.responses import JSONResponse # from .env_load import load_agent_runtime_dotenv # from .business_id_agents import get_runtime_overrides_from_agents_table # from .mcube_defaults import apply_voice_defaults_to_dict, get_default_mcube_call_config # from .service_log import configure_mcube_file_logging # log = logging.getLogger("mcube.webhook") # load_agent_runtime_dotenv() # REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0") # MCUBE_WEBHOOK_PATH = os.getenv("MCUBE_WEBHOOK_PATH", "/webhooks/mcube") # MCUBE_OUTBOUND_PATH = os.getenv("MCUBE_OUTBOUND_PATH", "/api/mcube/outbound-call") # MCUBE_OUTBOUND_PATH_BASE = MCUBE_OUTBOUND_PATH.rstrip("/") # def _mcube_public_urls() -> tuple[str, str]: # """ # Re-read backend/.env on each outbound call so ngrok URL updates apply without # restarting the webhook (stale MCUBE_PUBLIC_* was causing Mcube to open WSS to an old tunnel). # """ # load_agent_runtime_dotenv() # base = os.getenv("MCUBE_PUBLIC_BASE_URL", "").strip() # wss = os.getenv("MCUBE_PUBLIC_WS_URL_BASE", "").strip() # return base, wss # def _url_netloc(url: str) -> str: # u = (url or "").strip() # if not u: # return "" # try: # return (urlparse(u).netloc or "").lower() # except Exception: # return "" # MCUBE_WS_PATH_PREFIX = os.getenv("MCUBE_WS_PATH_PREFIX", "/bid/websocket") # # When set (e.g. 8028), refurl becomes wss://host/{id}/websocket/{refid} (matches proxy route). # MCUBE_WS_BUSINESS_ID = os.getenv("MCUBE_WS_BUSINESS_ID", "").strip() # # Backend (Django) base URL used to fetch per-agent MCube config. # # Example: http://localhost:8000 # AGENT_BACKEND_BASE_URL = os.getenv("AGENT_BACKEND_BASE_URL", "http://localhost:8000").rstrip("/") # APP = FastAPI(title="MCube Webhook Receiver") # _redis: redis_async.Redis | None = None # @APP.on_event("startup") # async def _startup() -> None: # global _redis # _redis = redis_async.from_url(REDIS_URL, decode_responses=False) # # Probe connection quickly # await _redis.ping() # log.info("mcube webhook: connected to redis %s", REDIS_URL) # @APP.get("/health") # async def health() -> dict[str, Any]: # return {"status": "ok"} # def _inspect_authorized(request: Request) -> bool: # """ # Protect /api/mcube/inspect/* — set MCUBE_INSPECT_KEY and send header X-Mcube-Inspect-Key. # """ # expected = os.getenv("MCUBE_INSPECT_KEY", "").strip() # if not expected: # return False # got = request.headers.get("X-Mcube-Inspect-Key", "").strip() # return got == expected # def _preview_cfg_field(val: Any, max_chars: int = 280) -> str: # if val is None: # return "" # s = str(val).replace("\\n", "\n").replace("\r\n", "\n") # s = s.replace("\n", " ").strip() # if len(s) <= max_chars: # return s # return s[: max_chars - 1] + "…" # def _summarize_stored_config(call_id: str, cfg: dict[str, Any]) -> dict[str, Any]: # sp = cfg.get("system_prompt") # if isinstance(sp, str): # sp = sp.replace("\\n", "\n") # fm = (cfg.get("first_message") or "").strip() # return { # "call_id": call_id, # "business_id": cfg.get("business_id"), # "bot_id": cfg.get("bot_id"), # "agent_id": cfg.get("agent_id"), # "agent_name": cfg.get("agent_name"), # "user_id": cfg.get("user_id"), # "agent_email": cfg.get("agent_email"), # "llm_model": cfg.get("llm_model"), # "llm_provider": cfg.get("llm_provider"), # "stt_provider": cfg.get("stt_provider"), # "tts_provider": cfg.get("tts_provider"), # "system_prompt_preview": _preview_cfg_field(sp, 280), # "first_message_preview": _preview_cfg_field(fm, 280), # } # @APP.get("/api/mcube/inspect/active") # async def inspect_active_mcube_sessions(request: Request) -> JSONResponse: # """ # Live PSTN legs where MCube sent `start` and we loaded Redis config (mcube_ws_active:*). # Requires X-Mcube-Inspect-Key matching MCUBE_INSPECT_KEY. # """ # if not _inspect_authorized(request): # return JSONResponse( # {"ok": False, "error": "unauthorized", "hint": "Set MCUBE_INSPECT_KEY and send X-Mcube-Inspect-Key"}, # status_code=401, # ) # redis = _redis # assert redis is not None # max_k = int(os.getenv("MCUBE_INSPECT_MAX_KEYS", "200")) # max_k = max(1, min(2000, max_k)) # rows: list[dict[str, Any]] = [] # n = 0 # try: # async for key in redis.scan_iter(match="mcube_ws_active:*"): # if n >= max_k: # break # n += 1 # ks = key.decode("utf-8", errors="replace") if isinstance(key, (bytes, bytearray)) else str(key) # try: # raw = await redis.get(key) # if not raw: # rows.append({"redis_key": ks, "error": "empty"}) # continue # raw_str = raw.decode("utf-8", errors="replace") if isinstance(raw, (bytes, bytearray)) else str(raw) # payload = json.loads(raw_str) # payload["redis_key"] = ks # rows.append(payload) # except Exception as e: # rows.append({"redis_key": ks, "error": str(e)}) # except Exception as e: # log.exception("inspect active failed") # return JSONResponse({"ok": False, "error": str(e)}, status_code=500) # return JSONResponse({"ok": True, "count": len(rows), "active": rows}) # @APP.get("/api/mcube/inspect/redis-configs") # async def inspect_redis_call_configs(request: Request) -> JSONResponse: # """ # Recent `mcube_call_config:*` blobs written at outbound-call time (TTL ~6h). Useful to see which # bot/prompt was stored before MCube connects — compare with /inspect/active when a call is up. # """ # if not _inspect_authorized(request): # return JSONResponse( # {"ok": False, "error": "unauthorized", "hint": "Set MCUBE_INSPECT_KEY and send X-Mcube-Inspect-Key"}, # status_code=401, # ) # redis = _redis # assert redis is not None # max_k = int(os.getenv("MCUBE_INSPECT_MAX_KEYS", "200")) # max_k = max(1, min(2000, max_k)) # rows: list[dict[str, Any]] = [] # n = 0 # try: # async for key in redis.scan_iter(match="mcube_call_config:*"): # if n >= max_k: # break # n += 1 # ks = key.decode("utf-8", errors="replace") if isinstance(key, (bytes, bytearray)) else str(key) # suffix = ks.split("mcube_call_config:", 1)[-1] # try: # raw = await redis.get(key) # if not raw: # rows.append({"redis_key": ks, "call_id": suffix, "error": "empty"}) # continue # raw_str = raw.decode("utf-8", errors="replace") if isinstance(raw, (bytes, bytearray)) else str(raw) # cfg = json.loads(raw_str) # if not isinstance(cfg, dict): # rows.append({"redis_key": ks, "call_id": suffix, "error": "not_a_dict"}) # continue # row = _summarize_stored_config(suffix, cfg) # row["redis_key"] = ks # rows.append(row) # except Exception as e: # rows.append({"redis_key": ks, "call_id": suffix, "error": str(e)}) # except Exception as e: # log.exception("inspect redis-configs failed") # return JSONResponse({"ok": False, "error": str(e)}, status_code=500) # return JSONResponse({"ok": True, "count": len(rows), "stored_configs": rows}) # @APP.post(MCUBE_WEBHOOK_PATH) # async def mcube_webhook(request: Request) -> JSONResponse: # try: # payload: dict[str, Any] = await request.json() # call_sid = payload.get("call_id") or payload.get("callId") or payload.get("callID") # status = str(payload.get("status", "")).lower() # duration = payload.get("duration", 0) # answered_by = payload.get("answered_by", "human") # if not call_sid: # return JSONResponse({"ok": False, "error": "missing call_id"}, status_code=400) # redis = _redis # assert redis is not None # idem_key = f"mcube_webhook_processed:{call_sid}:{status}" # # Idempotency guard: skip duplicates for the same call+status. # already = await redis.set(idem_key, 1, nx=True, ex=30) # if not already: # return JSONResponse({"ok": True, "skipped": True}) # # Store latest status for use by other components. # await redis.set( # f"mcube_call_status:{call_sid}", # status.encode("utf-8") if isinstance(status, str) else status, # ex=3600 * 24, # ) # await redis.set( # f"mcube_call_duration:{call_sid}", # str(duration).encode("utf-8") if isinstance(duration, (int, float)) else str(duration).encode("utf-8"), # ex=3600 * 24, # ) # await redis.set( # f"mcube_call_answered_by:{call_sid}", # str(answered_by).encode("utf-8"), # ex=3600 * 24, # ) # # Convenience “ended” marker for terminal statuses. # terminal = status in {"not_answered", "failed", "voicemail", "blocked", "completed", "no-answer", "busy"} # if terminal: # await redis.set(f"mcube_call_ended:{call_sid}", 1, ex=3600 * 6) # return JSONResponse({"ok": True}) # except Exception as e: # log.exception("mcube webhook failed") # return JSONResponse( # {"ok": False, "error": "mcube_webhook_failed", "detail": str(e)}, # status_code=500, # ) # def main() -> None: # logging.basicConfig(level=logging.INFO) # configure_mcube_file_logging("webhook") # import uvicorn # host = os.getenv("MCUBE_WEBHOOK_HOST", "0.0.0.0") # port = int(os.getenv("MCUBE_WEBHOOK_PORT", "8002")) # uvicorn.run(APP, host=host, port=port) # async def _outbound_call_handler( # request: Request, # *, # path_business_id: int | None = None, # path_bot_id: int | None = None, # ) -> JSONResponse: # """ # Minimal endpoint to kick off an outbound MCube click-to-call. # This integrates with MCube's Restmcube-api/outbound-calls endpoint which expects: # - HTTP header: Authorization # - JSON body keys: custnumber, exenumber, gid, refurl, refid (as per your doc) # URL forms (path overrides business_id / bot_id in JSON when set): # - POST /api/mcube/outbound-call/{business_id}/{bot_id} # - POST /api/mcube/outbound-call/{bot_id} (legacy: bot in path, business_id in body) # - POST /api/mcube/outbound-call (all identifiers in body) # Body examples supported: # - { "to": "+1555..." , "exenumber": "8700...", "gid": "1", "call_id": "optional" } # - or use env defaults for exenumber/gid/auth. # """ # from uuid import uuid4 # body = await request.json() # # Optional: if provided, we use Django agent config (mcube_exenumber/mcube_gid) # # instead of relying purely on request body/env defaults. # agent_name = str( # body.get("agent_name") # or body.get("agentName") # or body.get("agent") # or "" # ).strip() # business_id = body.get("business_id") or body.get("businessId") or body.get("bid") # bot_id = body.get("bot_id") or body.get("botId") # if path_business_id is not None: # business_id = path_business_id # if path_bot_id is not None: # bot_id = path_bot_id # fetch_timeout_s = float(os.getenv("AGENT_BACKEND_FETCH_TIMEOUT_S", "15.0")) # async def _fetch_agent_mcube_config(name: str) -> dict[str, Any]: # if not name: # return {} # url = f"{AGENT_BACKEND_BASE_URL}/api/agents/{name}/config/" # timeout = aiohttp.ClientTimeout(total=fetch_timeout_s) # try: # async with aiohttp.ClientSession(timeout=timeout) as session: # async with session.get(url) as resp: # if resp.status != 200: # log.warning( # "mcube outbound: agent config fetch failed agent=%s status=%s", # name, # resp.status, # ) # return {} # return await resp.json() # except Exception: # log.exception("mcube outbound: agent config fetch errored agent=%s", name) # return {} # async def _fetch_cluster_bot_mcube_config(business_id_val: Any, bot_id_val: Any) -> dict[str, Any]: # if business_id_val in (None, "") or bot_id_val in (None, ""): # return {} # try: # bid_int = int(business_id_val) # bot_int = int(bot_id_val) # except Exception: # return {} # url = f"{AGENT_BACKEND_BASE_URL}/api/agents/cluster/bots/{bid_int}/{bot_int}/mcube-config/" # timeout = aiohttp.ClientTimeout(total=fetch_timeout_s) # try: # async with aiohttp.ClientSession(timeout=timeout) as session: # async with session.get(url) as resp: # if resp.status != 200: # log.warning( # "mcube outbound: cluster bot config fetch failed bid=%s bot_id=%s status=%s", # bid_int, # bot_int, # resp.status, # ) # return {} # return await resp.json() # except Exception: # log.exception("mcube outbound: cluster bot config fetch errored bid=%s bot_id=%s", bid_int, bot_int) # return {} # # If client provides refurl as an HTTP(S) "agent hint", extract agent_name from it. # # Example: https:///agent/ # refurl = str(body.get("refurl") or os.getenv("MCUBE_REFURL", "")).strip() # if (not agent_name) and (refurl.startswith("http://") or refurl.startswith("https://")): # parts = refurl.split("?", 1)[0].strip("/").split("/") # try: # idx = parts.index("agent") # agent_name = parts[idx + 1].strip() # except Exception: # pass # # MCube docs use custnumber/exenumber. We accept `to` as an alias for custnumber. # custnumber_in = (body.get("custnumber") or body.get("to") or "").strip() # if not custnumber_in: # return JSONResponse( # {"ok": False, "error": "missing 'to' (or 'custnumber') value"}, # status_code=400, # ) # # Fetch Django configs in parallel — sequential HTTP timeouts stacked to ~10s and blocked # # Redis+MCube initiate until done ("call dies after ~5–6s" when PSTN connected early). # async def _agent_cfg_wrapped() -> dict[str, Any]: # if not agent_name or str(agent_name).strip().lower() in ("default", ""): # return {} # return await _fetch_agent_mcube_config(agent_name) # outbound_prep_t0 = time.perf_counter() # agent_cfg, cluster_bot_cfg = await asyncio.gather( # _agent_cfg_wrapped(), # _fetch_cluster_bot_mcube_config(business_id, bot_id), # ) # log.info( # "mcube outbound: django config fetch done ms=%.0f agent_keys=%s cluster_keys=%s", # (time.perf_counter() - outbound_prep_t0) * 1000.0, # len(agent_cfg or {}), # len(cluster_bot_cfg or {}), # ) # # Internal correlation id we send as `refid` to MCube. # # MCube uses refid (lead ID). We accept: # # - call_id as alias for refid (used by WS bridge correlation) # # - refid directly if provided # refid = str(body.get("refid") or body.get("call_id") or uuid4()) # # MCube required body fields. # # Keep the leading '+' if the caller provided it (legacy live_calls uses +91...). # custnumber = custnumber_in # # Precedence: # # 1) explicit request body # # 2) Django agent config (if agent_name provided) # # 3) env defaults # exenumber_body = body.get("exenumber") # gid_body = body.get("gid") # exenumber_db = cluster_bot_cfg.get("mcube_exenumber") or agent_cfg.get("mcube_exenumber") # gid_db = cluster_bot_cfg.get("mcube_gid") or agent_cfg.get("mcube_gid") # exenumber = ( # str(exenumber_body) # if exenumber_body not in (None, "") # else str(exenumber_db or os.getenv("MCUBE_EXENUMBER", "")).strip() # ) # gid = ( # str(gid_body) # if gid_body not in (None, "") # else str(gid_db or os.getenv("MCUBE_GID", "1")).strip() # ) # # MCube expects `refurl` for websocket association. # # We may receive an HTTP(S) hint here; later we'll convert to the WS url. # # Auth token. # # MCube examples often use HTTP_AUTHORIZATION in the JSON body; we support both. # http_authorization = str( # body.get("HTTP_AUTHORIZATION") # or body.get("http_authorization") # or body.get("httpAuthorization") # or body.get("authorization") # or body.get("Authorization") # or request.headers.get("HTTP_AUTHORIZATION") # or request.headers.get("Authorization") # or os.getenv("HTTP_AUTHORIZATION", "") # or os.getenv("MCUBE_HTTP_AUTHORIZATION", "") # ).strip() # # Persist per-call config, but we only know the websocket callId after MCube returns `called`. # assert _redis is not None # defaults = get_default_mcube_call_config() # def _pick(*vals: Any) -> Any: # for v in vals: # if v is None: # continue # if isinstance(v, str) and v.strip() == "": # continue # return v # return None # # Precedence (low -> high): # # env defaults -> cluster bot -> agent config -> request body # call_config = dict(defaults) # for k in call_config.keys(): # call_config[k] = _pick(cluster_bot_cfg.get(k), agent_cfg.get(k), call_config.get(k)) # # Per logged-in agent / tenant: `livekitvoicebot_cluster.business_id_agents` (+ JSON config). # if business_id not in (None, ""): # try: # bid_int = int(business_id) # except Exception: # bid_int = None # if bid_int is not None: # bia = await asyncio.to_thread( # get_runtime_overrides_from_agents_table, # bid_int, # agent_id=body.get("agent_id") or body.get("agentId"), # user_id=body.get("user_id") or body.get("userId"), # email=body.get("email") or body.get("agent_email"), # name=body.get("agent_name") or body.get("name"), # ) # for bk, bv in bia.items(): # if bv: # call_config[bk] = bv # # Pass through additional per-business/per-agent fields when present. # # These are not required for the MCube runtime pipeline today, but we store them in Redis # # so downstream components (or future features) can remain fully dynamic. # for extra_key in ( # "message_inbound", # "message_outbound", # "platform_settings", # "conversation_behavior", # ): # ev = _pick( # body.get(extra_key), # cluster_bot_cfg.get(extra_key), # agent_cfg.get(extra_key), # bia.get(extra_key) if isinstance(bia, dict) else None, # ) # if ev is not None and (not isinstance(ev, str) or ev.strip() != ""): # call_config[extra_key] = ev # body_system_prompt = body.get("system_prompt") # if isinstance(body_system_prompt, str): # body_system_prompt = body_system_prompt.replace("\\n", "\n") # call_config["system_prompt"] = _pick(body_system_prompt, call_config.get("system_prompt")) or call_config["system_prompt"] # call_config["first_message"] = ( # _pick( # body.get("first_message"), # body.get("agent_first_message"), # call_config.get("first_message"), # ) # or "" # ).strip() # for k in [ # "llm_model", # "llm_provider", # "stt_provider", # "stt_language_code", # "stt_model_id", # "tts_provider", # "tts_model", # "tts_voice_id", # "tts_encoding", # "tts_chunk_ms", # "tts_gain", # "playback_pace_factor", # "checkpoint_every", # ]: # call_config[k] = _pick(body.get(k), call_config.get(k)) or call_config[k] # # Correlation for ai_worker: load `business_id_agents` (cluster) by business + agent keys. # def _cfg_nonempty(val: Any) -> bool: # if val is None: # return False # if isinstance(val, str) and val.strip() == "": # return False # return True # if _cfg_nonempty(business_id): # call_config["business_id"] = business_id # if _cfg_nonempty(bot_id): # call_config["bot_id"] = bot_id # for k in ("agent_id", "user_id", "agent_name"): # v = body.get(k) # if v is not None and (not isinstance(v, str) or v.strip() != ""): # call_config[k] = v # if body.get("email") is not None and str(body.get("email") or "").strip() != "": # call_config["agent_email"] = str(body.get("email")).strip() # if body.get("agent_email") is not None and str(body.get("agent_email") or "").strip() != "": # call_config["agent_email"] = str(body.get("agent_email")).strip() # apply_voice_defaults_to_dict(call_config) # # If we're missing auth or exenumber, we can't initiate. But we still store under refid for local smoke tests. # if not http_authorization or not exenumber: # await _redis.set( # f"mcube_call_config:{refid}", # json.dumps(call_config).encode("utf-8"), # ex=3600 * 6, # ) # return JSONResponse( # { # "ok": True, # "call_id": refid, # "mcube_call_sid": None, # "status": "not_initiated", # "warning": "MCube auth token and/or exenumber not set", # "stored_config": True, # } # ) # from .providers.mcube_provider import MCubeProvider # provider = MCubeProvider(http_authorization=http_authorization) # pub_base, pub_wss = _mcube_public_urls() # expected_http_host = _url_netloc(pub_base) # expected_wss_host = _url_netloc(pub_wss) # body_cb = (body.get("callback_url") or body.get("callbackUrl") or "").strip() # if body_cb and expected_http_host and _url_netloc(body_cb) != expected_http_host: # log.warning( # "mcube outbound: ignoring stale client callback_url host=%r (MCUBE_PUBLIC_BASE_URL host=%r)", # _url_netloc(body_cb), # expected_http_host, # ) # body_cb = "" # callback_url = body_cb or (f"{pub_base.rstrip('/')}{MCUBE_WEBHOOK_PATH}" if pub_base else "") # ws_base = pub_wss.rstrip("/") if pub_wss else "" # # Only use /{bid}/websocket/... when explicitly configured (env or body). Do not infer from # # generic `business_id` or outbound calls would silently change path vs MCUBE_WS_PATH_PREFIX. # bid_ws = ( # MCUBE_WS_BUSINESS_ID # or str(body.get("mcube_ws_business_id") or body.get("mcubeWsBusinessId") or "").strip() # ) # body_ws = (body.get("websocket_url") or body.get("websocketUrl") or "").strip() # if body_ws and expected_wss_host and _url_netloc(body_ws) != expected_wss_host: # log.warning( # "mcube outbound: ignoring stale client websocket_url host=%r (MCUBE_PUBLIC_WS_URL_BASE host=%r)", # _url_netloc(body_ws), # expected_wss_host, # ) # body_ws = "" # websocket_url = body_ws.strip() # if not websocket_url and ws_base: # if bid_ws and str(bid_ws).isdigit(): # websocket_url = f"{ws_base}/{bid_ws}/websocket/{refid}" # else: # websocket_url = f"{ws_base}{MCUBE_WS_PATH_PREFIX}/{refid}" # # If refurl wasn't explicitly provided, default it to the websocket_url. # # This matches the legacy `live_calls/homebook/services/make_calls.py` behavior, # # where `refurl` points at the WS endpoint. # # # # If the client provides an HTTP(S) "refurl hint" (ex: `/agent/`), we compute the WS URL # # because MCube expects ws/wss here. # # # # If body / MCUBE_REFURL still has wss:// with a *different* host than MCUBE_PUBLIC_WS_URL_BASE # # (stale ngrok), keep callback/websocket aligned but refurl was slipping through — replace it. # if websocket_url: # ru = (refurl or "").strip() # if not ru or ru.startswith(("http://", "https://")): # refurl = websocket_url # else: # ru_host = _url_netloc(ru) # if expected_wss_host and ru_host and ru_host != expected_wss_host: # log.warning( # "mcube outbound: ignoring stale refurl host=%r (MCUBE_PUBLIC_WS_URL_BASE host=%r)", # ru_host, # expected_wss_host, # ) # refurl = websocket_url # # Store before MCube initiate so the WS bridge never hits an empty Redis key if the # # carrier connects quickly (avoids "no mcube_call_config" + slow greeting → PSTN timeout). # cfg_blob = json.dumps(call_config).encode("utf-8") # await _redis.set(f"mcube_call_config:{refid}", cfg_blob, ex=3600 * 6) # log.info( # "mcube outbound: calling MCube initiate refid=%s prep_since_fetch_ms=%.0f", # refid, # (time.perf_counter() - outbound_prep_t0) * 1000.0, # ) # try: # result = await provider.initiate_call( # custnumber=custnumber, # exenumber=exenumber, # gid=gid, # refurl=refurl, # refid=refid, # callback_url=callback_url or None, # websocket_url=websocket_url or None, # ) # except Exception as e: # # Make failures visible to the caller (instead of generic 500) so we can debug quickly. # log.exception("mcube outbound: initiate_call failed call_id=%s", refid) # return JSONResponse( # {"ok": False, "error": "mcube_initiate_call_failed", "call_id": refid, "detail": str(e)}, # status_code=502, # ) # # Refresh TTL and duplicate under MCube call id when it differs from refid (WS URL may use either). # await _redis.set(f"mcube_call_config:{refid}", cfg_blob, ex=3600 * 6) # if result.call_sid: # await _redis.set(f"mcube_call_config:{result.call_sid}", cfg_blob, ex=3600 * 6) # st = str(result.status or "").lower() # mcube_sid = result.call_sid or None # resp: dict[str, Any] = { # "ok": True, # "call_id": refid, # "mcube_call_sid": mcube_sid, # "status": result.status, # "callback_url": callback_url, # "websocket_url": websocket_url, # "mcube_refurl": refurl, # } # if mcube_sid is None and st and "succ" not in st: # resp["warning"] = ( # "MCube did not return a call id and status does not look successful; " # "check mcube outbound logs (full response body) and the destination number." # ) # elif mcube_sid is None: # resp["warning"] = ( # "MCube did not return a callid in the response; outbound may not have been queued. " # "See mcube.provider logs for the raw body." # ) # elif st and "succ" not in st and "success" not in st: # resp["warning"] = ( # f"MCube status was {result.status!r}; confirm in MCube dashboard whether the call was placed." # ) # return JSONResponse(resp) # @APP.post(f"{MCUBE_OUTBOUND_PATH_BASE}/{{business_id}}/{{bot_id}}") # async def outbound_call_scoped(business_id: int, bot_id: int, request: Request) -> JSONResponse: # return await _outbound_call_handler(request, path_business_id=business_id, path_bot_id=bot_id) # @APP.post(f"{MCUBE_OUTBOUND_PATH_BASE}/{{legacy_path_bot_id}}") # async def outbound_call_legacy_path_bot(legacy_path_bot_id: int, request: Request) -> JSONResponse: # return await _outbound_call_handler(request, path_bot_id=legacy_path_bot_id) # @APP.post(MCUBE_OUTBOUND_PATH) # async def outbound_call(request: Request) -> JSONResponse: # return await _outbound_call_handler(request) # if __name__ == "__main__": # main()