import asyncio import json from livekit import rtc, api from livekit.rtc import DataPacket # --- CONFIGURATION: PUT YOUR LIVEKIT DETAILS HERE --- LIVEKIT_URL = "xxxx" API_KEY = "xxxxx" API_SECRET = "xxxxxx" ROOM_NAME = "xxxxx" # ---------------------------------------------------- async def generate_token(): # Generates a token for our automated "bot" to join the room #metadata added by me (put actual values) metadata = json.dumps( { "is_preview": True, "bot_external_id": "Test-bot", } ) token = api.AccessToken(API_KEY, API_SECRET) token.with_identity("Preview_Mode_user_557") token.with_name("QA Tester") token.with_metadata(metadata) token.with_grants(api.VideoGrants(room_join=True, room=ROOM_NAME)) return token.to_jwt() async def test_welcome_message(): TOKEN = await generate_token() room = rtc.Room() print("Connecting to LiveKit...") await room.connect(LIVEKIT_URL, TOKEN) print("Connected.") welcome_received = asyncio.Event() welcome_text = {"data": None} @room.on("data_received") def on_data(packet: DataPacket): try: message = packet.data.decode() print("Received:", message) welcome_text["data"] = message welcome_received.set() except Exception as e: print("Error decoding message:", e) try: await asyncio.wait_for(welcome_received.wait(), timeout=30) except asyncio.TimeoutError: await room.disconnect() raise Exception("❌ No welcome message received") # Assertion if "hello" in welcome_text["data"].lower() or "welcome" in welcome_text["data"].lower(): print("✅ Welcome message received successfully") else: raise Exception("❌ Welcome message content not as expected") await room.disconnect() print("Test complete.") asyncio.run(test_welcome_message()) is this script correct getting No welcome message received

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.

did not get print inside connected eveent

Thanks — I hear you: your

connected

handler never printed, and you want to know why. Quick checklist and fixes to try (they're ordered from most common to less common): 1) Register the handler before calling

connect()

• The handler must be attached to the

Room

object before you call

await room.connect(...)

. If you register it after connect, it may never fire. 2) Use the correct event name / registration style • For the Python SDK, register like this:

from livekit import rtc

room = rtc.Room()

@room.on("connected")
def on_connected():
    print("Connected to room!")

Then call:

await room.connect(LIVEKIT_URL, TOKEN)

3) Confirm

connect()

actually succeeds • Make sure you see your earlier "Connecting to LiveKit..." and "Connected." prints. If

connect()

raises or fails, the

connected

event won't fire. • Add a try/except around

connect()

to log any exceptions. 4) Add other lifecycle handlers to detect failures • Add

disconnected

and

connection_error

style handlers to see if the client is disconnecting or failing:

@room.on("disconnected")
def on_disconnected():
    print("Disconnected")

@room.on("connection_error")
def on_conn_error(err):
    print("Connection error:", err)

5) Ensure your environment runs the event loop correctly • If you run synchronous code that blocks the asyncio loop, callbacks may not execute. Use

asyncio.run(...)

(which you already do) and avoid blocking calls between registering handlers and connecting. 6) Short test: publish a self-message after connect to confirm event loop and data handling • Once you confirm

on_connected

prints, add a self-publish to test data receipt:

@room.on("connected")
def on_connected():
    print("Connected to room!")
    asyncio.create_task(room.local_participant.publish_data(b"test", topic="test"))

7) Verify token/URL and network • If the token is invalid or the URL is unreachable, the client may fail silently. Logging

connect()

exceptions will reveal that. If you want, paste the minimal code you’re running (the portion where you create

room

, register handlers, and call

connect

) and I’ll point out any placement or ordering issues. Which of the checks above have you tried so far? Sources: Handling asynchronous metadata updates in LiveKit | LiveKit | client-sdk-js | livekit/livekit_composite | DeepWiki

how to use track_published event

Thanks — I’ll show you how to use the

track_published

event in the LiveKit Python SDK and how to act when a remote participant publishes a track. Short summary: register the handler on the

Room

before connecting. The handler receives a

RemoteTrackPublication

and the

RemoteParticipant

. You can inspect

publication.kind

(audio/video), access

publication.track

once subscribed, and then consume frames or attach rendering/processing logic. Example (minimal, complete pattern)

import asyncio
from livekit import rtc

LIVEKIT_URL = "<wss://your-livekit-url>"
TOKEN = "YOUR_TOKEN"

room = rtc.Room()

@room.on("track_published")
def on_track_published(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
    print(f"Track published: sid={publication.sid}, kind={publication.kind}, from={participant.identity}")

    # If the track is already subscribed, publication.track may be non-None.
    # For audio you might process frames; for video you might render frames.
    if publication.kind == rtc.TrackKind.KIND_AUDIO:
        audio_track = publication.track
        if audio_track is not None:
            print("Audio track already available — create AudioStream to consume frames")
            audio_stream = rtc.AudioStream(audio_track)

            async def process_audio():
                async for frame in audio_stream:
                    # handle frame (example: print timestamp)
                    print("audio frame ts:", frame.timestamp)

            asyncio.create_task(process_audio())
        else:
            print("Audio track not yet subscribed — wait for subscription or call subscribe() if needed")

    elif publication.kind == rtc.TrackKind.KIND_VIDEO:
        video_track = publication.track
        if video_track is not None:
            print("Video track available — create VideoStream to consume frames")
            video_stream = rtc.VideoStream(video_track)

            async def process_video():
                async for frame in video_stream:
                    # handle video frame
                    print("video frame:", frame.height, "x", frame.width)

            asyncio.create_task(process_video())
        else:
            print("Video track not yet subscribed — wait for subscription or call subscribe() if needed")

@room.on("connected")
def on_connected():
    print("Connected to room")

async def run():
    await room.connect(LIVEKIT_URL, TOKEN)
    # keep running so async tasks can process frames
    await asyncio.Event().wait()

asyncio.run(run())

Notes and tips • Register the

track_published

handler before

room.connect(...)

. •

publication.track

may be None if you're not automatically subscribed to remote tracks. If you need to force subscription, check the SDK methods for subscribing to a publication (or ensure auto-subscribe is enabled). • Use

rtc.AudioStream(publication.track)

and

rtc.VideoStream(publication.track)

to iterate frames asynchronously. • You can also listen for

track_subscribed

or similar events if you want to act only after subscription is complete. • If you only need metadata (e.g., track name, source), inspect

publication.name

,

publication.sid

, and

publication.source

. If you want, share whether you want to: • auto-subscribe to incoming tracks, • process audio for transcription, • or render video frames in a GUI — and I’ll provide a tailored snippet for that use case. Sources: An open source stack for real-time multimodal AI | Job lifecycle | LiveKit Documentation | Capturing the First Few Seconds of a Call for Processing

not getting print("Connected to room")

Thanks — I know it’s frustrating when the

connected

handler never fires. I’ll help you narrow this down step‑by‑step and give a minimal example you can run to verify the event loop, registration order, and connection. Quick summary of likely causes • The handler was registered after calling

connect()

(must register before). •

connect()

failed (invalid token, network error) so the connected event never emits. • The event loop is blocked or the handler signature/registration style is incorrect. • You're not seeing logs because exceptions are swallowed. Please try the checklist and the minimal example below. Checklist — try these first 1. Register the

@room.on("connected")

handler before calling

await room.connect(...)

. 2. Print/log immediately before and after

await room.connect(...)

to confirm whether

connect()

completes: ◦ If you see the "Connected." print after

connect()

but not the handler, registration order is wrong. ◦ If you do not see it,

connect()

likely failed — capture the exception. 1. Wrap

connect()

in try/except and print the exception to reveal token/connection errors. 2. Ensure you’re using

asyncio.run(...)

and not blocking the event loop elsewhere. 3. Confirm the token and URL are valid and that you see an initial "Connecting..." print. Minimal, self-contained example to copy/paste • This registers the handler first, logs exceptions, and keeps the program alive so handlers can run.

import asyncio
from livekit import rtc

LIVEKIT_URL = "<wss://your-livekit-url>"   # replace
TOKEN = "YOUR_TOKEN"                     # replace

room = rtc.Room()

@room.on("connected")
def on_connected():
    print("Connected to room (handler fired)")

@room.on("disconnected")
def on_disconnected():
    print("Disconnected")

@room.on("connection_error")
def on_conn_error(err):
    print("Connection error:", err)

async def main():
    print("About to connect...")
    try:
        await room.connect(LIVEKIT_URL, TOKEN)
        print("connect() returned successfully")
    except Exception as e:
        print("connect() raised exception:", e)
        return

    # keep running so event callbacks can run
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

How to interpret results • You see "connect() returned successfully" and also "Connected to room (handler fired)" → success. • You see "connect() returned successfully" but NOT the handler → handler registration likely occurred after connect() in your original code; ensure registration is before

connect()

. • You see an exception from

connect()

→ token/URL/network issue. Paste the exception text and I’ll help interpret it. • No prints at all or program hangs → event loop issues or your process environment is swallowing stdout. Try a simple asyncio script that just prints to confirm the environment. If the handler still doesn’t fire after running the snippet • Paste the exact output you get (including any exception stack traces). • Paste the short snippet you’re actually running (the part where you create

room

, register handlers, and call

connect

) so I can check ordering and signatures. Would you like me to: • Check your token validity with a code snippet to decode/check expiration, or • Provide a companion script that connects as a second participant and publishes a track/message so you can test event flow? Tell me which output you saw from the minimal example and I’ll guide the next step.

connection is successfull also request is received on backend server but events are not working

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!