LiveKit

import os, aiohttp
from livekit.agents import AgentSession

async def forward_to_zelto(
    session: AgentSession, room_name: str, agent_id: str,
    agent_version: str, system_prompt: str,
    version_config: dict | None = None,
):
    turns = [
        {"role": item.role, "content": item.text_content}
        for item in session.history.items
        if item.text_content
    ]
    async with aiohttp.ClientSession() as http:
        await http.post(
            "https://ingest.zelto.ai/webhooks/calls",
            headers={
                "Authorization": f"Bearer {os.environ['ZELTO_API_KEY']}",
                "X-Zelto-Provider": "livekit",
            },
            json={
                "agent": {"externalId": agent_id},
                "version": agent_version,
                **({"versionConfig": version_config} if version_config is not None else {}),
                "systemPrompt": system_prompt,
                "call": {"externalId": room_name},
                "transcript": {"turns": turns},
            },
        )

def build_turns(session: AgentSession):
    # Index tool outputs by the call id they answer.
    outputs = {
        item.call_id: item.output
        for item in session.history.items
        if item.type == "function_call_output"
    }
    turns = []
    for item in session.history.items:
        if item.type == "message" and item.text_content:
            turns.append({"role": item.role, "content": item.text_content})
        elif item.type == "function_call":
            turns.append({
                "role": "tool",
                "toolCall": {
                    "name": item.name,
                    "toolCallId": item.call_id,
                    "arguments": item.arguments,       # JSON or a JSON string
                    "result": outputs.get(item.call_id),
                    "status": "success" if item.call_id in outputs else "pending",
                },
            })
    return turns

{
  "externalId": "room-example-001",
  "startedAt": "2026-09-14T16:00:00Z",
  "durationSeconds": 20,
  "endedReason": "customer_hangup",
  "recordingUrl": "https://recordings.example.com/room-example-001.wav"
}

from copy import deepcopy

# original_payload es el cuerpo completo enviado previamente para esta sesión.
recorded_payload = deepcopy(original_payload)
recorded_payload["call"]["recordingUrl"] = recording_url
# Envía recorded_payload con la misma clave de API y el encabezado livekit.

import asyncio
import os

from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from livekit.agents import JobContext
from livekit.agents.telemetry import set_tracer_provider

async def entrypoint(ctx: JobContext):
    provider = TracerProvider(resource=Resource.create({
        "service.name": "livekit-agent",
        "zelto.agent_external_id": os.environ["ZELTO_AGENT_EXTERNAL_ID"],
        "zelto.agent_provider": "livekit",
    }))
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    set_tracer_provider(provider, metadata={"session.id": ctx.room.name})

    async def flush_traces():
        await asyncio.to_thread(provider.force_flush, timeout_millis=10_000)

    ctx.add_shutdown_callback(flush_traces)

    # ... build your AgentSession and await session.start(...) here ...

ZELTO_AGENT_EXTERNAL_ID="support-agent"
OTEL_EXPORTER_OTLP_ENDPOINT="https://ingest.zelto.ai"
OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer%20<your-ingest-key>"
OTEL_EXPORTER_OTLP_PROTOCOL="http/protobuf"