apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-07-16 11:31:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

Paygent integration new with revert pipecat/realtime changes (#539)

Browse source 
* added paygent integration

* fix(paygent): resolve PR code review issues for cost tracking and billing

* docs(integrations): add Paygent integration guide

* docs(paygent): align step 1 with 3-step agent creation UI screenshots

* removed pipecat/realtime changes and review comments solved
This commit is contained in:
Nihalkumar Dwivedi 2026-07-15 13:09:05 +05:30 • committed by GitHub
parent 076edd1bd0
commit 3739ebaf21
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1503 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

31
api/services/integrations/paygent/__init__.py Normal file
View file

@ -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"]

310
api/services/integrations/paygent/client.py Normal file
View file

@ -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"),
}

583
api/services/integrations/paygent/collector.py Normal file
View file

@ -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,
)

180
api/services/integrations/paygent/completion.py Normal file
View file

@ -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

149
api/services/integrations/paygent/node.py Normal file
View file

@ -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",),
)

142
api/services/integrations/paygent/runtime.py Normal file
View file

@ -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)]

3
docs/docs.json
View file

@ -117,7 +117,8 @@
"tag": "NEW",
"pages": [
"integrations/mcp",
"integrations/tuner"
"integrations/tuner",
"integrations/paygent"
]
}
]

BIN
docs/images/paygent-create-agent-image-1.webp Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 30 KiB

BIN
docs/images/paygent-create-agent-image-2.webp Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 10 KiB

BIN
docs/images/paygent-create-agent-image-3.webp Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 22 KiB

106
docs/integrations/paygent.mdx Normal file
View file

@ -0,0 +1,106 @@
---
title: "Paygent Integration"
description: "Connect Dograh to Paygent — real-time cost tracking, multimodal usage monitoring, and billing for voice agents"
---
<iframe
width="100%"
height="400"
src="https://www.loom.com/embed/09fce8f875564f51821429665778e8e8"
title="Paygent Integration Walkthrough"
frameBorder="0"
allowFullScreen
></iframe>
## 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**:
<img
src="/images/paygent-create-agent-image-1.webp"
alt="Creating a new agent in Paygent - Agent Name and 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:
<img
src="/images/paygent-create-agent-image-2.webp"
alt="Setting your indicator name in Paygent"
/>
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:
<img
src="/images/paygent-create-agent-image-3.webp"
alt="Configuring custom pricing margins and rate cards in Paygent"
/>
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