diff --git a/api/services/integrations/paygent/__init__.py b/api/services/integrations/paygent/__init__.py index 79394ca1..a7229b34 100644 --- a/api/services/integrations/paygent/__init__.py +++ b/api/services/integrations/paygent/__init__.py @@ -10,6 +10,7 @@ Provides: - ``create_runtime_sessions`` – live-call observer that accumulates usage data - ``run_completion`` – post-call REST delivery to the Paygent API """ + from __future__ import annotations from api.services.integrations.base import IntegrationPackageSpec diff --git a/api/services/integrations/paygent/client.py b/api/services/integrations/paygent/client.py index e39e3263..6bae33fa 100644 --- a/api/services/integrations/paygent/client.py +++ b/api/services/integrations/paygent/client.py @@ -5,13 +5,13 @@ coroutine used by the completion handler. The individual tracker functions (session, STT, TTS, LLM, STS, indicator) mirror the exact shape of the Paygent REST API documented in ``paygent_sdk/voice_client.py``. """ + from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Any import httpx -from loguru import logger from pydantic import BaseModel, field_validator _DEFAULT_BASE_URL = "https://cp-api.withpaygent.com" @@ -169,7 +169,6 @@ async def deliver( errors: list[str] = [] async with httpx.AsyncClient(timeout=_REQUEST_TIMEOUT) as client: - # 1. Initialize voice session ---------------------------------------- try: await _post( @@ -256,8 +255,14 @@ async def deliver( metadata = snapshot.sts_usage_metadata or {} # Only append connection minutes if we don't already have a rich token payload # (e.g. from OpenAI Realtime or Gemini Live) - if "connection" not in metadata and "prompt_tokens" not in metadata and "input" not in metadata: - metadata["connection"] = {"minutes": snapshot.total_duration_seconds / 60.0} + if ( + "connection" not in metadata + and "prompt_tokens" not in metadata + and "input" not in metadata + ): + metadata["connection"] = { + "minutes": snapshot.total_duration_seconds / 60.0 + } try: await _post( diff --git a/api/services/integrations/paygent/collector.py b/api/services/integrations/paygent/collector.py index bf8473fc..d83e26cc 100644 --- a/api/services/integrations/paygent/collector.py +++ b/api/services/integrations/paygent/collector.py @@ -10,6 +10,7 @@ Design mirrors ``api/services/integrations/tuner/collector.py`` exactly: - Build a serialisable snapshot in ``build_snapshot`` - Return it from ``on_call_finished`` so it lands in ``workflow_run.logs`` """ + from __future__ import annotations import time @@ -43,15 +44,22 @@ def _detect_provider(name: str, fallback: str = "unknown") -> str: if "gemini" in clean_name: return "google" suffixes = [ - "service", "multimodallive", "realtime", - "vertex", "llm", "tts", "stt", "helper", "transport" + "service", + "multimodallive", + "realtime", + "vertex", + "llm", + "tts", + "stt", + "helper", + "transport", ] changed = True while changed: changed = False for suffix in suffixes: if clean_name.endswith(suffix): - clean_name = clean_name[:-len(suffix)].rstrip("_").rstrip("-") + clean_name = clean_name[: -len(suffix)].rstrip("_").rstrip("-") changed = True break return clean_name or fallback @@ -81,13 +89,13 @@ class _UsageAccumulator: call_end_abs_ns: int | None = None # STT: timestamp of when user started speaking; None when not speaking _user_started_speaking_ns: int | None = field(default=None, repr=False) - + @property def total_duration_seconds(self) -> int: if self.call_end_abs_ns is None: return int((time.time_ns() - self.call_start_abs_ns) / 1_000_000_000) return int((self.call_end_abs_ns - self.call_start_abs_ns) / 1_000_000_000) - + def get_stt_audio_seconds(self) -> float: """Return measured STT audio seconds accumulated from the pipeline. @@ -109,7 +117,7 @@ class _UsageAccumulator: if not self._has_tts_metrics: self._has_tts_metrics = True self.tts_characters = 0 # Ignore manual count if metrics emit natively - + # Extremely robust extraction val = 0 if isinstance(data, (int, float)): @@ -120,11 +128,13 @@ class _UsageAccumulator: val = getattr(data, "characters", 0) or 0 elif isinstance(data, dict): val = data.get("value") or data.get("characters") or 0 - + try: self.tts_characters += int(val or 0) except Exception as exc: - logger.warning("[paygent] Failed to accumulate TTS characters (val={!r}): {}", val, exc) + logger.warning( + "[paygent] Failed to accumulate TTS characters (val={!r}): {}", val, exc + ) def add_tts_manual(self, text: str) -> None: if not self._has_tts_metrics: @@ -138,7 +148,9 @@ class _UsageAccumulator: def on_user_stopped_speaking(self) -> None: """Accumulate the completed utterance duration into stt_audio_seconds.""" if self._user_started_speaking_ns is not None: - elapsed_s = (time.time_ns() - self._user_started_speaking_ns) / 1_000_000_000 + elapsed_s = ( + time.time_ns() - self._user_started_speaking_ns + ) / 1_000_000_000 self.stt_audio_seconds += elapsed_s self._user_started_speaking_ns = None @@ -162,9 +174,11 @@ def _google_live_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]: return None for k in keys: if isinstance(obj, dict): - if k in obj: return obj[k] + if k in obj: + return obj[k] else: - if hasattr(obj, k): return getattr(obj, k) + if hasattr(obj, k): + return getattr(obj, k) return None def _get_list(obj, *keys): @@ -202,20 +216,36 @@ def _google_live_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]: return total prompt_details = _get_list(usage, "prompt_tokens_details", "promptTokensDetails") - response_details = _get_list(usage, "response_tokens_details", "responseTokensDetails") - tool_details = _get_list(usage, "tool_use_prompt_tokens_details", "toolUsePromptTokensDetails") + response_details = _get_list( + usage, "response_tokens_details", "responseTokensDetails" + ) + tool_details = _get_list( + usage, "tool_use_prompt_tokens_details", "toolUsePromptTokensDetails" + ) cache_details = _get_list(usage, "cache_tokens_details", "cacheTokensDetails") # input side: TEXT + DOCUMENT + AUDIO + IMAGE + VIDEO - text_in = _modality_token_count(prompt_details, "TEXT") + _modality_token_count(tool_details, "TEXT") - audio_in = _modality_token_count(prompt_details, "AUDIO") + _modality_token_count(tool_details, "AUDIO") - image_in = _modality_token_count(prompt_details, "IMAGE") + _modality_token_count(tool_details, "IMAGE") - video_in = _modality_token_count(prompt_details, "VIDEO") + _modality_token_count(tool_details, "VIDEO") - doc_as_text = _modality_token_count(prompt_details, "DOCUMENT") + _modality_token_count(tool_details, "DOCUMENT") + text_in = _modality_token_count(prompt_details, "TEXT") + _modality_token_count( + tool_details, "TEXT" + ) + audio_in = _modality_token_count(prompt_details, "AUDIO") + _modality_token_count( + tool_details, "AUDIO" + ) + image_in = _modality_token_count(prompt_details, "IMAGE") + _modality_token_count( + tool_details, "IMAGE" + ) + video_in = _modality_token_count(prompt_details, "VIDEO") + _modality_token_count( + tool_details, "VIDEO" + ) + doc_as_text = _modality_token_count( + prompt_details, "DOCUMENT" + ) + _modality_token_count(tool_details, "DOCUMENT") text_in += doc_as_text # fallback aggregate mapping - tutc = _optional_int(usage, "tool_use_prompt_token_count", "toolUsePromptTokenCount") + tutc = _optional_int( + usage, "tool_use_prompt_token_count", "toolUsePromptTokenCount" + ) if tutc is not None and not tool_details: text_in += int(tutc) @@ -224,8 +254,12 @@ def _google_live_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]: text_in += int(ptc) # output side: TEXT + DOCUMENT + AUDIO + VIDEO + THINKING - text_out = _modality_token_count(response_details, "TEXT") + _modality_token_count(response_details, "DOCUMENT") - audio_out = _modality_token_count(response_details, "AUDIO") + _modality_token_count(response_details, "VIDEO") + text_out = _modality_token_count(response_details, "TEXT") + _modality_token_count( + response_details, "DOCUMENT" + ) + audio_out = _modality_token_count( + response_details, "AUDIO" + ) + _modality_token_count(response_details, "VIDEO") rtc = _optional_int(usage, "response_token_count", "responseTokenCount") if text_out == 0 and audio_out == 0 and rtc is not None: @@ -234,35 +268,55 @@ def _google_live_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]: # Thinking / reasoning tokens (Gemini 2.5+ thinking models). # Emitted as a separate output modality so Paygent has full billing visibility. - thinking_tokens = _optional_int( - usage, - "thoughts_token_count", "thoughtsTokenCount", - "thinking_token_count", "thinkingTokenCount", - ) or 0 + thinking_tokens = ( + _optional_int( + usage, + "thoughts_token_count", + "thoughtsTokenCount", + "thinking_token_count", + "thinkingTokenCount", + ) + or 0 + ) # Cache breakdowns - cached_text = _modality_token_count(cache_details, "TEXT") + _modality_token_count(cache_details, "DOCUMENT") - cached_audio = _modality_token_count(cache_details, "AUDIO") + _modality_token_count(cache_details, "VIDEO") + cached_text = _modality_token_count(cache_details, "TEXT") + _modality_token_count( + cache_details, "DOCUMENT" + ) + cached_audio = _modality_token_count( + cache_details, "AUDIO" + ) + _modality_token_count(cache_details, "VIDEO") cached_image = _modality_token_count(cache_details, "IMAGE") - cached_legacy = _optional_int(usage, "cached_content_token_count", "cachedContentTokenCount") + cached_legacy = _optional_int( + usage, "cached_content_token_count", "cachedContentTokenCount" + ) # Build response payload out = {"schemaVersion": 1} # Input Side inp = {} - if text_in > 0: inp["text"] = {"tokens": text_in} - if audio_in > 0: inp["audio"] = {"tokens": audio_in} - if image_in > 0: inp["image"] = {"tokens": image_in} - if video_in > 0: inp["video"] = {"tokens": video_in} - if inp: out["input"] = inp + if text_in > 0: + inp["text"] = {"tokens": text_in} + if audio_in > 0: + inp["audio"] = {"tokens": audio_in} + if image_in > 0: + inp["image"] = {"tokens": image_in} + if video_in > 0: + inp["video"] = {"tokens": video_in} + if inp: + out["input"] = inp # Output Side o = {} - if text_out > 0: o["text"] = {"tokens": text_out} - if audio_out > 0: o["audio"] = {"tokens": audio_out} - if thinking_tokens > 0: o["thinking"] = {"tokens": thinking_tokens} - if o: out["output"] = o + if text_out > 0: + o["text"] = {"tokens": text_out} + if audio_out > 0: + o["audio"] = {"tokens": audio_out} + if thinking_tokens > 0: + o["thinking"] = {"tokens": thinking_tokens} + if o: + out["output"] = o # Cached breakdown has_split = bool(cached_text or cached_audio or cached_image) @@ -270,15 +324,18 @@ def _google_live_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]: out["cached"] = {"tokens": int(cached_legacy)} elif has_split: cd = {} - if cached_text > 0: cd["text"] = {"tokens": cached_text} - if cached_audio > 0: cd["audio"] = {"tokens": cached_audio} - if cached_image > 0: cd["image"] = {"tokens": cached_image} - if cd: out["cached"] = cd + if cached_text > 0: + cd["text"] = {"tokens": cached_text} + if cached_audio > 0: + cd["audio"] = {"tokens": cached_audio} + if cached_image > 0: + cd["image"] = {"tokens": cached_image} + if cd: + out["cached"] = cd return out - def _openai_realtime_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]: """ Pure Python translation of OpenAI Realtime usage_metadata to @@ -292,9 +349,11 @@ def _openai_realtime_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, A return None for k in keys: if isinstance(obj, dict): - if k in obj: return obj[k] + if k in obj: + return obj[k] else: - if hasattr(obj, k): return getattr(obj, k) + if hasattr(obj, k): + return getattr(obj, k) return None total_in = int(_get_val(usage, "input_tokens", "inputTokens") or 0) @@ -307,17 +366,29 @@ def _openai_realtime_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, A text_in = int(_get_val(in_details, "text_tokens", "textTokens") or 0) image_in = int(_get_val(in_details, "image_tokens", "imageTokens") or 0) - cached_total = int(_get_val(usage, "cached_tokens", "cachedTokens") or _get_val(in_details, "cached_tokens", "cachedTokens") or 0) + cached_total = int( + _get_val(usage, "cached_tokens", "cachedTokens") + or _get_val(in_details, "cached_tokens", "cachedTokens") + or 0 + ) - cached_details = _get_val(in_details, "cached_tokens_details", "cachedTokensDetails") or {} + cached_details = ( + _get_val(in_details, "cached_tokens_details", "cachedTokensDetails") or {} + ) cached_audio = int(_get_val(cached_details, "audio_tokens", "audioTokens") or 0) cached_text = int(_get_val(cached_details, "text_tokens", "textTokens") or 0) cached_image = int(_get_val(cached_details, "image_tokens", "imageTokens") or 0) if not (cached_audio or cached_text or cached_image): - cached_audio = int(_get_val(in_details, "cached_audio_tokens", "cachedAudioTokens") or 0) - cached_text = int(_get_val(in_details, "cached_text_tokens", "cachedTextTokens") or 0) - cached_image = int(_get_val(in_details, "cached_image_tokens", "cachedImageTokens") or 0) + cached_audio = int( + _get_val(in_details, "cached_audio_tokens", "cachedAudioTokens") or 0 + ) + cached_text = int( + _get_val(in_details, "cached_text_tokens", "cachedTextTokens") or 0 + ) + cached_image = int( + _get_val(in_details, "cached_image_tokens", "cachedImageTokens") or 0 + ) audio_out = int(_get_val(out_details, "audio_tokens", "audioTokens") or 0) text_out = int(_get_val(out_details, "text_tokens", "textTokens") or 0) @@ -327,25 +398,36 @@ def _openai_realtime_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, A out = {"schemaVersion": 1} inp = {} - if text_in > 0: inp["text"] = {"tokens": text_in} - if audio_in > 0: inp["audio"] = {"tokens": audio_in} - if image_in > 0: inp["image"] = {"tokens": image_in} - if inp: out["input"] = inp + if text_in > 0: + inp["text"] = {"tokens": text_in} + if audio_in > 0: + inp["audio"] = {"tokens": audio_in} + if image_in > 0: + inp["image"] = {"tokens": image_in} + if inp: + out["input"] = inp o = {} - if text_out > 0: o["text"] = {"tokens": text_out} - if audio_out > 0: o["audio"] = {"tokens": audio_out} - if o: out["output"] = o + if text_out > 0: + o["text"] = {"tokens": text_out} + if audio_out > 0: + o["audio"] = {"tokens": audio_out} + if o: + out["output"] = o has_split = bool(cached_text or cached_audio or cached_image) if cached_total > 0 and not has_split: out["cached"] = {"tokens": int(cached_total)} elif has_split: cd = {} - if cached_text > 0: cd["text"] = {"tokens": cached_text} - if cached_audio > 0: cd["audio"] = {"tokens": cached_audio} - if cached_image > 0: cd["image"] = {"tokens": cached_image} - if cd: out["cached"] = cd + if cached_text > 0: + cd["text"] = {"tokens": cached_text} + if cached_audio > 0: + cd["audio"] = {"tokens": cached_audio} + if cached_image > 0: + cd["image"] = {"tokens": cached_image} + if cd: + out["cached"] = cd return out @@ -359,14 +441,18 @@ def _merge_sts_metadata(existing: dict, new: dict) -> dict: n_val = new.get(key, {}) if not e_val and not n_val: continue - + merged_cat: dict = {} # Prefer per-modality merge when either side has per-modality detail. # Only use the flat aggregate{"tokens": N} form when neither side has # any per-modality breakdown at all (e.g. legacy schema). - e_has_modalities = any(m in e_val for m in ("text", "audio", "image", "video", "thinking")) - n_has_modalities = any(m in n_val for m in ("text", "audio", "image", "video", "thinking")) + e_has_modalities = any( + m in e_val for m in ("text", "audio", "image", "video", "thinking") + ) + n_has_modalities = any( + m in n_val for m in ("text", "audio", "image", "video", "thinking") + ) if e_has_modalities or n_has_modalities: for modality in ("text", "audio", "image", "video", "thinking"): @@ -388,20 +474,25 @@ def _merge_sts_metadata(existing: dict, new: dict) -> dict: if merged_cat: out[key] = merged_cat - + # retain any other keys, summing up numeric ones to keep metadata consistent for k, v in existing.items(): if k not in ("schemaVersion", "input", "output", "cached"): out[k] = v for k, v in new.items(): if k not in ("schemaVersion", "input", "output", "cached"): - if k in out and isinstance(out[k], (int, float)) and isinstance(v, (int, float)): + if ( + k in out + and isinstance(out[k], (int, float)) + and isinstance(v, (int, float)) + ): out[k] = out[k] + v else: out[k] = v - + return out + class PaygentCollector(BaseObserver): """Pipecat observer that accumulates usage data for a single call. @@ -514,37 +605,65 @@ class PaygentCollector(BaseObserver): if is_sts_frame: # Normalise the raw provider slug so that variants like # "openai_realtime", "azure_realtime", etc. route correctly. - raw_provider = ( - getattr(self, "_sts_provider", "") or getattr(self, "_llm_provider", "") + raw_provider = getattr( + self, "_sts_provider", "" + ) or getattr(self, "_llm_provider", "") + provider = ( + _detect_provider(raw_provider) + if raw_provider + else "unknown" ) - provider = _detect_provider(raw_provider) if raw_provider else "unknown" if provider not in ("grok", "ultravox"): usage = item.value - raw_metadata = getattr(usage, "raw_usage_metadata", None) + raw_metadata = getattr( + usage, "raw_usage_metadata", None + ) if raw_metadata: # OpenAI Realtime and Azure Realtime (azure→openai via _detect_provider) # share the same wire format. if provider in ("openai", "azure"): - new_meta = _openai_realtime_usage_to_sts_metadata(raw_metadata) + new_meta = ( + _openai_realtime_usage_to_sts_metadata( + raw_metadata + ) + ) else: - new_meta = _google_live_usage_to_sts_metadata(raw_metadata) + new_meta = _google_live_usage_to_sts_metadata( + raw_metadata + ) else: - prompt_tokens = getattr(usage, "prompt_tokens", 0) or 0 - completion_tokens = getattr(usage, "completion_tokens", 0) or 0 - cached_tokens = (getattr(usage, "cache_read_input_tokens", 0) or getattr(usage, "cached_tokens", 0) or 0) + prompt_tokens = ( + getattr(usage, "prompt_tokens", 0) or 0 + ) + completion_tokens = ( + getattr(usage, "completion_tokens", 0) or 0 + ) + cached_tokens = ( + getattr(usage, "cache_read_input_tokens", 0) + or getattr(usage, "cached_tokens", 0) + or 0 + ) new_meta = {"schemaVersion": 1} if prompt_tokens > 0: - new_meta.setdefault("input", {})["text"] = {"tokens": prompt_tokens} + new_meta.setdefault("input", {})["text"] = { + "tokens": prompt_tokens + } if completion_tokens > 0: - new_meta.setdefault("output", {})["text"] = {"tokens": completion_tokens} + new_meta.setdefault("output", {})["text"] = { + "tokens": completion_tokens + } if cached_tokens > 0: new_meta["cached"] = {"tokens": cached_tokens} - + if hasattr(usage, "__dict__"): for k, v in vars(usage).items(): - if not k.startswith("_") and v is not None and k not in new_meta: + if ( + not k.startswith("_") + and v is not None + and k not in new_meta + ): new_meta[k] = v - + self._acc.sts_usage_metadata = _merge_sts_metadata( self._acc.sts_usage_metadata or {}, new_meta ) @@ -579,5 +698,7 @@ class PaygentCollector(BaseObserver): except Exception as exc: logger.warning( "[paygent] Unexpected error processing frame {!r} in collector: {}", - type(data.frame).__name__, exc, exc_info=True, + type(data.frame).__name__, + exc, + exc_info=True, ) diff --git a/api/services/integrations/paygent/completion.py b/api/services/integrations/paygent/completion.py index 4893f663..689118ae 100644 --- a/api/services/integrations/paygent/completion.py +++ b/api/services/integrations/paygent/completion.py @@ -12,6 +12,7 @@ Mirrors ``tuner/completion.py`` exactly: - call ``deliver(config, snapshot)`` - collect results keyed by ``paygent_{node_id}`` """ + from __future__ import annotations from datetime import UTC, datetime @@ -38,8 +39,8 @@ def _build_snapshot( # snapshot value is never used to override it, preventing billing drift # if the log is stale or corrupted. session_id=str(workflow_run_id), - agent_id=raw.get("agent_id", ""), # filled from node config below - customer_id=raw.get("customer_id", ""), # filled from node config below + agent_id=raw.get("agent_id", ""), # filled from node config below + customer_id=raw.get("customer_id", ""), # filled from node config below is_realtime=raw.get("is_realtime", False), stt_provider=raw.get("stt_provider", ""), stt_model=raw.get("stt_model", ""), @@ -90,7 +91,9 @@ async def run_completion( continue # ---- Build typed objects ------------------------------------------- - snapshot = _build_snapshot(raw_snapshot, workflow_run_id=context.workflow_run_id) + snapshot = _build_snapshot( + raw_snapshot, workflow_run_id=context.workflow_run_id + ) # Inject node-level credentials into the snapshot snapshot.agent_id = (node_data.paygent_agent_id or "").strip() snapshot.customer_id = (node_data.paygent_customer_id or "").strip() @@ -102,7 +105,11 @@ async def run_completion( # Only fallback to pipeline-level llm usage if this is NOT a realtime pipeline. # In realtime pipelines, the collector properly segregates STS and LLM tokens; # falling back here would duplicate the STS tokens into the LLM bucket. - if snapshot.llm_prompt_tokens == 0 and snapshot.llm_completion_tokens == 0 and not snapshot.is_realtime: + if ( + snapshot.llm_prompt_tokens == 0 + and snapshot.llm_completion_tokens == 0 + and not snapshot.is_realtime + ): llm_providers: list[str] = [] llm_models: list[str] = [] for key, val in usage_info.get("llm", {}).items(): @@ -112,7 +119,9 @@ async def run_completion( continue snapshot.llm_prompt_tokens += val.get("prompt_tokens", 0) snapshot.llm_completion_tokens += val.get("completion_tokens", 0) - snapshot.llm_cached_tokens += val.get("cache_read_input_tokens", 0) + val.get("cache_creation_input_tokens", 0) + snapshot.llm_cached_tokens += val.get( + "cache_read_input_tokens", 0 + ) + val.get("cache_creation_input_tokens", 0) parts = key.split("|||") if len(parts) == 2: llm_providers.append(parts[0]) @@ -153,7 +162,11 @@ async def run_completion( # substitute total_duration_seconds — that would overbill wall-clock time # (silence, hold, agent speech) as STT input. except Exception as exc: - logger.warning("[paygent] Failed to apply usage_info fallback for run {}: {}", context.workflow_run_id, exc) + logger.warning( + "[paygent] Failed to apply usage_info fallback for run {}: {}", + context.workflow_run_id, + exc, + ) try: config = PaygentDeliveryConfig( diff --git a/api/services/integrations/paygent/runtime.py b/api/services/integrations/paygent/runtime.py index 14f0a2f5..6d752b27 100644 --- a/api/services/integrations/paygent/runtime.py +++ b/api/services/integrations/paygent/runtime.py @@ -12,11 +12,11 @@ Lifecycle: integration framework, which persists it in ``workflow_run.logs`` under the key ``"paygent_snapshot"``. """ + from __future__ import annotations from typing import Any -from api.services.configuration.registry import ServiceProviders from api.services.integrations.base import ( IntegrationRuntimeContext, IntegrationRuntimeSession, @@ -48,12 +48,12 @@ def _resolve_model_labels( llm_provider = getattr(user_config.llm, "provider", "") or "" llm_model = getattr(user_config.llm, "model", "") or "" return ( - "", # stt_provider (no separate STT in realtime) - "", # stt_model + "", # stt_provider (no separate STT in realtime) + "", # stt_model llm_provider, llm_model, - "", # tts_provider (no separate TTS in realtime) - "", # tts_model + "", # tts_provider (no separate TTS in realtime) + "", # tts_model realtime_provider, realtime_model, ) @@ -90,9 +90,7 @@ class PaygentRuntimeSession(IntegrationRuntimeSession): gathered_context: dict[str, Any], ) -> dict[str, Any] | None: """Seal the snapshot and hand it to the framework for persistence.""" - self._collector.set_call_disposition( - gathered_context.get("call_disposition") - ) + self._collector.set_call_disposition(gathered_context.get("call_disposition")) snapshot = self._collector.build_snapshot() return {"paygent_snapshot": snapshot} @@ -109,8 +107,7 @@ def create_runtime_sessions( paygent_nodes = [ node for node in context.workflow_graph.nodes.values() - if node.node_type == "paygent" - and getattr(node.data, "paygent_enabled", True) + if node.node_type == "paygent" and getattr(node.data, "paygent_enabled", True) ] if not paygent_nodes: return [] diff --git a/api/services/telephony/providers/cloudonix/provider.py b/api/services/telephony/providers/cloudonix/provider.py index 9051f506..3c1cd0fc 100644 --- a/api/services/telephony/providers/cloudonix/provider.py +++ b/api/services/telephony/providers/cloudonix/provider.py @@ -1096,9 +1096,7 @@ class CloudonixProvider(TelephonyProvider): from_number = random.choice(self.from_numbers) backend_endpoint, _ = await get_backend_endpoints() - callback_url = ( - f"{backend_endpoint}/api/v1/telephony/cloudonix/transfer-result/{transfer_id}" - ) + callback_url = f"{backend_endpoint}/api/v1/telephony/cloudonix/transfer-result/{transfer_id}" endpoint = f"{self.base_url}/calls/{self.domain_id}/application" data: Dict[str, Any] = { @@ -1111,9 +1109,7 @@ class CloudonixProvider(TelephonyProvider): data.update(kwargs) headers = self._get_auth_headers() - masked_destination = ( - f"***{destination[-4:]}" if len(destination) > 4 else "***" - ) + masked_destination = f"***{destination[-4:]}" if len(destination) > 4 else "***" logger.info( f"[Cloudonix Transfer] Dialing {masked_destination} into conference " f"{conference_name} (transfer_id={transfer_id})" diff --git a/api/services/telephony/providers/cloudonix/routes.py b/api/services/telephony/providers/cloudonix/routes.py index 502aa450..2cba23ed 100644 --- a/api/services/telephony/providers/cloudonix/routes.py +++ b/api/services/telephony/providers/cloudonix/routes.py @@ -79,7 +79,7 @@ async def handle_cloudonix_transfer_result(transfer_id: str, request: Request): original_call_sid = transfer_context.original_call_sid conference_name = transfer_context.conference_name - if (conferenceStatus == "participant-join"): + if conferenceStatus == "participant-join": event = TransferEvent( type=TransferEventType.DESTINATION_ANSWERED, transfer_id=transfer_id, @@ -89,7 +89,7 @@ async def handle_cloudonix_transfer_result(transfer_id: str, request: Request): message="Great! The destination answered. Connecting you now.", status="success", action="destination_answered", - ) + ) elif outboundCallStatus in _CLOUDONIX_TRANSFER_FAILURE_STATUSES: event = TransferEvent( type=TransferEventType.TRANSFER_FAILED, diff --git a/api/services/telephony/providers/cloudonix/strategies.py b/api/services/telephony/providers/cloudonix/strategies.py index 32ae389e..79d2178a 100644 --- a/api/services/telephony/providers/cloudonix/strategies.py +++ b/api/services/telephony/providers/cloudonix/strategies.py @@ -120,7 +120,9 @@ class CloudonixConferenceStrategy(TransferStrategy): manager = await get_call_transfer_manager() await manager.remove_transfer_context(transfer_id) except Exception as e: - logger.error(f"[Cloudonix Transfer] Error cleaning up transfer context: {e}") + logger.error( + f"[Cloudonix Transfer] Error cleaning up transfer context: {e}" + ) class CloudonixHangupStrategy(HangupStrategy):