import asyncio import json import logging from signal import SIGINT, SIGTERM from typing import Union import os from livekit import api, rtc # ensure LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET are set LIVEKIT_URL = "xxxxxxxxx" API_KEY = "xxxxxxx" API_SECRET = "xxxxxx" async def main(room: rtc.Room) -> None: @room.on("participant_connected") def on_participant_connected(participant: rtc.RemoteParticipant) -> None: logging.info("participant connected: %s %s", participant.sid, participant.identity) @room.on("participant_disconnected") def on_participant_disconnected(participant: rtc.RemoteParticipant): logging.info("participant disconnected: %s %s", participant.sid, participant.identity) @room.on("local_track_published") def on_local_track_published( publication: rtc.LocalTrackPublication, track: Union[rtc.LocalAudioTrack, rtc.LocalVideoTrack], ): logging.info("local track published: %s", publication.sid) @room.on("active_speakers_changed") def on_active_speakers_changed(speakers: list[rtc.Participant]): logging.info("active speakers changed: %s", speakers) @room.on("local_track_unpublished") def on_local_track_unpublished(publication: rtc.LocalTrackPublication): logging.info("local track unpublished: %s", publication.sid) @room.on("track_published") def on_track_published( publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant ): logging.info( "track published: %s from participant %s (%s)", publication.sid, participant.sid, participant.identity, ) @room.on("track_unpublished") def on_track_unpublished( publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant ): logging.info("track unpublished: %s", publication.sid) @room.on("track_subscribed") def on_track_subscribed( track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, ): logging.info("track subscribed: %s", publication.sid) if track.kind == rtc.TrackKind.KIND_VIDEO: _video_stream = rtc.VideoStream(track) # video_stream is an async iterator that yields VideoFrame elif track.kind == rtc.TrackKind.KIND_AUDIO: print("Subscribed to an Audio Track") _audio_stream = rtc.AudioStream(track) # audio_stream is an async iterator that yields AudioFrame @room.on("track_unsubscribed") def on_track_unsubscribed( track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, ): logging.info("track unsubscribed: %s", publication.sid) @room.on("track_muted") def on_track_muted(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant): logging.info("track muted: %s", publication.sid) @room.on("track_unmuted") def on_track_unmuted( publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant ): logging.info("track unmuted: %s", publication.sid) @room.on("data_received") def on_data_received(data: rtc.DataPacket): logging.info("received data from %s: %s", data.participant.identity, data.data) @room.on("connection_quality_changed") def on_connection_quality_changed(participant: rtc.Participant, quality: rtc.ConnectionQuality): logging.info("connection quality changed for %s", participant.identity) @room.on("track_subscription_failed") def on_track_subscription_failed( participant: rtc.RemoteParticipant, track_sid: str, error: str ): logging.info("track subscription failed: %s %s", participant.identity, error) @room.on("connection_state_changed") def on_connection_state_changed(state: rtc.ConnectionState): logging.info("connection state changed: %s", state) @room.on("connected") def on_connected() -> None: logging.info("connected") @room.on("disconnected") def on_disconnected() -> None: logging.info("disconnected") @room.on("reconnecting") def on_reconnecting() -> None: logging.info("reconnecting") @room.on("reconnected") def on_reconnected() -> None: logging.info("reconnected") @room.on("room_updated") def on_room_updated() -> None: logging.info(f"room updated, participants: {room.num_participants}") metadata = json.dumps( { "is_preview": True, "bot_external_id": "AIBOT-43D15653C7", } ) token = ( api.AccessToken(API_KEY, API_SECRET) .with_identity("python-bot") .with_name("Python Bot") .with_grants( api.VideoGrants( room_join=True, room="my-room", ) ).with_metadata(metadata) .to_jwt() ) await room.connect(LIVEKIT_URL, token) logging.info(f"connected to room {room.name}, created {room.creation_time}") logging.info(f"participants: {room.remote_participants}") await asyncio.sleep(2) await room.local_participant.publish_data("hello world") if name == "__main__": logging.basicConfig( level=logging.INFO, handlers=[logging.FileHandler("basic_room.log"), logging.StreamHandler()], ) loop = asyncio.get_event_loop() room = rtc.Room(loop=loop) async def cleanup(): await room.disconnect() loop.stop() asyncio.ensure_future(main(room)) for signal in [SIGINT, SIGTERM]: loop.add_signal_handler(signal, lambda: asyncio.ensure_future(cleanup())) try: loop.run_forever() finally: loop.close() how to print the audio transcription

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

give the updated code

Thanks — I’ll update your existing Python client to print audio transcriptions. I’ll provide two runnable options so you can pick what fits your setup: • Option A (recommended): Use a LiveKit Agent with Deepgram STT — easiest, server-side, less low-level audio handling. • Option B: Directly stream audio frames from your client to Deepgram’s realtime API — works from your client/backend but requires encoding frames and a Deepgram realtime WebSocket. Pick the option you prefer. Below are both complete examples. Both assume you have keys for Deepgram (set DEEPGRAM_API_KEY) and LiveKit creds (LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET). Replace placeholders with your values. Option A — LiveKit Agent + Deepgram (recommended) • Installs: pip install "livekit-agents[deepgram]" • agent_transcriber.py:

import asyncio
import os
from livekit.agents import AgentSession, JobContext, server
from livekit.plugins import deepgram, silero

# Ensure environment variables:
# LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, DEEPGRAM_API_KEY

@server.rtc_session(agent_name="transcriber")
async def entrypoint(ctx: JobContext):
    # create an AgentSession that uses Deepgram STT and optional VAD
    session = AgentSession(
        stt=deepgram.STT(api_key=os.environ.get("DEEPGRAM_API_KEY")),
        vad=silero.VAD.load()
    )

    # Listen for transcriptions from the AgentSession
    @session.on("user_input_transcribed")
    def on_transcribed(transcript):
        # transcript.is_final indicates the final result for an utterance
        if getattr(transcript, "is_final", False):
            print(f"[Transcription] {transcript.transcript}")

    # Start session attached to the same room the agent will join
    await session.start(agent=None, room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    # Run the agent server (this process will host the agent and connect to LiveKit)
    # The livekit-agents server entrypoint runs automatically; the decorator registers the session.
    # Use the agent runner/manager recommended by livekit-agents to run this file.
    print("Start your LiveKit Agents runner to host this transcriber session.")

Notes: • This example uses livekit-agents plugin helpers; the AgentSession will subscribe to user audio automatically and call

user_input_transcribed

when Deepgram returns text. • Print statements in

on_transcribed

output final transcripts. Option B — Client-side: read rtc.AudioStream frames and stream to Deepgram Realtime • Installs: pip install livekit-sdk deepgram-sdk websockets soundfile aiohttp • This example connects to LiveKit, subscribes to remote audio, converts frames to 16-bit PCM mono 16kHz if needed, and streams to Deepgram Realtime via WebSocket. It prints partial/final transcriptions returned by Deepgram. • client_transcribe.py:

import asyncio
import json
import logging
import os
import websockets
import struct
from signal import SIGINT, SIGTERM
from typing import Union

from livekit import api, rtc

# Environment variables:
LIVEKIT_URL = os.environ.get("LIVEKIT_URL", "<https://your-livekit-url>")
API_KEY = os.environ.get("LIVEKIT_API_KEY", "YOUR_KEY")
API_SECRET = os.environ.get("LIVEKIT_API_SECRET", "YOUR_SECRET")
DEEPGRAM_API_KEY = os.environ.get("DEEPGRAM_API_KEY", "YOUR_DEEPGRAM_KEY")

logging.basicConfig(level=<http://logging.INFO|logging.INFO>, handlers=[logging.StreamHandler()])

async def deepgram_realtime(ws_url: str):
    """
    Opens a Deepgram realtime WebSocket. Returns the websocket.
    We will send raw PCM16 little-endian 16k samples to the ws.
    """
    headers = [("Authorization", f"Token {DEEPGRAM_API_KEY}")]
    ws = await websockets.connect(ws_url, extra_headers=headers)
    return ws

async def send_pcm_to_deepgram(ws, pcm_bytes: bytes):
    """
    Deepgram realtime expects JSON messages for config and binary for audio.
    Here we send raw binary audio frames.
    """
    try:
        await ws.send(pcm_bytes)
    except Exception:
        pass

async def deepgram_recv_loop(ws):
    async for msg in ws:
        try:
            data = json.loads(msg)
        except Exception:
            continue
        # Deepgram returns interim and final transcripts as JSON
        if "channel" in data and "alternatives" in data["channel"]:
            alt = data["channel"]["alternatives"][0]
            transcript = alt.get("transcript", "")
            is_final = data.get("is_final", False) or data.get("type") == "transcript_final"
            print(f"[Deepgram {'FINAL' if is_final else 'PARTIAL'}] {transcript}")

async def process_audio_track(track: rtc.Track):
    """
    Reads frames from rtc.AudioStream and forwards to Deepgram WS.
    You must ensure the audio format matches Deepgram's expectations (PCM16 16000Hz mono).
    This example assumes frames are PCM float32 samples. If not, adapt accordingly.
    """
    audio_stream = rtc.AudioStream(track)
    # Open a new Deepgram realtime connection per track (or reuse one)
    ws = await deepgram_realtime("<wss://api.deepgram.com/v1/listen?encoding=linear16&sample_rate=16000&channels=1>")
    # Start a receiving task
    asyncio.create_task(deepgram_recv_loop(ws))

    async for frame in audio_stream:
        # `frame` structure depends on the SDK; convert to 16-bit PCM little-endian.
        # Example conversion if frame.samples is a bytes-like PCM16 already:
        # pcm_bytes = frame.data
        # If frame provides floats, convert float32 -> int16:
        # Here we attempt to be generic:
        try:
            # Try float buffer conversion
            samples = frame.samples  # may be an ndarray or bytes
            # If samples is a bytes buffer of float32:
            if isinstance(samples, bytes):
                # Interpret as float32 little-endian
                count = len(samples) // 4
                floats = struct.unpack("<" + "f" * count, samples)
                ints = [max(-32768, min(32767, int(s * 32767))) for s in floats]
                pcm_bytes = struct.pack("<" + "h" * len(ints), *ints)
            else:
                # fallback: assume it's already PCM bytes
                pcm_bytes = bytes(samples)
        except Exception:
            # fallback attempt
            pcm_bytes = frame.data if hasattr(frame, "data") else b""

        if not pcm_bytes:
            continue

        await send_pcm_to_deepgram(ws, pcm_bytes)

    # close ws when done
    await ws.close()

async def main(room: rtc.Room):
    # Create LiveKit access token
    metadata = json.dumps({"is_preview": True, "bot_external_id": "AIBOT-43D15653C7"})
    token = (
        api.AccessToken(API_KEY, API_SECRET)
        .with_identity("python-bot")
        .with_name("Python Bot")
        .with_grants(api.VideoGrants(room_join=True, room="my-room"))
        .with_metadata(metadata)
        .to_jwt()
    )

    await room.connect(LIVEKIT_URL, token)
    <http://logging.info|logging.info>(f"connected to room {room.name}")

    # Subscribe to audio tracks and send to Deepgram
    @room.on("track_subscribed")
    def on_track_subscribed(track: rtc.Track, publication, participant):
        <http://logging.info|logging.info>("track_subscribed: %s kind=%s from %s", publication.sid, track.kind, participant.identity)
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            asyncio.create_task(process_audio_track(track))

    # keep the client running
    await asyncio.Event().wait()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    room = rtc.Room(loop=loop)

    async def cleanup():
        await room.disconnect()
        loop.stop()

    asyncio.ensure_future(main(room))
    for sig in (SIGINT, SIGTERM):
        loop.add_signal_handler(sig, lambda: asyncio.ensure_future(cleanup()))
    try:
        loop.run_forever()
    finally:
        loop.close()

Important notes for Option B: • You must ensure the audio frames provided by

rtc.AudioStream

are converted to the expected PCM format (16-bit little-endian, 16000Hz, mono) for Deepgram. The example includes a generic conversion attempt but you’ll likely need to adapt it depending on how the SDK exposes frame samples. • Using Deepgram realtime requires sending initial config query params (shown in the ws URL) and authenticating with your API key. • Deepgram enforces rate/format limits. For production, reuse sockets or batch frames carefully. Which option would you like as a final, more detailed runnable snippet? • I can expand Option A into a fully runnable agent script with exact livekit-agents runner instructions, or • I can tighten Option B to match the precise frame format produced by your environment (tell me whether frames are float32 arrays, PCM16 bytes, sample rate, and channels).