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 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 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] # # 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()