feat(beta): add post-call CRM/GTM telemetry collector by abhishektangudu · Pull Request #6716 · livekit/agents

abhishektangudu · 2026-08-05T19:36:50Z

Description

When building enterprise voice agents, syncing session data to systems of record
(Salesforce, HubSpot, custom webhooks) is a core requirement.

While LiveKit offers OpenTelemetry for APM traces and Server Webhooks for
room lifecycle events, developers often need a single, structured JSON document
containing full transcripts and @function_tool execution records to send to
business webhooks. This PR introduces livekit.agents.beta.gtm_telemetry, an
opt-in collector that bridges this gap.

Related Issue: (#6664)

Example generated payload

{
  "report_id": "rep_12345",
  "room_name": "sales-discovery-abc",
  "room_id": "rm_9876543210",
  "participant_identity": "user_jane",
  "turns": [
    {"speaker": "user", "text": "Do you integrate with Salesforce?", "timestamp": 1721850000.0},
    {"speaker": "agent", "text": "Yes! Let me check your account.", "timestamp": 1721850005.0}
  ],
  "tool_invocations": [
    {
      "call_id": "call_001",
      "tool_name": "lookup_salesforce_contact",
      "arguments": {"email": "test@example.com"},
      "result": "{\"Status\": \"Active\"}",
      "duration_ms": 1050.0,
      "status": "done"
    }
  ],
  "metrics": {
    "total_duration_seconds": 45.0,
    "user_speech_duration_seconds": 15.2,
    "agent_speech_duration_seconds": 20.8,
    "total_tool_calls": 1,
    "failed_tool_calls": 0,
    "avg_llm_ttft_ms": 180.5
  }
}

What's included

Component	Description
`beta/gtm_telemetry/models.py`	Pydantic schemas: `PostCallReport`, `ToolInvocationRecord`, `TranscriptTurn`, `CallMetrics`
`beta/gtm_telemetry/collector.py`	`PostCallTelemetryCollector` — binds to `AgentSession` events, self-times tool latencies, handles deferred tools and rejected calls
`beta/gtm_telemetry/webhook.py`	`WebhookDispatcher` — HMAC-SHA256 signing, exponential backoff retries, redirect handling
`beta/gtm_telemetry/adapters.py`	Pure payload builders for HubSpot Engagements and Salesforce Tasks
`examples/voice_agents/gtm_telemetry_agent.py`	End-to-end example using `on_session_end` hook
`tests/test_gtm_telemetry.py`	17 unit tests

How it complements existing features

Does not replace OpenTelemetry — OTel continues to trace micro-spans for APM.
Does not replace Server Webhooks — they continue handling room lifecycle.
Does not duplicate SessionReport — that path is Cloud-only and lacks tool-duration/CRM fields.
Provides self-hosted and cloud users with an application-level GTM summary payload.

How to test

make check                                        # ruff + mypy strict
uv run pytest tests/test_gtm_telemetry.py --unit  # targeted tests
bash -c "ulimit -n 65536; uv run pytest --unit"   # full suite regression

Checklist

Code follows project style (make fix / ruff)
Type-checked with make check (mypy strict, 0 errors)
Tests added and passing (17 new, 0 regressions)
New classes/methods documented (pdoc3-compatible docstrings)
Placed in beta/ namespace per the repo's opt-in feature convention

Implements PostCallTelemetryCollector, PostCallReport models, WebhookDispatcher with HMAC-SHA256 signing and resilient retry, and CRM adapter payload builders for HubSpot/Salesforce. Placed in livekit/agents/beta/gtm_telemetry/ (the repo's opt-in staging area) with float-epoch timestamps matching repo convention. Wired into beta/__init__.py re-exports.

15 test cases covering: turn/tool accumulation, error-text precedence, batch-only untimed records, deferred-merge rule, metrics aggregation, nullable metrics, HMAC webhook signature, retry/4xx/exhaustion, CRM adapter output shape, lazy http_context resolution, RuntimeError guard, and a fake-session end-to-end integration test.

Demonstrates PostCallTelemetryCollector wiring with AgentServer, @server.rtc_session(on_session_end=...) for flush lifecycle, optional webhook dispatch, and CRM adapter output logging.

…metry feat(beta): add post-call CRM/GTM telemetry collector and webhook dispatcher

CLAassistant · 2026-08-05T19:36:58Z

All committers have signed the CLA.

…telemetry

…g leak

…PII leakage

…guide

…edgecases

…status to the neutral "running" so unknown-outcome calls aren't reported as successful

…and CRM use generate_report() now caches the PostCallReport on first call and returns the same instance on subsequent calls. This ensures the webhook dispatch (triggered by _flush_impl) and the user's CRM adapter call (in on_session_end) receive identical report_id and created_at values, enabling downstream correlation and de-duplication. The cache is cleared in aclose() and attach() for a symmetric lifecycle.

…le snapshots Previously the report was cached unconditionally on first generate_report() call. If called mid-session (before the close event), the cached report froze the transcript, tool records, and duration at that instant, and the auto-flush at session close dispatched stale data. Now the cache is only populated once _close_event is set, so: - Mid-call calls return fresh snapshots (accumulated state is live) - Post-close calls return the same cached instance (webhook and CRM adapter share identical report_id/created_at) Tests updated: test_generate_report_snapshots_mid_call verifies fresh snapshots, test_generate_report_idempotent_after_close verifies the cache kicks in after close.

…repend scheme The class and module docstrings incorrectly claimed the HMAC covered "the exact JSON body bytes". In reality the code prepends f"{timestamp}." before hashing. The verify() snippet was already correct; only the prose was wrong. Receivers following the old docs would compute a different hash and reject all deliveries.

devin-ai-integration

Devin Review found 2 new potential issues.

View 4 additional findings in Devin Review.

devin-ai-integration · 2026-08-06T07:50:10Z

+        except asyncio.CancelledError:
+            if self._flush_task.cancelled():
+                # the flush task itself was cancelled, not this caller
+                logger.warning("post-call report flush was cancelled before completing")
+                return
+            logger.warning("post-call report flush was cancelled before completing")
+            raise


🟡 Shutdown cancellation can be silently ignored while waiting for the report to be sent

A cancellation request aimed at the caller is swallowed and turned into a normal return (return at livekit-agents/livekit/agents/beta/gtm_telemetry/collector.py:240) whenever the sending job also happens to be cancelled, so code that asked to stop waiting keeps running instead of stopping.

Impact: During process shutdown the hook that waits for the post-call report can ignore its own cancellation and continue executing, delaying or confusing shutdown.

Why the cancelled-flag check misattributes the cancellation

In aflush() (livekit-agents/livekit/agents/beta/gtm_telemetry/collector.py:229-242), asyncio.CancelledError can be raised for two independent reasons: (a) the shielded _flush_task itself was cancelled, or (b) the awaiting task (e.g. the on_session_end hook) was cancelled. The code disambiguates using self._flush_task.cancelled(), but during an event-loop/job shutdown both are typically cancelled at the same time. In that case the branch takes the return path and never re-raises, so the caller's cancellation is absorbed. Per asyncio convention, CancelledError must be re-raised unless the coroutine genuinely handled the cancellation of a sub-task. A more reliable disambiguation is to check whether the current task has a pending cancellation (asyncio.current_task().cancelling() / re-raise unless the wait itself completed) rather than relying solely on the flush task's state.

Was this helpful? React with 👍 or 👎 to provide feedback.

devin-ai-integration · 2026-08-06T07:50:12Z

+        if url.lower().startswith("http://"):
+            logger.warning(
+                "Webhook URL uses http:// scheme. Sensitive conversation data "
+                "will be transmitted in cleartext."
+            )
+        self._url = url
+        self._webhook_secret = webhook_secret
+        self._max_retries = max_retries
+        self._base_delay = base_delay
+        self._timeout = timeout
+        self._http_session = http_session


🟨 Webhook URL is not restricted to HTTPS; plaintext transcripts sent over http:// only warn

The dispatcher accepts any user-supplied URL and only logs a warning when the scheme is http:// (livekit-agents/livekit/agents/beta/gtm_telemetry/webhook.py:63-67), then POSTs the full report — transcripts, tool arguments/results, and diagnostic exception text — to it. If an operator misconfigures POST_CALL_WEBHOOK_URL (as in examples/voice_agents/gtm_telemetry_agent.py:89), sensitive conversation data plus the HMAC signature are transmitted in cleartext and can be intercepted.

Was this helpful? React with 👍 or 👎 to provide feedback.

abhishektangudu and others added 8 commits July 24, 2026 21:35

docs(examples): add gtm_telemetry_agent example

e66b12f

Demonstrates PostCallTelemetryCollector wiring with AgentServer, @server.rtc_session(on_session_end=...) for flush lifecycle, optional webhook dispatch, and CRM adapter output logging.

refactor(beta): simplify gtm_telemetry implementation

c67ebe2

fix(beta): address review feedback for gtm_telemetry

84f20f4

Merge pull request #1 from abhishektangudu/vorflux/gtm-post-call-tele…

86d4651

…metry feat(beta): add post-call CRM/GTM telemetry collector and webhook dispatcher

Merge branch 'livekit:main' into main

f331e4f

Merge branch 'livekit:main' into main

9b9bb01

abhishektangudu requested a review from a team as a code owner August 5, 2026 19:36