feat(beta): add post-call CRM/GTM telemetry collector

examples/voice_agents/gtm_telemetry_agent.py

@@ -0,0 +1,112 @@
+"""Post-call GTM/CRM telemetry example.
+
+Demonstrates how to wire a PostCallTelemetryCollector into an AgentSession so that
+a structured PostCallReport is sent to a webhook (and printed as Salesforce/HubSpot
+payloads) at the end of every call.
+
+The flush hook uses @server.rtc_session(on_session_end=...) which runs after
+session.aclose() with a 300s default budget (WorkerOptions.session_end_timeout),
+comfortably covering the dispatcher's ~31.5s retry worst case. JobContext.add_shutdown_callback
+is NOT used: those callbacks run under the ~10s shutdown_process_timeout, which cannot cover
+the full retry budget.
+"""
+
+import logging
+import os
+
+from dotenv import load_dotenv
+
+from livekit.agents import Agent, AgentServer, AgentSession, JobContext, RunContext, cli, inference
+from livekit.agents.beta.gtm_telemetry import (
+    PostCallTelemetryCollector,
+    WebhookDispatcher,
+)
+from livekit.agents.llm import function_tool
+
+logger = logging.getLogger("gtm-telemetry-agent")
+
+load_dotenv()
+
+# Module-level registry keyed by job ID so the on_session_end hook can find
+# the collector for the closing session. Using job ID (not room name) avoids
+# collisions when two overlapping jobs share the same room under threaded execution.
+_collectors: dict[str, PostCallTelemetryCollector] = {}
+
+
+class SalesAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(
+            instructions=(
+                "You are a sales assistant for Acme Corp. "
+                "Help the caller with product information and qualification."
+            ),
+        )
+
+    @function_tool
+    async def lookup_salesforce_contact(self, context: RunContext, email: str) -> str:
+        """Look up a Salesforce contact record by email address."""
+        # Mock implementation — in production, call the Salesforce REST API
+        logger.info("Looking up contact: %s", email)
+        return (
+            f'{{"Name": "Jane Smith", "Email": "{email}", '
+            f'"Account": "Acme Corp", "Status": "Active"}}'
+        )
+
+
+async def on_session_end(ctx: JobContext) -> None:
+    """Flush the collector and print CRM adapter outputs."""
+    collector = _collectors.pop(ctx.job.id, None)
+    if collector is None:
+        return
+
+    try:
+        await collector.aflush()
+        report = collector.generate_report()
+        logger.info("Generated PostCallReport with %d turns", len(report.turns))
+
+        # NOTE: To prevent PII/transcript leakage into application log aggregation
+        # systems, avoid logging the full report or adapter payloads in production.
+        # If you need to inspect payloads during local development, uncomment below:
+        # print(report.model_dump_json(indent=2))
+    finally:
+        await collector.aclose()
+
+
+server = AgentServer()
+
+
+@server.rtc_session(on_session_end=on_session_end)
+async def gtm_agent(ctx: JobContext) -> None:
+    await ctx.connect()
+
+    session = AgentSession(
+        stt=inference.STT("deepgram/nova-3"),
+        llm=inference.LLM("openai/gpt-4.1-mini"),
+        tts=inference.TTS("cartesia/sonic-3"),
+    )
+
+    # Build the optional webhook dispatcher from environment
+    webhook_url = os.environ.get("POST_CALL_WEBHOOK_URL")
+    dispatcher: WebhookDispatcher | None = None
+    if webhook_url:
+        dispatcher = WebhookDispatcher(
+            webhook_url,
+            webhook_secret=os.environ.get("POST_CALL_WEBHOOK_SECRET"),
+        )
+
+    collector = PostCallTelemetryCollector(
+        session,
+        room_name=ctx.room.name,
+        metadata={"campaign": "demo"},
+        dispatcher=dispatcher,
+    )
+    collector.attach()
+
+    # Store so on_session_end can find it (keyed by job ID for thread safety)
+    _collectors[ctx.job.id] = collector
+
+    await session.start(agent=SalesAgent(), room=ctx.room)
+
+
+if __name__ == "__main__":
+    cli.run_app(server)

livekit-agents/livekit/agents/beta/__init__.py

@@ -1,5 +1,5 @@
 from ..llm.chat_context import Instructions
-from . import workflows
+from . import gtm_telemetry, workflows
 from .tools.end_call import EndCallTool
 
-__all__ = ["Instructions", "workflows", "EndCallTool"]
+__all__ = ["Instructions", "workflows", "EndCallTool", "gtm_telemetry"]

livekit-agents/livekit/agents/beta/gtm_telemetry/__init__.py

@@ -0,0 +1,32 @@
+"""Post-call GTM/CRM telemetry — models, collector, webhook dispatcher, and CRM adapters.
+
+Usage::
+
+    from livekit.agents.beta.gtm_telemetry import (
+        PostCallTelemetryCollector,
+        PostCallReport,
+        WebhookDispatcher,
+        to_salesforce_task,
+        to_hubspot_engagement,
+    )
+"""
+
+from .adapters import to_hubspot_engagement, to_salesforce_task
+from .collector import PostCallTelemetryCollector
+from .models import CallMetrics, PostCallReport, ToolInvocationRecord, TranscriptTurn
+from .webhook import WebhookDispatcher
+
+__all__ = [
+    # collector
+    "PostCallTelemetryCollector",
+    # webhook
+    "WebhookDispatcher",
+    # models
+    "PostCallReport",
+    "ToolInvocationRecord",
+    "TranscriptTurn",
+    "CallMetrics",
+    # adapters
+    "to_hubspot_engagement",
+    "to_salesforce_task",
+]

livekit-agents/livekit/agents/beta/gtm_telemetry/adapters.py

@@ -0,0 +1,70 @@
+"""CRM payload adapters for :class:`~.models.PostCallReport`.
+
+These are pure payload builders: no network calls, no CRM SDK dependencies.
+Users POST the returned dicts to the CRM APIs with their own authenticated
+client (e.g. a HubSpot private-app token or a Salesforce connected app).
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from .models import PostCallReport
+
+
+def _format_transcript(report: PostCallReport) -> str:
+    """Render the transcript turns plus a tool-invocation summary as plain text."""
+    lines: list[str] = []
+    for turn in report.turns:
+        marker = " [interrupted]" if turn.interrupted else ""
+        lines.append(f"{turn.speaker.capitalize()}: {turn.text}{marker}")
+
+    if report.tool_invocations:
+        lines.append("")
+        lines.append("Tool invocations:")
+        for rec in report.tool_invocations:
+            duration = f"{rec.duration_ms:.0f}ms" if rec.duration_ms is not None else "untimed"
+            detail = rec.error if rec.error is not None else rec.result
+            suffix = f" — {detail}" if detail else ""
+            lines.append(f"- {rec.tool_name} ({rec.status}, {duration}){suffix}")
+
+    return "\n".join(lines)
+
+
+def to_hubspot_engagement(report: PostCallReport) -> dict[str, Any]:
+    """Build a HubSpot v3 calls-engagement payload from a post-call report.
+
+    Payload builder only — POST it to ``/crm/v3/objects/calls`` with your own
+    authenticated HubSpot client. ``hs_call_duration`` is expressed in
+    milliseconds as a string and ``hs_timestamp`` in epoch milliseconds, per
+    HubSpot's engagement conventions.
+    """
+    title = f"Call: {report.room_name or report.report_id}"
+    return {
+        "properties": {
+            "hs_call_title": title,
+            "hs_call_body": _format_transcript(report),
+            "hs_call_duration": str(int(report.metrics.total_duration_seconds * 1000)),
+            "hs_call_direction": "INBOUND",
+            "hs_call_status": "COMPLETED",
+            "hs_timestamp": int(report.created_at * 1000),
+        }
+    }
+
+
+def to_salesforce_task(report: PostCallReport) -> dict[str, Any]:
+    """Build a Salesforce Task (ActivityHistory) payload from a post-call report.
+
+    Payload builder only — POST it to ``/services/data/vXX.X/sobjects/Task``
+    with your own authenticated Salesforce client.
+    """
+    return {
+        "Subject": f"Call: {report.room_name or report.report_id}",
+        "Description": _format_transcript(report),
+        "Status": "Completed",
+        "TaskSubtype": "Call",
+        "CallType": "Inbound",
+        "CallDurationInSeconds": int(report.metrics.total_duration_seconds),
+        "ActivityDate": time.strftime("%Y-%m-%d", time.gmtime(report.created_at)),
+    }