diff --git a/api/services/integrations/paygent/__init__.py b/api/services/integrations/paygent/__init__.py
new file mode 100644
index 00000000..79394ca1
--- /dev/null
+++ b/api/services/integrations/paygent/__init__.py
@@ -0,0 +1,31 @@
+"""Paygent integration package.
+
+Self-registers on import via ``register_package``. Auto-discovered by
+``api/services/integrations/loader.py`` (scans all submodules of
+``api.services.integrations`` except ``base``, ``loader``, and ``registry``).
+
+Provides:
+- ``PaygentNodeData`` – Pydantic config node shown in the Dograh UI under
+ INTEGRATIONS → "Paygent"
+- ``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
+from api.services.integrations.registry import register_package
+
+from .completion import run_completion
+from .node import NODE
+from .runtime import create_runtime_sessions
+
+PACKAGE = register_package(
+ IntegrationPackageSpec(
+ name="paygent",
+ nodes=(NODE,),
+ create_runtime_sessions=create_runtime_sessions,
+ run_completion=run_completion,
+ )
+)
+
+__all__ = ["PACKAGE"]
diff --git a/api/services/integrations/paygent/client.py b/api/services/integrations/paygent/client.py
new file mode 100644
index 00000000..e39e3263
--- /dev/null
+++ b/api/services/integrations/paygent/client.py
@@ -0,0 +1,310 @@
+"""Paygent REST API client (pure httpx, no SDK).
+
+All network I/O goes through ``post_paygent`` which is the single delivery
+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 typing import Any
+
+import httpx
+from loguru import logger
+from pydantic import BaseModel, field_validator
+
+_DEFAULT_BASE_URL = "https://cp-api.withpaygent.com"
+_REQUEST_TIMEOUT = 15 # seconds – generous for post-call delivery
+
+
+# ---------------------------------------------------------------------------
+# Config model
+# ---------------------------------------------------------------------------
+
+
+class PaygentDeliveryConfig(BaseModel):
+ """Validated delivery configuration, filled from the node data."""
+
+ base_url: str = _DEFAULT_BASE_URL
+ api_key: str
+ agent_id: str
+ customer_id: str
+
+ @field_validator("api_key", "agent_id", "customer_id")
+ @classmethod
+ def _must_not_be_empty(cls, value: str) -> str:
+ if not value or not value.strip():
+ raise ValueError("must not be empty")
+ return value.strip()
+
+ @field_validator("base_url")
+ @classmethod
+ def _normalise_base_url(cls, value: str) -> str:
+ return (value or _DEFAULT_BASE_URL).rstrip("/")
+
+
+# ---------------------------------------------------------------------------
+# Live-call snapshot (collected during the call, delivered after)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PaygentCallSnapshot:
+ """Immutable snapshot produced at call-finish; passed to ``deliver``."""
+
+ session_id: str
+ agent_id: str
+ customer_id: str
+ is_realtime: bool
+
+ # Usage buckets filled from PipelineMetricsAggregator + user_config
+ stt_provider: str = ""
+ stt_model: str = ""
+ stt_audio_seconds: float = 0.0
+
+ llm_provider: str = ""
+ llm_model: str = ""
+ llm_prompt_tokens: int = 0
+ llm_completion_tokens: int = 0
+ llm_cached_tokens: int = 0
+
+ tts_provider: str = ""
+ tts_model: str = ""
+ tts_characters: int = 0
+
+ sts_provider: str = ""
+ sts_model: str = ""
+ sts_usage_metadata: dict[str, Any] | None = None
+
+ # Final call status / total duration seconds
+ call_disposition: str = "completed"
+ total_duration_seconds: int = 0
+ indicator: str = "per-minute-call"
+
+ def to_dict(self) -> dict[str, Any]:
+ return {
+ "session_id": self.session_id,
+ "agent_id": self.agent_id,
+ "customer_id": self.customer_id,
+ "is_realtime": self.is_realtime,
+ "stt": {
+ "provider": self.stt_provider,
+ "model": self.stt_model,
+ "audio_seconds": self.stt_audio_seconds,
+ },
+ "llm": {
+ "provider": self.llm_provider,
+ "model": self.llm_model,
+ "prompt_tokens": self.llm_prompt_tokens,
+ "completion_tokens": self.llm_completion_tokens,
+ "cached_tokens": self.llm_cached_tokens,
+ },
+ "tts": {
+ "provider": self.tts_provider,
+ "model": self.tts_model,
+ "characters": self.tts_characters,
+ },
+ "sts": {
+ "provider": self.sts_provider,
+ "model": self.sts_model,
+ "usage_metadata": self.sts_usage_metadata,
+ },
+ "call_disposition": self.call_disposition,
+ "total_duration_seconds": self.total_duration_seconds,
+ "indicator": self.indicator,
+ }
+
+
+# ---------------------------------------------------------------------------
+# REST delivery helpers
+# ---------------------------------------------------------------------------
+
+
+def _headers(api_key: str) -> dict[str, str]:
+ return {
+ "Content-Type": "application/json",
+ "paygent-api-key": api_key,
+ }
+
+
+async def _post(
+ client: httpx.AsyncClient,
+ url: str,
+ api_key: str,
+ payload: dict[str, Any],
+ *,
+ label: str,
+) -> None:
+ """POST ``payload`` to ``url``; raises on 4xx/5xx or network failure.
+
+ Intentionally non-swallowing: callers in ``deliver()`` each wrap this in
+ their own try/except to build the ``errors`` list and the ``status`` field.
+ """
+ resp = await client.post(url, json=payload, headers=_headers(api_key))
+ resp.raise_for_status()
+
+
+async def deliver(
+ config: PaygentDeliveryConfig,
+ snapshot: PaygentCallSnapshot,
+) -> dict[str, Any]:
+ """
+ Execute the full Paygent REST call sequence for one completed call:
+
+ 1. initialize_voice_session
+ 2. track_stt (if STT is used, i.e. not realtime-only)
+ 3. track_llm
+ 4. track_tts (if TTS is used, i.e. not realtime-only)
+ 5. track_sts (if realtime / STS model used)
+ 6. set_indicator (always; marks end of session)
+
+ Returns a result dict merged into ``workflow_run.annotations``.
+ """
+ base = config.base_url
+ api_key = config.api_key
+ session_id = snapshot.session_id
+
+ delivered_steps: list[str] = []
+ errors: list[str] = []
+
+ async with httpx.AsyncClient(timeout=_REQUEST_TIMEOUT) as client:
+
+ # 1. Initialize voice session ----------------------------------------
+ try:
+ await _post(
+ client,
+ f"{base}/api/v1/voice/session",
+ api_key,
+ {
+ "sessionId": session_id,
+ "agentId": snapshot.agent_id,
+ "customerId": snapshot.customer_id,
+ },
+ label="initialize_voice_session",
+ )
+ delivered_steps.append("session_init")
+ except Exception as exc:
+ errors.append(f"session_init: {exc}")
+
+ # 2. Track STT (only for non-realtime pipelines) ---------------------
+ if not snapshot.is_realtime and snapshot.stt_audio_seconds > 0:
+ try:
+ await _post(
+ client,
+ f"{base}/api/v1/voice/stt",
+ api_key,
+ {
+ "sessionId": session_id,
+ "audioMinutes": snapshot.stt_audio_seconds / 60.0,
+ "provider": snapshot.stt_provider,
+ "model": snapshot.stt_model,
+ "plan": "",
+ },
+ label="track_stt",
+ )
+ delivered_steps.append("track_stt")
+ except Exception as exc:
+ errors.append(f"track_stt: {exc}")
+
+ # 3. Track LLM -------------------------------------------------------
+ if snapshot.llm_prompt_tokens > 0 or snapshot.llm_completion_tokens > 0:
+ llm_payload: dict[str, Any] = {
+ "sessionId": session_id,
+ "provider": snapshot.llm_provider,
+ "model": snapshot.llm_model,
+ "plan": "",
+ "promptTokens": snapshot.llm_prompt_tokens,
+ "completionTokens": snapshot.llm_completion_tokens,
+ }
+ if snapshot.llm_cached_tokens > 0:
+ llm_payload["cachedTokens"] = snapshot.llm_cached_tokens
+ try:
+ await _post(
+ client,
+ f"{base}/api/v1/voice/llm",
+ api_key,
+ llm_payload,
+ label="track_llm",
+ )
+ delivered_steps.append("track_llm")
+ except Exception as exc:
+ errors.append(f"track_llm: {exc}")
+
+ # 4. Track TTS (only for non-realtime pipelines) ---------------------
+ if not snapshot.is_realtime and snapshot.tts_characters > 0:
+ try:
+ await _post(
+ client,
+ f"{base}/api/v1/voice/tts",
+ api_key,
+ {
+ "sessionId": session_id,
+ "provider": snapshot.tts_provider,
+ "model": snapshot.tts_model,
+ "plan": "",
+ "characters": snapshot.tts_characters,
+ },
+ label="track_tts",
+ )
+ delivered_steps.append("track_tts")
+ except Exception as exc:
+ errors.append(f"track_tts: {exc}")
+
+ # 5. Track STS (Speech-to-Speech) for Realtime Models ----------------
+ if snapshot.is_realtime:
+ 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}
+
+ try:
+ await _post(
+ client,
+ f"{base}/api/v1/voice/speech-to-speech",
+ api_key,
+ {
+ "sessionId": session_id,
+ "provider": snapshot.sts_provider,
+ "model": snapshot.sts_model,
+ "plan": "",
+ "usageMetadata": metadata,
+ },
+ label="track_sts",
+ )
+ delivered_steps.append("track_sts")
+ except Exception as exc:
+ errors.append(f"track_sts: {exc}")
+
+ # 6. Set indicator (end-of-session marker) ---------------------------
+ try:
+ await _post(
+ client,
+ f"{base}/api/v1/voice/indicator",
+ api_key,
+ {
+ "sessionId": session_id,
+ "indicator": snapshot.indicator,
+ "totalDuration": snapshot.total_duration_seconds / 60.0,
+ },
+ label="set_indicator",
+ )
+ delivered_steps.append("set_indicator")
+ except Exception as exc:
+ errors.append(f"set_indicator: {exc}")
+
+ return _result(session_id, delivered_steps, errors)
+
+
+def _result(
+ session_id: str,
+ delivered_steps: list[str],
+ errors: list[str],
+) -> dict[str, Any]:
+ return {
+ "session_id": session_id,
+ "delivered_steps": delivered_steps,
+ "errors": errors,
+ "status": "ok" if not errors else ("partial" if delivered_steps else "failed"),
+ }
diff --git a/api/services/integrations/paygent/collector.py b/api/services/integrations/paygent/collector.py
new file mode 100644
index 00000000..bf8473fc
--- /dev/null
+++ b/api/services/integrations/paygent/collector.py
@@ -0,0 +1,583 @@
+"""Paygent live-call collector.
+
+Attaches to the pipecat pipeline as a ``BaseObserver`` to accumulate per-call
+usage metrics (STT audio seconds, LLM tokens, TTS characters, STS metadata)
+in memory during the call. No network I/O happens here; all delivery is
+deferred to the post-call completion handler.
+
+Design mirrors ``api/services/integrations/tuner/collector.py`` exactly:
+- Attach to the task in ``PaygentRuntimeSession.attach``
+- 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
+from dataclasses import dataclass, field
+from typing import Any, Dict
+
+from loguru import logger
+from pipecat.frames.frames import (
+ CancelFrame,
+ EndFrame,
+ MetricsFrame,
+ StartFrame,
+ TTSTextFrame,
+ UserStartedSpeakingFrame,
+ UserStoppedSpeakingFrame,
+)
+from pipecat.metrics.metrics import (
+ LLMTokenUsage,
+ LLMUsageMetricsData,
+ TTSUsageMetricsData,
+)
+from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.processors.frame_processor import FrameDirection
+
+
+def _detect_provider(name: str, fallback: str = "unknown") -> str:
+ """Map a processor/model name to a canonical Paygent provider slug dynamically."""
+ if not name:
+ return fallback
+ clean_name = name.lower().strip()
+ if "gemini" in clean_name:
+ return "google"
+ suffixes = [
+ "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("-")
+ changed = True
+ break
+ return clean_name or fallback
+
+
+@dataclass
+class _UsageAccumulator:
+ """In-memory accumulator for per-call usage data."""
+
+ # STT
+ stt_audio_seconds: float = 0.0
+
+ # LLM (aggregated across all turns)
+ llm_prompt_tokens: int = 0
+ llm_completion_tokens: int = 0
+ llm_cached_tokens: int = 0
+
+ # TTS
+ tts_characters: int = 0
+ _has_tts_metrics: bool = False
+
+ # STS / realtime (last seen usage_metadata dict; callers merge these)
+ sts_usage_metadata: dict[str, Any] | None = None
+
+ # Call timing
+ call_start_abs_ns: int = field(default_factory=time.time_ns)
+ 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.
+
+ NOTE: This is the real measured STT audio duration collected from the
+ pipeline's STT metrics frames, NOT the total call wall-clock duration.
+ The call wall-clock duration is available separately via
+ ``total_duration_seconds``.
+ """
+ return self.stt_audio_seconds
+
+ def add_llm(self, usage: LLMTokenUsage) -> None:
+ self.llm_prompt_tokens += usage.prompt_tokens or 0
+ self.llm_completion_tokens += usage.completion_tokens or 0
+ self.llm_cached_tokens += (usage.cache_read_input_tokens or 0) + (
+ usage.cache_creation_input_tokens or 0
+ )
+
+ def add_tts_metrics(self, data: Any) -> None:
+ 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)):
+ val = data
+ elif hasattr(data, "value"):
+ val = getattr(data, "value", 0) or 0
+ elif hasattr(data, "characters"):
+ 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)
+
+ def add_tts_manual(self, text: str) -> None:
+ if not self._has_tts_metrics:
+ self.tts_characters += len(text)
+
+ def on_user_started_speaking(self) -> None:
+ """Mark the start of a user utterance for STT audio metering."""
+ if self._user_started_speaking_ns is None:
+ self._user_started_speaking_ns = time.time_ns()
+
+ 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
+ self.stt_audio_seconds += elapsed_s
+ self._user_started_speaking_ns = None
+
+ def finalize(self) -> None:
+ if self.call_end_abs_ns is None:
+ self.call_end_abs_ns = time.time_ns()
+ # If user was mid-utterance when the call ended, close the interval.
+ self.on_user_stopped_speaking()
+
+
+def _google_live_usage_to_sts_metadata(usage: Dict[str, Any]) -> Dict[str, Any]:
+ """
+ Pure Python translation of Google GenAI Live usage_metadata to
+ Paygent's canonical speech-to-speech /api/v1/voice/speech-to-speech API schema.
+ """
+ if not usage:
+ return {"schemaVersion": 1}
+
+ def _get_val(obj, *keys):
+ if not obj:
+ return None
+ for k in keys:
+ if isinstance(obj, dict):
+ if k in obj: return obj[k]
+ else:
+ if hasattr(obj, k): return getattr(obj, k)
+ return None
+
+ def _get_list(obj, *keys):
+ val = _get_val(obj, *keys)
+ if val is None:
+ return None
+ return list(val) if not isinstance(val, list) else val
+
+ def _optional_int(obj, *keys):
+ val = _get_val(obj, *keys)
+ if val is not None:
+ try:
+ return int(val)
+ except (TypeError, ValueError):
+ return None
+ return None
+
+ def _modality_token_count(details, modality_name):
+ if not details:
+ return 0
+ want = modality_name.upper()
+ total = 0
+ for d in details:
+ try:
+ mod = _get_val(d, "modality")
+ if mod is None:
+ continue
+ label = _get_val(mod, "name") or _get_val(mod, "value") or mod
+ if str(label).upper() != want:
+ continue
+ tc = _get_val(d, "token_count", "tokenCount")
+ total += int(tc or 0)
+ except Exception:
+ continue
+ 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")
+ 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 += doc_as_text
+
+ # fallback aggregate mapping
+ tutc = _optional_int(usage, "tool_use_prompt_token_count", "toolUsePromptTokenCount")
+ if tutc is not None and not tool_details:
+ text_in += int(tutc)
+
+ ptc = _optional_int(usage, "prompt_token_count", "promptTokenCount")
+ if ptc is not None and not prompt_details and not tool_details:
+ 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")
+
+ rtc = _optional_int(usage, "response_token_count", "responseTokenCount")
+ if text_out == 0 and audio_out == 0 and rtc is not None:
+ # Default fallback to audio output for STS audio connection
+ audio_out = int(rtc)
+
+ # 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
+
+ # 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_image = _modality_token_count(cache_details, "IMAGE")
+ 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
+
+ # 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
+
+ # Cached breakdown
+ has_split = bool(cached_text or cached_audio or cached_image)
+ if cached_legacy is not None and cached_legacy > 0 and not has_split:
+ 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
+
+ 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
+ Paygent's canonical speech-to-speech /api/v1/voice/speech-to-speech API schema.
+ """
+ if not usage:
+ return {"schemaVersion": 1}
+
+ def _get_val(obj, *keys):
+ if not obj:
+ return None
+ for k in keys:
+ if isinstance(obj, dict):
+ if k in obj: return obj[k]
+ else:
+ if hasattr(obj, k): return getattr(obj, k)
+ return None
+
+ total_in = int(_get_val(usage, "input_tokens", "inputTokens") or 0)
+ total_out = int(_get_val(usage, "output_tokens", "outputTokens") or 0)
+
+ in_details = _get_val(usage, "input_token_details", "inputTokenDetails") or {}
+ out_details = _get_val(usage, "output_token_details", "outputTokenDetails") or {}
+
+ audio_in = int(_get_val(in_details, "audio_tokens", "audioTokens") or 0)
+ 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_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)
+
+ audio_out = int(_get_val(out_details, "audio_tokens", "audioTokens") or 0)
+ text_out = int(_get_val(out_details, "text_tokens", "textTokens") or 0)
+
+ if not (text_in or audio_in or image_in) and total_in > 0:
+ text_in = total_in - cached_total
+
+ 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
+
+ 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
+
+ return out
+
+
+def _merge_sts_metadata(existing: dict, new: dict) -> dict:
+ if not existing:
+ return new
+ out = {"schemaVersion": 1}
+ for key in ("input", "output", "cached"):
+ e_val = existing.get(key, {})
+ 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"))
+
+ if e_has_modalities or n_has_modalities:
+ for modality in ("text", "audio", "image", "video", "thinking"):
+ e_mod = e_val.get(modality, {}).get("tokens", 0)
+ n_mod = n_val.get(modality, {}).get("tokens", 0)
+ total = e_mod + n_mod
+ if total > 0:
+ merged_cat[modality] = {"tokens": total}
+ # Also sum any lingering aggregate total so no tokens are lost
+ e_agg = e_val.get("tokens", 0) if not e_has_modalities else 0
+ n_agg = n_val.get("tokens", 0) if not n_has_modalities else 0
+ if e_agg or n_agg:
+ # Incorporate the unbroken-down side into the "text" bucket as
+ # a best-effort attribution rather than silently dropping it.
+ existing_text = merged_cat.get("text", {}).get("tokens", 0)
+ merged_cat["text"] = {"tokens": existing_text + e_agg + n_agg}
+ elif "tokens" in e_val or "tokens" in n_val:
+ merged_cat["tokens"] = e_val.get("tokens", 0) + n_val.get("tokens", 0)
+
+ 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)):
+ out[k] = out[k] + v
+ else:
+ out[k] = v
+
+ return out
+
+class PaygentCollector(BaseObserver):
+ """Pipecat observer that accumulates usage data for a single call.
+
+ Accumulates:
+ - LLM token usage from ``MetricsFrame / LLMUsageMetricsData``
+ - TTS character usage from ``MetricsFrame / TTSUsageMetricsData``
+ - STT audio seconds from ``MetricsFrame`` (when exposed by the pipeline)
+ - Call start / end timestamps for ``total_duration_seconds``
+
+ Does **not** do any network I/O.
+ """
+
+ def __init__(
+ self,
+ *,
+ workflow_run_id: int,
+ is_realtime: bool,
+ stt_provider: str = "",
+ stt_model: str = "",
+ llm_provider: str = "",
+ llm_model: str = "",
+ tts_provider: str = "",
+ tts_model: str = "",
+ sts_provider: str = "",
+ sts_model: str = "",
+ ) -> None:
+ super().__init__()
+ self._workflow_run_id = workflow_run_id
+ self._is_realtime = is_realtime
+ self._stt_provider = stt_provider
+ self._stt_model = stt_model
+ self._llm_provider = llm_provider
+ self._llm_model = llm_model
+ self._tts_provider = tts_provider
+ self._tts_model = tts_model
+ self._sts_provider = sts_provider
+ self._sts_model = sts_model
+ self._acc = _UsageAccumulator()
+ self._call_disposition: str = "completed"
+ # Dedup guard: pipecat can re-deliver frames. This collector is created
+ # fresh per call (see create_runtime_sessions) so the set size is bounded
+ # by call duration. We intentionally do NOT trim the set: trimming would
+ # evict old IDs and allow re-delivered frames to be processed a second time.
+ self._seen_frame_ids: set[int] = set()
+
+ # ------------------------------------------------------------------
+ # Public hooks
+ # ------------------------------------------------------------------
+
+ def set_call_disposition(self, disposition: str | None) -> None:
+ if disposition:
+ self._call_disposition = disposition
+
+ def build_snapshot(self) -> dict[str, Any]:
+ """Return a JSON-serialisable dict stored in ``workflow_run.logs``."""
+ self._acc.finalize()
+ stt_audio_sec = self._acc.get_stt_audio_seconds()
+
+ return {
+ "workflow_run_id": self._workflow_run_id,
+ "is_realtime": self._is_realtime,
+ "stt_provider": self._stt_provider,
+ "stt_model": self._stt_model,
+ "stt_audio_seconds": stt_audio_sec,
+ "llm_provider": self._llm_provider,
+ "llm_model": self._llm_model,
+ "llm_prompt_tokens": self._acc.llm_prompt_tokens,
+ "llm_completion_tokens": self._acc.llm_completion_tokens,
+ "llm_cached_tokens": self._acc.llm_cached_tokens,
+ "tts_provider": self._tts_provider,
+ "tts_model": self._tts_model,
+ "tts_characters": self._acc.tts_characters,
+ "sts_provider": self._sts_provider,
+ "sts_model": self._sts_model,
+ "sts_usage_metadata": self._acc.sts_usage_metadata,
+ "call_disposition": self._call_disposition,
+ "total_duration_seconds": self._acc.total_duration_seconds,
+ }
+
+ # ------------------------------------------------------------------
+ # BaseObserver implementation
+ # ------------------------------------------------------------------
+
+ async def on_push_frame(self, data: FramePushed) -> None: # type: ignore[override]
+ try:
+ # Only process downstream frames; ignore upstream (mic → STT direction)
+ if data.direction != FrameDirection.DOWNSTREAM:
+ return
+
+ frame = data.frame
+
+ # Dedup: per-call set; grows with the call but is GC’d when the
+ # call ends. Never trim — trimming would reopen a re-delivery window.
+ if frame.id in self._seen_frame_ids:
+ return
+ self._seen_frame_ids.add(frame.id)
+
+ if isinstance(frame, StartFrame):
+ self._acc.call_start_abs_ns = time.time_ns()
+
+ elif isinstance(frame, MetricsFrame):
+ for item in frame.data:
+ if isinstance(item, LLMUsageMetricsData):
+ is_sts_frame = False
+ proc_lower = (item.processor or "").lower()
+ if getattr(self, "_is_realtime", False):
+ if "realtime" in proc_lower or "live" in proc_lower:
+ is_sts_frame = True
+
+ 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", "")
+ )
+ 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)
+ 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)
+ else:
+ 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)
+ new_meta = {"schemaVersion": 1}
+ if prompt_tokens > 0:
+ new_meta.setdefault("input", {})["text"] = {"tokens": prompt_tokens}
+ if completion_tokens > 0:
+ 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:
+ new_meta[k] = v
+
+ self._acc.sts_usage_metadata = _merge_sts_metadata(
+ self._acc.sts_usage_metadata or {}, new_meta
+ )
+ else:
+ self._acc.add_llm(item.value)
+ elif isinstance(item, TTSUsageMetricsData):
+ chars_val = getattr(item, "value", 0) or 0
+ self._acc.add_tts_metrics(chars_val)
+ # STT usage is exposed as a float in TTSUsageMetricsData-like
+ # structure by some providers; we also pull from the aggregator
+ # snapshot at call-finish (see runtime.py) for robustness.
+
+ elif isinstance(frame, TTSTextFrame):
+ # Fallback character counting for providers that don't emit native TTS metrics.
+ # TTSTextFrame carries only the text actually sent to the TTS engine;
+ # using base TextFrame would incorrectly include user transcriptions.
+ self._acc.add_tts_manual(frame.text)
+
+ elif isinstance(frame, UserStartedSpeakingFrame):
+ # Measure real STT audio seconds from VAD events rather than
+ # relying on wall-clock time. Skipped for realtime pipelines
+ # which have no separate STT stage.
+ if not self._is_realtime:
+ self._acc.on_user_started_speaking()
+
+ elif isinstance(frame, UserStoppedSpeakingFrame):
+ if not self._is_realtime:
+ self._acc.on_user_stopped_speaking()
+
+ elif isinstance(frame, (EndFrame, CancelFrame)):
+ self._acc.finalize()
+ except Exception as exc:
+ logger.warning(
+ "[paygent] Unexpected error processing frame {!r} in collector: {}",
+ type(data.frame).__name__, exc, exc_info=True,
+ )
diff --git a/api/services/integrations/paygent/completion.py b/api/services/integrations/paygent/completion.py
new file mode 100644
index 00000000..4893f663
--- /dev/null
+++ b/api/services/integrations/paygent/completion.py
@@ -0,0 +1,180 @@
+"""Paygent post-call completion handler.
+
+Reads the ``paygent_snapshot`` that the runtime session stored in
+``workflow_run.logs``, reconstructs the full ``PaygentCallSnapshot``, and
+drives the ordered REST delivery sequence via ``client.deliver()``.
+
+Mirrors ``tuner/completion.py`` exactly:
+- validate each node with Pydantic
+- skip disabled nodes
+- read runtime snapshot from ``context.workflow_run.logs``
+- build a ``PaygentDeliveryConfig`` per node
+- call ``deliver(config, snapshot)``
+- collect results keyed by ``paygent_{node_id}``
+"""
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from loguru import logger
+
+from api.services.integrations.base import IntegrationCompletionContext
+
+from .client import PaygentCallSnapshot, PaygentDeliveryConfig, deliver
+from .node import PaygentNodeData
+
+_DEFAULT_BASE_URL = "https://cp-api.withpaygent.com"
+
+
+def _build_snapshot(
+ raw: dict[str, Any],
+ *,
+ workflow_run_id: int,
+) -> PaygentCallSnapshot:
+ """Reconstruct a ``PaygentCallSnapshot`` from the persisted log dict."""
+ return PaygentCallSnapshot(
+ # session_id is always the authoritative workflow_run_id; the persisted
+ # 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
+ is_realtime=raw.get("is_realtime", False),
+ stt_provider=raw.get("stt_provider", ""),
+ stt_model=raw.get("stt_model", ""),
+ stt_audio_seconds=float(raw.get("stt_audio_seconds", 0.0)),
+ llm_provider=raw.get("llm_provider", ""),
+ llm_model=raw.get("llm_model", ""),
+ llm_prompt_tokens=int(raw.get("llm_prompt_tokens", 0)),
+ llm_completion_tokens=int(raw.get("llm_completion_tokens", 0)),
+ llm_cached_tokens=int(raw.get("llm_cached_tokens", 0)),
+ tts_provider=raw.get("tts_provider", ""),
+ tts_model=raw.get("tts_model", ""),
+ tts_characters=int(raw.get("tts_characters", 0)),
+ sts_provider=raw.get("sts_provider", ""),
+ sts_model=raw.get("sts_model", ""),
+ sts_usage_metadata=raw.get("sts_usage_metadata"),
+ call_disposition=raw.get("call_disposition", "completed"),
+ total_duration_seconds=int(raw.get("total_duration_seconds", 0)),
+ )
+
+
+async def run_completion(
+ nodes: list[dict[str, Any]],
+ context: IntegrationCompletionContext,
+) -> dict[str, Any]:
+ """Post-call completion handler: deliver usage data to Paygent REST API."""
+ results: dict[str, Any] = {}
+
+ raw_snapshot: dict[str, Any] | None = (context.workflow_run.logs or {}).get(
+ "paygent_snapshot"
+ )
+
+ for node in nodes:
+ node_id = node.get("id", "unknown")
+
+ # ---- Validate the node config via Pydantic -------------------------
+ try:
+ node_data = PaygentNodeData.model_validate(node.get("data", {}))
+ except Exception:
+ results[f"paygent_{node_id}"] = {"error": "validation_failed"}
+ continue
+
+ if not node_data.paygent_enabled:
+ continue
+
+ # ---- Guard: runtime snapshot must exist ----------------------------
+ if not raw_snapshot:
+ results[f"paygent_{node_id}"] = {"error": "missing_runtime_snapshot"}
+ continue
+
+ # ---- Build typed objects -------------------------------------------
+ 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()
+ snapshot.indicator = (node_data.paygent_indicator or "per-minute-call").strip()
+
+ # Fallback to usage_info if snapshot has 0s (Pipecat metrics might be missing)
+ usage_info = context.workflow_run.usage_info or {}
+ try:
+ # 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:
+ llm_providers: list[str] = []
+ llm_models: list[str] = []
+ for key, val in usage_info.get("llm", {}).items():
+ # Skip post-call QA analysis entries — they must not be billed
+ # as in-conversation LLM usage.
+ if key.startswith("QAAnalysis|||"):
+ 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)
+ parts = key.split("|||")
+ if len(parts) == 2:
+ llm_providers.append(parts[0])
+ llm_models.append(parts[1])
+ if not snapshot.llm_provider and llm_providers:
+ snapshot.llm_provider = ",".join(dict.fromkeys(llm_providers))
+ if not snapshot.llm_model and llm_models:
+ snapshot.llm_model = ",".join(dict.fromkeys(llm_models))
+
+ if snapshot.tts_characters == 0:
+ tts_providers: list[str] = []
+ tts_models: list[str] = []
+ for key, val in usage_info.get("tts", {}).items():
+ snapshot.tts_characters += val
+ parts = key.split("|||")
+ if len(parts) == 2:
+ tts_providers.append(parts[0])
+ tts_models.append(parts[1])
+ if not snapshot.tts_provider and tts_providers:
+ snapshot.tts_provider = ",".join(dict.fromkeys(tts_providers))
+ if not snapshot.tts_model and tts_models:
+ snapshot.tts_model = ",".join(dict.fromkeys(tts_models))
+
+ if snapshot.stt_audio_seconds == 0:
+ stt_providers: list[str] = []
+ stt_models: list[str] = []
+ for key, val in usage_info.get("stt", {}).items():
+ snapshot.stt_audio_seconds += val
+ parts = key.split("|||")
+ if len(parts) == 2:
+ stt_providers.append(parts[0])
+ stt_models.append(parts[1])
+ if not snapshot.stt_provider and stt_providers:
+ snapshot.stt_provider = ",".join(dict.fromkeys(stt_providers))
+ if not snapshot.stt_model and stt_models:
+ snapshot.stt_model = ",".join(dict.fromkeys(stt_models))
+ # Note: if STT audio seconds remain 0 after all fallbacks, we do NOT
+ # 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)
+
+ try:
+ config = PaygentDeliveryConfig(
+ api_key=(node_data.paygent_api_key or "").strip(),
+ agent_id=snapshot.agent_id,
+ customer_id=snapshot.customer_id,
+ )
+ except Exception as exc:
+ results[f"paygent_{node_id}"] = {"error": f"invalid_config: {exc}"}
+ continue
+
+ # ---- REST delivery -------------------------------------------------
+ try:
+ delivery_result = await deliver(config, snapshot)
+ results[f"paygent_{node_id}"] = {
+ **delivery_result,
+ "agent_id": snapshot.agent_id,
+ "customer_id": snapshot.customer_id,
+ "exported_at": datetime.now(UTC).isoformat(),
+ }
+ except Exception as exc:
+ results[f"paygent_{node_id}"] = {"error": str(exc)}
+
+ return results
diff --git a/api/services/integrations/paygent/node.py b/api/services/integrations/paygent/node.py
new file mode 100644
index 00000000..8ecd2c8d
--- /dev/null
+++ b/api/services/integrations/paygent/node.py
@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+from pydantic import model_validator
+
+from api.services.integrations.base import IntegrationNodeRegistration
+from api.services.workflow.node_data import BaseNodeData
+from api.services.workflow.node_specs._base import (
+ GraphConstraints,
+ NodeCategory,
+ NodeExample,
+ PropertyType,
+)
+from api.services.workflow.node_specs.model_spec import (
+ build_spec,
+ node_spec,
+ spec_field,
+)
+
+
+@node_spec(
+ name="paygent",
+ display_name="Paygent",
+ description="Cost Tracking and Billing",
+ llm_hint=(
+ "Paygent is a post-call usage-tracking and billing integration. "
+ "It does not participate in the conversation graph and should not be connected to other nodes."
+ ),
+ category=NodeCategory.integration,
+ icon="CreditCard",
+ examples=[
+ NodeExample(
+ name="paygent_tracking",
+ data={
+ "name": "Paygent Tracking",
+ "paygent_enabled": True,
+ "paygent_api_key": "pg_live_xxxxxxxxxxxxxxxx",
+ "paygent_agent_id": "my-voice-agent-prod",
+ "paygent_customer_id": "org-123",
+ "paygent_indicator": "per-minute-call",
+ },
+ )
+ ],
+ graph_constraints=GraphConstraints(
+ min_incoming=0, max_incoming=0, min_outgoing=0, max_outgoing=0, max_instances=1
+ ),
+ property_order=(
+ "name",
+ "paygent_enabled",
+ "paygent_api_key",
+ "paygent_agent_id",
+ "paygent_customer_id",
+ "paygent_indicator",
+ ),
+ field_overrides={
+ "name": {
+ "spec_default": "Paygent",
+ "description": "Short identifier for this Paygent configuration.",
+ },
+ "paygent_enabled": {
+ "display_name": "Enabled",
+ "description": "When false, Dograh skips all Paygent tracking for this call.",
+ },
+ "paygent_api_key": {
+ "display_name": "Paygent API Key",
+ "description": "API key used to authenticate requests to the Paygent REST API.",
+ "required": True,
+ },
+ "paygent_agent_id": {
+ "display_name": "Agent ID",
+ "description": "The agent identifier registered in your Paygent account.",
+ "required": True,
+ },
+ "paygent_customer_id": {
+ "display_name": "Customer ID",
+ "description": "Your Paygent customer / organisation ID.",
+ "required": True,
+ },
+ "paygent_indicator": {
+ "display_name": "Indicator",
+ "description": "The indicator event name sent at the end of the call (e.g. per-minute-call).",
+ "required": True,
+ "spec_default": "per-minute-call",
+ },
+ },
+)
+class PaygentNodeData(BaseNodeData):
+ paygent_enabled: bool = spec_field(
+ default=True,
+ ui_type=PropertyType.boolean,
+ display_name="Enabled",
+ description="When false, Dograh skips all Paygent tracking for this call.",
+ )
+ paygent_api_key: str | None = spec_field(
+ default=None,
+ ui_type=PropertyType.string,
+ display_name="Paygent API Key",
+ description="API key used to authenticate requests to the Paygent REST API.",
+ )
+ paygent_agent_id: str | None = spec_field(
+ default=None,
+ ui_type=PropertyType.string,
+ display_name="Agent ID",
+ description="The agent identifier registered in your Paygent account.",
+ )
+ paygent_customer_id: str | None = spec_field(
+ default=None,
+ ui_type=PropertyType.string,
+ display_name="Customer ID",
+ description="Your Paygent customer / organisation ID.",
+ )
+ paygent_indicator: str = spec_field(
+ default="per-minute-call",
+ ui_type=PropertyType.string,
+ display_name="Indicator",
+ description="The indicator event name sent at the end of the call (e.g. per-minute-call).",
+ )
+
+ @model_validator(mode="after")
+ def _validate_enabled_config(self) -> "PaygentNodeData":
+ if not self.paygent_enabled:
+ return self
+
+ missing: list[str] = []
+ if not self.paygent_api_key or not self.paygent_api_key.strip():
+ missing.append("paygent_api_key")
+ if not self.paygent_agent_id or not self.paygent_agent_id.strip():
+ missing.append("paygent_agent_id")
+ if not self.paygent_customer_id or not self.paygent_customer_id.strip():
+ missing.append("paygent_customer_id")
+ if not self.paygent_indicator or not self.paygent_indicator.strip():
+ missing.append("paygent_indicator")
+
+ if missing:
+ fields = ", ".join(missing)
+ raise ValueError(
+ f"Paygent node is enabled but missing required fields: {fields}"
+ )
+
+ return self
+
+
+SPEC = build_spec(PaygentNodeData)
+
+NODE = IntegrationNodeRegistration(
+ type_name="paygent",
+ data_model=PaygentNodeData,
+ node_spec=SPEC,
+ sensitive_fields=("paygent_api_key",),
+)
diff --git a/api/services/integrations/paygent/runtime.py b/api/services/integrations/paygent/runtime.py
new file mode 100644
index 00000000..14f0a2f5
--- /dev/null
+++ b/api/services/integrations/paygent/runtime.py
@@ -0,0 +1,142 @@
+"""Paygent runtime session.
+
+Wires the ``PaygentCollector`` into the live pipecat pipeline exactly the way
+``TunerRuntimeSession`` wires ``TunerCollector``.
+
+Lifecycle:
+ 1. ``create_runtime_sessions`` scans the workflow graph for an enabled
+ ``paygent`` node and, if found, builds a collector from context metadata.
+ 2. ``attach`` hooks the collector into the task as a pipeline observer so it
+ receives all ``MetricsFrame`` events during the call.
+ 3. ``on_call_finished`` seals the snapshot and returns it to the generic
+ 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,
+)
+
+from .collector import PaygentCollector
+
+
+def _label(provider: str | None, model: str | None) -> str:
+ """Compose a human-readable ``provider/model`` label."""
+ if provider and model:
+ return f"{provider}/{model}"
+ return model or provider or ""
+
+
+def _resolve_model_labels(
+ context: IntegrationRuntimeContext,
+) -> tuple[str, str, str, str, str, str, str, str]:
+ """Return (stt_provider, stt_model, llm_provider, llm_model,
+ tts_provider, tts_model, sts_provider, sts_model).
+
+ Mirrors the logic in ``tuner/runtime.py:_resolve_model_labels``.
+ """
+ user_config = context.user_config
+
+ if context.is_realtime and user_config.realtime:
+ realtime_provider = getattr(user_config.realtime, "provider", "") or ""
+ realtime_model = getattr(user_config.realtime, "model", "") or ""
+ 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
+ llm_provider,
+ llm_model,
+ "", # tts_provider (no separate TTS in realtime)
+ "", # tts_model
+ realtime_provider,
+ realtime_model,
+ )
+
+ return (
+ getattr(user_config.stt, "provider", "") or "",
+ getattr(user_config.stt, "model", "") or "",
+ getattr(user_config.llm, "provider", "") or "",
+ getattr(user_config.llm, "model", "") or "",
+ getattr(user_config.tts, "provider", "") or "",
+ getattr(user_config.tts, "model", "") or "",
+ "", # sts_provider
+ "", # sts_model
+ )
+
+
+class PaygentRuntimeSession(IntegrationRuntimeSession):
+ """Thin wrapper that connects the collector to the pipeline task."""
+
+ name = "paygent"
+
+ def __init__(self, collector: PaygentCollector) -> None:
+ self._collector = collector
+
+ # --- IntegrationRuntimeSession protocol --------------------------------
+
+ def attach(self, task: Any) -> None:
+ """Register the collector as a pipeline observer."""
+ task.add_observer(self._collector)
+
+ async def on_call_finished(
+ self,
+ *,
+ 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")
+ )
+ snapshot = self._collector.build_snapshot()
+ return {"paygent_snapshot": snapshot}
+
+
+# ---------------------------------------------------------------------------
+# Runtime session factory (called by the generic integration framework)
+# ---------------------------------------------------------------------------
+
+
+def create_runtime_sessions(
+ context: IntegrationRuntimeContext,
+) -> list[IntegrationRuntimeSession]:
+ """Return a ``PaygentRuntimeSession`` if a live, enabled paygent node exists."""
+ paygent_nodes = [
+ node
+ for node in context.workflow_graph.nodes.values()
+ if node.node_type == "paygent"
+ and getattr(node.data, "paygent_enabled", True)
+ ]
+ if not paygent_nodes:
+ return []
+
+ (
+ stt_provider,
+ stt_model,
+ llm_provider,
+ llm_model,
+ tts_provider,
+ tts_model,
+ sts_provider,
+ sts_model,
+ ) = _resolve_model_labels(context)
+
+ collector = PaygentCollector(
+ workflow_run_id=context.workflow_run_id,
+ is_realtime=context.is_realtime,
+ stt_provider=stt_provider,
+ stt_model=stt_model,
+ llm_provider=llm_provider,
+ llm_model=llm_model,
+ tts_provider=tts_provider,
+ tts_model=tts_model,
+ sts_provider=sts_provider,
+ sts_model=sts_model,
+ )
+
+ return [PaygentRuntimeSession(collector)]
diff --git a/docs/docs.json b/docs/docs.json
index 4dd3a2e6..61cac8da 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -117,7 +117,8 @@
"tag": "NEW",
"pages": [
"integrations/mcp",
- "integrations/tuner"
+ "integrations/tuner",
+ "integrations/paygent"
]
}
]
diff --git a/docs/images/paygent-create-agent-image-1.webp b/docs/images/paygent-create-agent-image-1.webp
new file mode 100644
index 00000000..15a944ac
Binary files /dev/null and b/docs/images/paygent-create-agent-image-1.webp differ
diff --git a/docs/images/paygent-create-agent-image-2.webp b/docs/images/paygent-create-agent-image-2.webp
new file mode 100644
index 00000000..2dbe06cc
Binary files /dev/null and b/docs/images/paygent-create-agent-image-2.webp differ
diff --git a/docs/images/paygent-create-agent-image-3.webp b/docs/images/paygent-create-agent-image-3.webp
new file mode 100644
index 00000000..b51d3356
Binary files /dev/null and b/docs/images/paygent-create-agent-image-3.webp differ
diff --git a/docs/integrations/paygent.mdx b/docs/integrations/paygent.mdx
new file mode 100644
index 00000000..401af5cd
--- /dev/null
+++ b/docs/integrations/paygent.mdx
@@ -0,0 +1,106 @@
+---
+title: "Paygent Integration"
+description: "Connect Dograh to Paygent — real-time cost tracking, multimodal usage monitoring, and billing for voice agents"
+---
+
+
+
+## Overview
+
+**Paygent** is a specialized, usage-based billing platform designed exclusively for AI voice agents.
+
+If you are building voice agents for clients or offering them as a SaaS product, calculating margins across different AI providers (LLMs, TTS, STT, and Speech-to-Speech models) can be incredibly complex. Paygent solves this by serving as your centralized billing engine.
+
+### How you charge your customers
+
+Instead of building custom tracking infrastructure, you simply connect your Dograh workflow to Paygent. As your agents handle calls, Dograh passively calculates the exact multimodal token usage and audio duration.
+
+This data is securely exported to Paygent after every call, where your custom pricing margins (rate cards) are applied. This seamless flow allows you to automatically invoice your end-users for the exact infrastructure they consume — turning your AI agents into a scalable, profitable business with zero engineering overhead.
+
+## Prerequisites
+
+- A [Paygent account](https://withpaygent.com)
+- A Dograh voice agent workflow
+
+## Setup
+
+### 1. Create an agent and configure pricing in Paygent
+
+Before connecting Dograh, you must register your agent in Paygent and define how you want to charge your customers.
+
+Log in to [Paygent](https://withpaygent.com) and click **Create Agent**. You will be prompted to define your agent's core details, including the **Agent Name** and **Agent ID**:
+
+
+
+Next, set the **Indicator Name**. This is the billing event identifier you will pass from Dograh (e.g. `per-minute-call`) to tell Paygent which rate to apply for this agent's calls:
+
+
+
+Finally, you will configure your **Pricing Strategy**. Paygent allows you to set custom markup rates (rate cards) for every modality. You can define exact margins for Speech-to-Text (STT) seconds, LLM tokens, Text-to-Speech (TTS) characters, and total call minutes:
+
+
+
+Once your pricing is configured, confirm your setup. You will use the **Agent ID** and **Indicator** from this process to connect your Dograh workflow.
+
+### 2. Gather your Paygent credentials
+
+You'll need three core values from your Paygent dashboard to link your Dograh agent:
+
+| Credential | Where to find it / Description |
+|---|---|
+| **Paygent API Key** | Your workspace API key used to authenticate requests (`pg_live_...`) |
+| **Agent ID** | The unique identifier configured for your agent in Step 1 |
+| **Customer ID** | Your Paygent organisation or customer ID |
+
+### 3. Add the Paygent node to your workflow
+
+In your Dograh workflow editor, click **Add node** and scroll to the **Integrations** section. Select **Paygent**. The node will appear on your canvas with a **Not configured** badge.
+
+### 4. Configure the node
+
+Click on the Paygent node and fill in the following fields:
+
+- **Paygent API Key** — Your `pg_live_...` secret key
+- **Agent ID** — The unique agent identifier from Paygent
+- **Customer ID** — Your Paygent organisation ID
+- **Indicator** — The billing event name (defaults to `per-minute-call`)
+- **Enabled** — Toggle on to activate the export
+
+Click **Save**, then **Publish** your workflow.
+
+### 5. Verify the connection
+
+Make a test call through your agent. Once the call completes, check your Paygent dashboard. The billing event and detailed multimodal usage breakdown should appear under your configured Agent ID within a few moments, with your configured pricing margins automatically applied.
+
+## Disabling the integration
+
+To temporarily stop exporting usage data to Paygent, open the Paygent node configuration and toggle **Enabled** off. Your credentials are preserved — toggle it back on anytime to resume tracking.
+
+## Troubleshooting
+
+| Issue | Solution |
+|---|---|
+| Usage data not appearing in Paygent | Verify all credentials are correct with no extra whitespace |
+| Node shows "Not configured" | Open the node and fill in API Key, Agent ID, Customer ID, and Indicator |
+| Workflow not sending data | Make sure the workflow is published, not just saved as a draft |
+| Missing realtime STS tokens | For OpenAI Realtime or Google Live models, verify the pipeline is running in realtime mode (`is_realtime=True`) |
+
+## Learn more
+
+- [Paygent](https://withpaygent.com) — Usage-based billing platform for AI agents