How can i log the usage for the stt-llm-tts pipeline agent in python

Thanks for reaching out, Aditya. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

is it possible to pass the usage cost and the transcription to the client when the session ends ?

You're asking whether you can pass usage cost and transcription to the client when the session ends — yes, you can. Here’s a concise approach and a minimal Python example pattern. What to collect • Transcription: use

session.history

or listen to

user_input_transcribed

/

conversation_item_added

events and build the transcript. • Usage: use

metrics.UsageCollector()

and collect

metrics_collected

events to accumulate STT/LLM/TTS metrics. How to deliver to the client • At session end, gather the transcript and usage summary, compute cost using your pricing rules, then send to the client via your app channel (HTTP response, WebSocket message, webhook, or store a file and return a download link). Minimal Python pattern

from livekit.agents import metrics, MetricsCollectedEvent
import logging
import json

logger = logging.getLogger(__name__)
usage_collector = metrics.UsageCollector()

@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
    usage_collector.collect(ev.metrics)

# build transcript live or read at end
def get_transcript():
    return session.history.to_dict()

async def on_session_end():
    summary = usage_collector.get_summary()
    transcript = get_transcript()
    # compute cost from `summary` using your pricing rules
    cost_estimate = compute_cost(summary)  # implement per your pricing
    payload = {"usage": summary, "cost": cost_estimate, "transcript": transcript}
    # send to client — choose your delivery:
    # - HTTP: POST to client webhook
    # - WebSocket: send message to connected socket
    # - Storage: save JSON to cloud storage and return link
    await send_to_client(payload)

Would you like a sample

compute_cost

implementation (given your pricing) or example code for sending the payload via WebSocket, HTTP webhook, or cloud storage? Sources: Events and error handling | LiveKit docs | Observability data hooks | LiveKit docs | Text and transcriptions | LiveKit docs

tell me more about livekit webhooks

You asked for more about LiveKit webhooks — here’s a focused summary of what they are, how they work, and how you can use them to deliver session data (like usage cost and transcripts) to your backend or clients. What LiveKit webhooks are • Server-sent HTTP POST notifications your backend receives when LiveKit room or media events occur. • Useful for automation: analytics, billing, recording/egress lifecycle, notifications, and triggering downstream workflows. Key events you’ll care about • Room lifecycle:

room_started

,

room_finished

• Participant lifecycle:

participant_joined

,

participant_left

• Track lifecycle:

track_published

,

track_unpublished

• Egress/ingress:

egress_started

,

egress_updated

,

egress_ended

,

ingress_started

,

ingress_ended

• Agent/AI events (in Agents):

conversation_item_added

,

speech_created

,

function_tools_executed

— useful for transcripts or AI outputs. Configuration options • LiveKit Cloud: configure webhooks in the project dashboard. • Self-hosted: configure in server config (yaml) with webhook URLs, api_key, and other options. • Per-egress request: you can pass a

WebhookConfig

to get egress-specific callbacks. Payload and security • POST with Content-Type: application/webhook+json containing a

WebhookEvent

JSON object. • Requests include an Authorization header (JWT) you must validate; SDK receiver libraries exist for Node, Go, Java, etc., to validate signatures for authenticity. Delivery guarantees and retry behavior • At-least-once delivery. LiveKit retries on non-2xx responses with exponential backoff. • Events are queued per resource; queues have limits (

max_age

,

max_depth

) to avoid unbounded memory use. • You can configure retry and queue settings (MaxRetries, RetryWaitMin, RetryWaitMax). Common implementation pattern for session data 1. Collect session metrics (UsageCollector or agent metrics collection) and transcripts during the session. 2. On

room_finished

or

participant_left

, receive the webhook in your backend. 3. Combine webhook metadata (timestamps, participant IDs) with your collected metrics/transcript. 4. Compute cost using your pricing rules. 5. Deliver final data to client via: ◦ HTTP response/webhook to client backend ◦ WebSocket message to connected frontend ◦ Store file (JSON) in cloud storage and return a download link Reliability and best practices • Validate webhook signature and reject invalid requests. • Respond quickly (2xx) to avoid retries; offload heavy processing to a background job. • Deduplicate idempotently since delivery is at-least-once. • Use per-resource queues and monitor for dropped events (tunable via config). If you want, I can: • Show a small example receiver in Python or Node that validates the webhook and ties it to stored session metrics/transcripts. • Draft a webhook payload mapping for computing cost from

room_finished

+ your

UsageCollector

summary. Which would you like next? Sources: Webhooks | LiveKit docs | examples/survey | livekit/livekit_composite | DeepWiki | Egress API | LiveKit docs | server-sdk-js | livekit/livekit_composite | DeepWiki | Agents Framework | livekit/livekit_composite | DeepWiki

what are the events supported and give me an example payload for each event.

You asked for the events LiveKit supports and an example payload for each — below is a concise list of common webhook events with sample JSON payloads you can use for testing or wiring your receiver. Room lifecycle 1. room_started

{
  "event": "room_started",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room",
    "empty_timeout": 300,
    "creation_time": 1678886600
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886600
}

1. room_finished

{
  "event": "room_finished",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room",
    "empty_timeout": 300,
    "creation_time": 1678886600
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678887200
}

Participant lifecycle 1. participant_joined

{
  "event": "participant_joined",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "participant": {
    "sid": "PA_XXXXXXXXXXXX",
    "identity": "user-identity",
    "name": "User Name",
    "state": "ACTIVE",
    "tracks": [],
    "metadata": "user-metadata"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886610
}

1. participant_left

{
  "event": "participant_left",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "participant": {
    "sid": "PA_XXXXXXXXXXXX",
    "identity": "user-identity",
    "name": "User Name"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678887100
}

1. participant_connection_aborted

{
  "event": "participant_connection_aborted",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "participant": {
    "sid": "PA_XXXXXXXXXXXX",
    "identity": "user-identity"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886605
}

Track lifecycle 1. track_published

{
  "event": "track_published",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "participant": {
    "sid": "PA_XXXXXXXXXXXX",
    "identity": "user-identity"
  },
  "track": {
    "sid": "TR_XXXXXXXXXXXX",
    "type": "video",
    "name": "camera",
    "source": "camera"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886620
}

1. track_unpublished

{
  "event": "track_unpublished",
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "participant": {
    "sid": "PA_XXXXXXXXXXXX",
    "identity": "user-identity"
  },
  "track": {
    "sid": "TR_XXXXXXXXXXXX",
    "type": "video",
    "name": "camera"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678887050
}

Egress (recording/streaming) 1. egress_started

{
  "event": "egress_started",
  "egress_info": {
    "egress_id": "EG_XXXXXXXXXXXX",
    "room_name": "my-room",
    "status": "active",
    "started_at": 1678886630
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886631
}

1. egress_updated

{
  "event": "egress_updated",
  "egress_info": {
    "egress_id": "EG_XXXXXXXXXXXX",
    "room_name": "my-room",
    "status": "active",
    "updated_at": 1678886700
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886701
}

1. egress_ended

{
  "event": "egress_ended",
  "egress_info": {
    "egress_id": "EG_XXXXXXXXXXXX",
    "room_name": "my-room",
    "status": "ended",
    "ended_at": 1678887000,
    "error": ""
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678887001
}

Ingress (RTMP/WHIP ingestion) 1. ingress_started

{
  "event": "ingress_started",
  "ingress_info": {
    "ingress_id": "IG_XXXXXXXXXXXX",
    "room_name": "my-room",
    "participant_identity": "ingress-participant",
    "stream_key": "my-stream-key",
    "input_type": "RTMP_INPUT",
    "state": {
      "status": "active"
    }
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886640
}

1. ingress_ended

{
  "event": "ingress_ended",
  "ingress_info": {
    "ingress_id": "IG_XXXXXXXXXXXX",
    "room_name": "my-room",
    "participant_identity": "ingress-participant",
    "stream_key": "my-stream-key",
    "input_type": "RTMP_INPUT",
    "state": {
      "status": "ended",
      "ended_reason": "publisher_left"
    }
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886900
}

Agent / AI-related events (useful with Agents) 1. conversation_item_added

{
  "event": "conversation_item_added",
  "conversation_item": {
    "id": "CI_XXXXXXXXXXXX",
    "role": "assistant",
    "content": "Transcribed or generated text here",
    "created_at": 1678886650
  },
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886651
}

1. speech_created

{
  "event": "speech_created",
  "speech": {
    "id": "SP_XXXXXXXXXXXX",
    "participant_sid": "PA_XXXXXXXXXXXX",
    "text": "Transcription text",
    "language": "en-US",
    "created_at": 1678886660
  },
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886661
}

1. function_tools_executed

{
  "event": "function_tools_executed",
  "tool_execution": {
    "id": "FT_XXXXXXXXXXXX",
    "tool_name": "my-tool",
    "input": {
      "foo": "bar"
    },
    "result": {
      "baz": "qux"
    },
    "created_at": 1678886670
  },
  "room": {
    "sid": "RM_XXXXXXXXXXXX",
    "name": "my-room"
  },
  "id": "WH_XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "created_at": 1678886671
}

Common fields included in webhook payloads • `id`: webhook event UUID • `created_at`: UNIX timestamp (seconds) • `event`: event type •

room

,

participant

,

track

,

egress_info

,

ingress_info

,

conversation_item

, etc., depending on event type Security and delivery notes • Requests use Content-Type: application/webhook+json and include an Authorization header (JWT) for signature validation. • Delivery is at-least-once; implement idempotent handling and signature verification, and return 2xx quickly to avoid retries. If you want, I can: • Provide a Python or Node example that validates the signature and parses these events. • Produce a compact mapping of event -> relevant fields to help you implement processing logic. Which would you prefer? Sources: Webhooks | LiveKit docs

does egress ended has the uploaded s3 url in the payload ?

No — the

egress_ended

webhook payload does not include the uploaded S3 URL. How to get the final output location: • Record the output details (bucket/filepath) you send in the original egress start request — that determines where the file will be saved. • Or call the ListEgress / Get Egress API after egress completes to retrieve final output details. If you want, I can show: • an example of capturing the output location from the egress start response, or • a sample call to the Egress API to fetch completed output URLs. Which would help more? Sources: Webhooks | LiveKit docs | RoomComposite Egress | LiveKit Docs | Egress API | LiveKit docs

{
  "event": "egress_ended",
  "event_id": "EVT_xxxxx",
  "created_at": 1734512345,
  "egressInfo": {
    "egressId": "EG_abc123",
    "roomName": "demo_call_42",
    "roomId": "RM_xxxxx",
    "status": "EGRESS_COMPLETE",
    "startedAt": 1734512000,
    "endedAt": 1734512300,
    "file": {
      "filename": "room-demo_call_42-1734512000.mp4",
      "location": "s3",
      "duration": 300,
      "size": 48213412,
      "output": {
        "s3": {
          "bucket": "your-recordings-bucket",
          "key": "recordings/2025/01/room-demo_call_42.mp4",
          "region": "us-east-1",
          "endpoint": "",
          "accessKey": ""
        }
      }
    },
    "error": ""
  }
}

Is this correct egress output payload ?

You're asking whether that JSON is a correct

egress_ended

payload — it's not. LiveKit's

egress_ended

webhook does not include the S3 output details (bucket/key/URL/credentials) in the payload. What to expect instead •

egress_ended

contains metadata about the egress (egressId, roomName/roomId, status, startedAt/endedAt, duration/size may be present) but it does NOT include the

output.s3

object or signed S3 URL/credentials. How to get the output location • Record the output config you sent when starting egress (the bucket/key/filepath you requested). • Or call the Egress API (ListEgress / Get Egress) after completion to retrieve final output details. If you want, I can: • Show a corrected minimal

egress_ended

example without S3 details, and • Provide sample code to fetch the egress output via the Egress API or capture output from the egress start response. Which would you like?