https://livekit.io logo
Join Slack
Powered by
i am trying to create a sip particiapnt using the ...
# ask-ai
b
i am trying to create a sip particiapnt using the below approach:
Copy code
# Create SIP participant via LiveKit API
            await ctx.api.sip.create_sip_participant(
                api.CreateSIPParticipantRequest(
                    room_name=room_name,
                    sip_trunk_id=trunk_id,
                    sip_call_to=dial_number,
                    participant_identity=participant_identity,
                    participant_name=f"Customer",
                    max_call_duration=timedelta(seconds=180), # 3 minutes max
                    wait_until_answered=True
                )
            )
how can i know if the user picks the call, is busy or rejected?
t
Thanks for reaching out, Utkarsh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
You're asking how to detect whether an outbound SIP call created with your snippet was answered, busy, or rejected. Below are the practical ways to detect each outcome. 1) Use wait_until_answered=True (your current call) • If the call connects, the 
create_sip_participant
call returns successfully. • At that point the participant’s 
sip.callStatus
will be 
active
(or 
automation
briefly for some automated flows). 2) Handle errors for busy / rejected / failed • If the call fails (rejected, busy, invalid number, etc.), 
create_sip_participant
raises an error you can catch. • Inspect the error metadata for SIP status information: 
sip_status_code
and 
sip_status
. • Common SIP codes to check: ◦ 
486
— Busy (Destination busy) ◦ 
487
— Request terminated (canceled/rejected) ◦ 
404
— Not found (invalid number) ◦ Other 4xx/5xx codes indicate other failures Example pattern (Python-like):
Copy code
try:
    await ctx.api.sip.create_sip_participant( ... wait_until_answered=True ...)
    # call answered -> participant connected, sip.callStatus == "active"
except api.TwirpError as e:
    sip_status_code = e.metadata.get("sip_status_code")
    sip_status = e.metadata.get("sip_status")
    # inspect sip_status_code to distinguish busy/rejected/not found/etc.
3) Real-time monitoring (alternative to blocking) • Set 
wait_until_answered=False
and watch participant events/attributes. • Monitor the participant’s 
sip.callStatus
values: ◦ 
dialing
/ 
ringing
— call in progress ◦ 
active
— answered/connected ◦ 
hangup
— call ended • Use these events if you need non-blocking behavior or finer-grained state transitions. If you’d like, tell me which approach you prefer (blocking with 
wait_until_answered
or non-blocking event monitoring) and I’ll show a precise code example for that path. Sources: SIP participant | LiveKit docs | Make outbound calls | LiveKit docs
b
i would like to see both the approaches
t
Thanks — I’ll show both approaches: (A) blocking with 
wait_until_answered=True
(simple, returns or errors) and (B) non‑blocking event/attribute monitoring (more control, real‑time states). Each example is Python-style matching your snippet. A — Blocking (wait until answered) 1. Call 
create_sip_participant
with 
wait_until_answered=True
. 2. If it returns, the call was answered (participant 
sip.callStatus
becomes 
active
). 3. If it raises an error, inspect 
e.metadata
for 
sip_status_code
/ 
sip_status
to distinguish busy/rejected/etc. Example:
Copy code
from datetime import timedelta

try:
    participant = await ctx.api.sip.create_sip_participant(
        api.CreateSIPParticipantRequest(
            room_name=room_name,
            sip_trunk_id=trunk_id,
            sip_call_to=dial_number,
            participant_identity=participant_identity,
            participant_name="Customer",
            max_call_duration=timedelta(seconds=180),
            wait_until_answered=True
        )
    )
    # Success -> call answered
    print("Call answered")
    # participant object available; you can check participant.metadata or attributes
    # e.g., participant.metadata.get("sip.callStatus") or query the participant later
except api.TwirpError as e:
    sip_status_code = e.metadata.get("sip_status_code")
    sip_status = e.metadata.get("sip_status")
    print(f"Call failed: {sip_status_code} {sip_status}")
    if sip_status_code == "486":
        print("Busy")
    elif sip_status_code == "487":
        print("Rejected / canceled")
    elif sip_status_code == "404":
        print("Number not found")
    else:
        print("Other failure")
When to use: quick implementation when you want the request to complete only once the call is answered (or definitively fails). Simpler server flow but ties up the caller until answer/failure. B — Non‑blocking (monitor events / participant attributes) 1. Create the SIP participant without blocking: 
wait_until_answered=False
(default). 2. Subscribe to room/participant events or poll participant attributes to observe 
sip.callStatus
transitions: 
dialing
, 
ringing
, 
active
, 
hangup
, etc. 3. Use those transitions to detect answered, busy, rejected, or ended. Example pattern:
Copy code
# 1) Create without waiting
await ctx.api.sip.create_sip_participant(
    api.CreateSIPParticipantRequest(
        room_name=room_name,
        sip_trunk_id=trunk_id,
        sip_call_to=dial_number,
        participant_identity=participant_identity,
        participant_name="Customer",
        max_call_duration=timedelta(seconds=180),
        wait_until_answered=False
    )
)

# 2) In your room/participant event handler (pseudo-code):
def on_participant_updated(participant):
    call_status = participant.metadata.get("sip.callStatus")  # or attribute on participant
    if call_status == "dialing":
        print("Dialing...")
    elif call_status == "ringing":
        print("Ringing at destination")
    elif call_status == "active":
        print("Call answered / connected")
    elif call_status == "hangup":
        # look for hangup reason or final sip status if provided
        print("Call ended")
Or subscribe to the room’s real-time participant events (your client/agent library event callbacks). Use those events to update UI, trigger timers, or log final SIP codes if your integration surfaces them in participant metadata/events. When to use: you need non-blocking behavior, want to update UI in real time, or need intermediate states like ringing/dialing. Slightly more complex but gives full lifecycle visibility. Notes and tips • Error metadata contains SIP response codes. Check 
sip_status_code
to detect busy (
486
), canceled/rejected (
487
), not found (
404
), etc. • 
sip.callStatus == "active"
indicates the call is connected. • Use blocking for simple flows; use event monitoring for real-time UI or complex logic. Which example would you like next: a full runnable server example showing event subscription, or a more detailed client-side participant event handler?
b
this is the code:
Copy code
# Create SIP participant via LiveKit API
            try:
                participant = await ctx.api.sip.create_sip_participant(
                api.CreateSIPParticipantRequest(
                    room_name=room_name,
                    sip_trunk_id=trunk_id,
                    sip_call_to=dial_number,
                    participant_identity=participant_identity,
                    participant_name=f"Customer_{call_context.customer_data.customer_name}",
                    max_call_duration=timedelta(seconds=180), # 3 minutes max
                    wait_until_answered=True
                )
            )
            
                <http://self.logger.info|self.logger.info>(
                    "[%s] ✅ SIP participant created: %s",
                    room_name,
                    participant_identity
                )
                <http://self.logger.info|self.logger.info>("**************Logging value of participant: %s", participant)
                return participant_identity
            except api.TwirpError as e:
                self.logger.error(
                    "[%s] ❌ Failed to create SIP participant: %s",
                    room_name,
                    e,
                    exc_info=True
                )
                raise RuntimeError(f"Failed to initiate outbound call: {e}") from e
{"level": "INFO", "name": "orchestrator.call_orch_service", "message": "**************Logging value of participant: participant_id: \"PA_kjjkvQ3Y6aAT\"\nparticipant_identity: \"sip_target_1234567890\"\nroom_name: \"collection-call-3148067097691554840-1765948700\"\nsip_call_id: \"SCL_nDQABwRQk457\"\n", "pid": 258, "job_id": "AJ_EG29f8wxUdD4", "room_id": "RM_CD8iPTnSg4ck", "timestamp": "2025-12-17T05:18:28.725359+00:00"}
but i do not see any metadata field
t
Thanks — I looked at your code and logs. You're seeing the basic SIP participant fields from 
create_sip_participant
(id, identity, sip_call_id) because that API returns only the initial participant info synchronously. The SIP call state is not included there as metadata. What to do next (two practical options): 1) Blocking flow (you already use this) • 
wait_until_answered=True
will only return successfully once the call is answered (or definitively fails). A successful return means the call is connected; check the participant later and you should see 
sip.callStatus == "active"
. • Immediate failures (invalid number, trunk error, provider rejections) raise 
api.TwirpError
. Inspect 
e.metadata.get("sip_status_code")
and 
e.metadata.get("sip_status")
to detect busy (486), canceled/rejected (487), not found (404), etc. 2) Non‑blocking / real‑time monitoring (recommended for full lifecycle) • Create the SIP participant with 
wait_until_answered=False
. • Subscribe to participant attribute change events or poll 
GetParticipant
. Watch the 
sip.callStatus
attribute for these transitions: ◦ 
dialing
— initiating ◦ 
ringing
— remote is ringing ◦ 
active
— answered/connected ◦ 
automation
— automated flow / DTMF ◦ 
hangup
— call ended/rejected • Example pattern:
Copy code
# create without waiting
await ctx.api.sip.create_sip_participant(..., wait_until_answered=False)

# later, get participant state
res = await ctx.api.room.get_participant(RoomParticipantIdentity(room=room_name, identity=participant_identity))
attrs = res.participant.attributes
status = attrs.get("sip.callStatus")
Or add an event handler to receive 
participant attributes changed
events and react to 
sip.callStatus
updates in real time. Debug tips • Log 
e.metadata
inside your exception handler when 
create_sip_participant
fails to capture SIP error codes. • Log attribute-change events after creation to confirm updates are arriving asynchronously. • Don’t expect call status to appear on the initial 
participant
return value — it’s populated later via attributes/events. If you want, I can: • Show a concrete async example that subscribes to participant attribute events, or • Show a fuller runnable snippet that calls 
get_participant
until 
sip.callStatus
transitions. Which would help you most? Sources: Making outbound calls | LiveKit docs | Make outbound calls | LiveKit docs | SIP participant | LiveKit docs | Handling events | LiveKit docs | Participant attributes and metadata | LiveKit docs
b
yes show the implementation for : • Show a fuller runnable snippet that calls 
get_participant
until 
sip.callStatus
transitions. Which would help you most?
t
Below is a concise async Python example that: • creates a SIP participant without blocking, • polls 
get_participant
until 
sip.callStatus
reaches a terminal/connected state or times out, • inspects Twirp error metadata for immediate failures. Adjust imports and API client calls to match your environment.
Copy code
import asyncio
from datetime import timedelta, datetime, timezone

# replace `api` and ctx.api with your actual LiveKit client objects
# this example assumes `ctx.api.sip.create_sip_participant` and `ctx.api.room.get_participant`
# behave like in your snippets and return the shown objects.

POLL_INTERVAL = 1.0  # seconds between polls
POLL_TIMEOUT = 30.0  # max seconds to wait for a status transition

async def create_and_wait_for_call(ctx, room_name, trunk_id, dial_number, participant_identity):
    try:
        # Create participant non-blocking (we'll poll for status)
        await ctx.api.sip.create_sip_participant(
            api.CreateSIPParticipantRequest(
                room_name=room_name,
                sip_trunk_id=trunk_id,
                sip_call_to=dial_number,
                participant_identity=participant_identity,
                participant_name=f"Customer_{participant_identity}",
                max_call_duration=timedelta(seconds=180),
                wait_until_answered=False
            )
        )
    except api.TwirpError as e:
        # Immediate failure from provider; inspect metadata
        sip_code = e.metadata.get("sip_status_code")
        sip_msg = e.metadata.get("sip_status")
        raise RuntimeError(f"Immediate SIP failure: {sip_code} {sip_msg}") from e

    # Poll for participant attributes until callStatus changes to a terminal state
    deadline = datetime.now(timezone.utc).timestamp() + POLL_TIMEOUT
    last_status = None

    while True:
        # timeout
        if datetime.now(timezone.utc).timestamp() > deadline:
            raise TimeoutError(f"Timed out waiting for sip.callStatus (last={last_status})")

        # fetch participant state
        res = await ctx.api.room.get_participant(
            api.RoomParticipantIdentity(room=room_name, identity=participant_identity)
        )

        participant = res.participant
        attrs = participant.attributes or {}
        call_status = attrs.get("sip.callStatus")

        if call_status != last_status:
            # log or act on state changes
            <http://ctx.logger.info|ctx.logger.info>("Participant %s sip.callStatus -> %s", participant_identity, call_status)
            last_status = call_status

        # Interpret statuses
        if call_status == "active":
            # answered / connected
            return {"status": "answered", "participant": participant}
        if call_status == "ringing":
            # still ringing; continue polling
            pass
        if call_status == "dialing":
            pass
        if call_status == "automation":
            # DTMF/automation in progress
            pass
        if call_status == "hangup":
            # final hangup - may include final SIP error details in participant attributes or logs
            return {"status": "hangup", "participant": participant}
        # If provider returns explicit SIP result codes via attributes, inspect them:
        sip_code_attr = attrs.get("sip.status_code") or attrs.get("sip_status_code")
        if sip_code_attr:
            # map common codes
            if sip_code_attr == "486":
                return {"status": "busy", "code": sip_code_attr, "participant": participant}
            if sip_code_attr == "487":
                return {"status": "rejected", "code": sip_code_attr, "participant": participant}
            if sip_code_attr == "404":
                return {"status": "not_found", "code": sip_code_attr, "participant": participant}

        await asyncio.sleep(POLL_INTERVAL)
Notes: • Use 
wait_until_answered=False
so you receive the initial participant quickly and can poll. • The authoritative call state is 
sip.callStatus
in participant attributes; initial create response will not contain this. • Immediate provider failures may raise 
TwirpError
— check 
e.metadata
for 
sip_status_code
/ 
sip_status
. • Tune 
POLL_INTERVAL
and 
POLL_TIMEOUT
to your latency and UX needs. If you want, I can convert this to a runnable script with mock client stubs or show how to wire this into your existing event loop and logging.
b
what if i want to use wait_until_answered=True ?
t
Good — here's how 
wait_until_answered=True
behaves and a concise example showing how to use it and handle outcomes. Behavior summary • 
wait_until_answered=True
blocks the API call until the outbound SIP INVITE reaches a terminal connected state (answered/connected) or the provider returns an error immediately. • If it returns successfully, the call is connected (you can expect 
sip.callStatus
to be 
active
shortly after). • If the call fails immediately (invalid number, carrier rejection, busy in some cases), the call raises 
api.TwirpError
. Inspect 
e.metadata
for SIP details (
sip_status_code
, 
sip_status
) to detect busy (486), canceled/rejected (487), not found (404), etc. • The synchronous return value contains basic participant info; full 
sip.callStatus
may still be populated asynchronously — but successful return implies connection. Example (blocking)
Copy code
from datetime import timedelta

try:
    participant = await ctx.api.sip.create_sip_participant(
        api.CreateSIPParticipantRequest(
            room_name=room_name,
            sip_trunk_id=trunk_id,
            sip_call_to=dial_number,
            participant_identity=participant_identity,
            participant_name=f"Customer_{participant_identity}",
            max_call_duration=timedelta(seconds=180),
            wait_until_answered=True
        )
    )
    <http://ctx.logger.info|ctx.logger.info>("Call answered — participant created: %s", participant_identity)

    # Optionally verify attributes once returned
    res = await ctx.api.room.get_participant(
        api.RoomParticipantIdentity(room=room_name, identity=participant_identity)
    )
    call_status = (res.participant.attributes or {}).get("sip.callStatus")
    <http://ctx.logger.info|ctx.logger.info>("sip.callStatus after return: %s", call_status)  # expect "active"

    return {"status": "answered", "participant": participant}

except api.TwirpError as e:
    sip_code = e.metadata.get("sip_status_code")
    sip_msg = e.metadata.get("sip_status")
    ctx.logger.error("Failed to create SIP participant: %s %s", sip_code, sip_msg)

    if sip_code == "486":
        return {"status": "busy", "code": sip_code}
    if sip_code == "487":
        return {"status": "rejected", "code": sip_code}
    if sip_code == "404":
        return {"status": "not_found", "code": sip_code}

    # other failures
    raise RuntimeError(f"Failed to initiate outbound call: {sip_code} {sip_msg}") from e
Notes and tips • Use the blocking option when you want a simple flow that only continues after a successful answer. It keeps your server flow straightforward but ties up the awaiting task until answer/failure. • Even with 
wait_until_answered=True
, log and inspect 
e.metadata
in the exception handler to classify busy/rejected/etc. • If you need intermediate states (ringing/dialing) or faster UI updates, use the non‑blocking approach with attribute events instead. Want a version of this example integrated into your existing function (with your logger and error handling)?
b
call_status = (res.participant.attributes or {}).get(\"sip.callStatus\")\n ^^^^^^^^^^^^^^^\nAttributeError: participant", "exc_info": "AttributeError: participant
t
Thanks — the error happens because the response shape is different than you expected. Quick fixes and a runnable example below. Why the AttributeError • 
create_sip_participant
returns a 
SIPParticipantInfo
(not an object with 
.participant
), so access 
res.attributes
directly instead of 
res.participant.attributes
. • 
get_participant
returns a response with a 
participant
field; that one has 
.attributes
. Runnable example (blocking with wait_until_answered=True)
Copy code
from datetime import timedelta

async def create_with_wait(ctx, room_name, trunk_id, dial_number, participant_identity):
    try:
        res = await ctx.api.sip.create_sip_participant(
            api.CreateSIPParticipantRequest(
                room_name=room_name,
                sip_trunk_id=trunk_id,
                sip_call_to=dial_number,
                participant_identity=participant_identity,
                participant_name=f"Customer_{participant_identity}",
                max_call_duration=timedelta(seconds=180),
                wait_until_answered=True
            )
        )
        # res is SIPParticipantInfo; attributes are on res.attributes
        call_status = (res.attributes or {}).get("sip.callStatus")
        <http://ctx.logger.info|ctx.logger.info>("create returned, sip.callStatus=%s", call_status)
        return {"status": "answered" if call_status == "active" else "unknown", "res": res}
    except api.TwirpError as e:
        sip_code = e.metadata.get("sip_status_code")
        sip_msg = e.metadata.get("sip_status")
        ctx.logger.error("create_sip_participant failed: %s %s", sip_code, sip_msg)
        if sip_code == "486":
            return {"status": "busy", "code": sip_code}
        if sip_code == "487":
            return {"status": "rejected", "code": sip_code}
        if sip_code == "404":
            return {"status": "not_found", "code": sip_code}
        raise
If you want to double-check after return (sometimes attributes populate slightly later), call `get_participant`:
Copy code
res2 = await ctx.api.room.get_participant(
    api.RoomParticipantIdentity(room=room_name, identity=participant_identity)
)
call_status = (res2.participant.attributes or {}).get("sip.callStatus")
<http://ctx.logger.info|ctx.logger.info>("polled sip.callStatus=%s", call_status)
If you want, I can convert these snippets to fit exactly into your existing function and logging. Which one should I adapt — the blocking flow you posted or a polling loop after the blocking call? Sources: Make outbound calls | LiveKit docs | SIP participant | LiveKit docs
b
what does the field max_call_duration do? Is any event emiited by it?
t
Good question — here’s what 
max_call_duration
does and what you can expect. • Purpose: it sets the maximum allowed length of the SIP call. The timer starts once the call is answered (
sip.callStatus == "active"
). • Behavior: when the duration is reached, LiveKit automatically ends the call. • Events emitted: ◦ The SIP participant will leave the room (you’ll get the normal participant-left/participant-removed event). ◦ A 
sip_call_ended
webhook event is emitted with termination context (reason like 
max_call_duration_reached
). • Notes: handle cleanup on your participant-left handler and/or the webhook to log the reason or perform post-call actions. If you want, I can show a small handler snippet that logs when a participant leaves and inspects the webhook payload for the max-duration reason. Sources: server-sdk-js | livekit/livekit_composite | DeepWiki | SIP APIs | LiveKit docs
b
yes
t
Great — below are two concise examples: 1) Participant-left handler (in-process event) — logs when a SIP participant leaves and checks attributes for a max-duration reason.
Copy code
# In your room/RTC event handlers
def on_participant_left(participant):
    # participant is rtc.Participant (has .identity and .attributes)
    identity = participant.identity
    attrs = participant.attributes or {}
    <http://ctx.logger.info|ctx.logger.info>("Participant left: %s, attrs=%s", identity, attrs)

    # Some providers or LiveKit may include a reason attribute
    reason = attrs.get("sip.call_end_reason") or attrs.get("call_end_reason")
    if reason == "max_call_duration_reached":
        <http://ctx.logger.info|ctx.logger.info>("Call ended due to max_call_duration for %s", identity)
    else:
        <http://ctx.logger.info|ctx.logger.info>("Participant %s left; reason=%s", identity, reason)
2) Webhook handler for 
sip_call_ended
— HTTP endpoint that receives webhook payloads and handles 
max_call_duration_reached
.
Copy code
# Example Flask-like handler
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
def webhook():
    payload = request.get_json()
    event = payload.get("event")
    data = payload.get("data", {})

    if event == "sip_call_ended":
        sip_call_id = data.get("sip_call_id")
        reason = data.get("reason")  # e.g., "max_call_duration_reached"
        room = data.get("room")
        identity = data.get("participant_identity")

        <http://app.logger.info|app.logger.info>("sip_call_ended: call=%s identity=%s reason=%s", sip_call_id, identity, reason)

        if reason == "max_call_duration_reached":
            # perform cleanup, logging, billing, or notify users
            <http://app.logger.info|app.logger.info>("Call %s ended because max_call_duration was reached", sip_call_id)
            # your post-call logic here

    return jsonify({"ok": True})
Notes • Use the in-process participant-left event for immediate cleanup in your app runtime. • Use the 
sip_call_ended
webhook for reliable server-side processing and for post-call analytics; it includes the termination reason like 
max_call_duration_reached
. • Add verification for webhook authenticity (signatures) per your webhook configuration. Would you like these adapted to your framework (async aiohttp/FastAPI) or added logging/details?
b
provide me with the handler implementation
t
Below are two concrete handlers you can drop into your service: (A) an in-process async participant-left handler (for your LiveKit client/room event loop), and (B) a FastAPI webhook endpoint to receive 
sip_call_ended
events. Both detect the 
max_call_duration_reached
termination and show where to perform cleanup/logging. A — Async participant-left handler (in-app)
Copy code
# adapt to your RTC client framework's event decorator/callback style
async def on_participant_left(participant):
    """
    Called when a participant leaves the room.
    `participant` is the runtime participant object with .identity and .attributes.
    """
    identity = getattr(participant, "identity", "<unknown>")
    attrs = getattr(participant, "attributes", {}) or {}

    # Log basic info
    <http://logger.info|logger.info>("Participant left: identity=%s, attrs=%s", identity, attrs)

    # Look for a reason attribute signaling max duration
    reason = attrs.get("sip.call_end_reason") or attrs.get("call_end_reason") or attrs.get("reason")
    if reason == "max_call_duration_reached":
        <http://logger.info|logger.info>("Call for %s ended due to max_call_duration", identity)
        # perform cleanup: update DB, notify users, mark billing, etc.
        await handle_max_duration_cleanup(identity, participant, attrs)
    else:
        <http://logger.info|logger.info>("Participant %s left (reason=%s)", identity, reason)
        await handle_normal_cleanup(identity, participant, attrs)

async def handle_max_duration_cleanup(identity, participant, attrs):
    # Implement your app-specific cleanup here
    # Example: mark call record ended, increment counters, notify downstream systems
    call_id = attrs.get("sip.call_id") or attrs.get("sip_call_id")
    <http://logger.info|logger.info>("Handling max-duration cleanup for call=%s identity=%s", call_id, identity)
    # ... your logic ...
Notes: • Exact attribute keys may vary by integration; check logs for attribute names. Use the attribute keys you see in your environment. • Register this handler with your LiveKit client/room event system (the wiring depends on your library). B — FastAPI webhook handler for 
sip_call_ended
Copy code
from fastapi import FastAPI, Request, HTTPException
import logging

app = FastAPI()
logger = logging.getLogger("sip_webhook")

# Optional: verify webhook signature here if configured

@app.post("/webhook")
async def webhook(request: Request):
    payload = await request.json()
    event = payload.get("event")
    data = payload.get("data", {}) or {}

    <http://logger.info|logger.info>("Received webhook event=%s data=%s", event, data)

    if event == "sip_call_ended":
        sip_call_id = data.get("sip_call_id") or data.get("call_id")
        reason = data.get("reason")  # e.g., "max_call_duration_reached"
        room = data.get("room")
        identity = data.get("participant_identity") or data.get("participant")

        <http://logger.info|logger.info>("sip_call_ended: call=%s identity=%s reason=%s room=%s", sip_call_id, identity, reason, room)

        if reason == "max_call_duration_reached":
            # perform server-side post-call processing
            await server_side_max_duration_tasks(sip_call_id, identity, room, data)
        else:
            # handle other end reasons
            await server_side_regular_end(sip_call_id, identity, room, data)

    # respond quickly to acknowledge
    return {"ok": True}

async def server_side_max_duration_tasks(sip_call_id, identity, room, data):
    <http://logger.info|logger.info>("Processing max duration end for call=%s identity=%s", sip_call_id, identity)
    # e.g., update DB, mark billing, send notifications
    # ... your logic ...
Deployment and verification tips • Add webhook signature verification if you enabled signing on LiveKit webhooks. • Log entire webhook payloads (safely) initially to confirm the exact fields your instance sends. • Use both handlers: the in-app participant-left for immediate UX cleanup and the webhook for reliable server-side processing and analytics. • If you don’t see the 
reason
you expect, confirm 
max_call_duration
was set on the SIP participant and check LiveKit server logs/webhook payloads for the exact field name. Would you like me to adapt either handler to your specific framework (aiohttp, Quart, Django, or the exact LiveKit event wiring you use)?
b
can you give me the implementation with : Register this handler with your LiveKit client/room event system (the wiring depends on your library).
t
Below are concrete examples showing how to register the participant-left / participant-attributes handler with common LiveKit client/server libraries. I include (A) Python async server-side using the livekit-server-sdk-like event wiring, and (B) JavaScript (Node) using the LiveKit Room SDK event emitter. Adapt names to your actual client objects. A — Python (async) — register handlers on your room/client
Copy code
# Example assumes you have an async LiveKit room client object that emits events.
# Adjust to your SDK's actual event API (this matches common Python event callback patterns).

import asyncio
import logging

logger = logging.getLogger("livekit_handlers")

async def handle_max_duration_cleanup(identity, participant, attrs):
    <http://logger.info|logger.info>("Handling max-duration cleanup for %s, attrs=%s", identity, attrs)
    # your cleanup logic here (DB update, notify, billing, etc.)

async def on_participant_left(participant):
    identity = getattr(participant, "identity", "<unknown>")
    attrs = getattr(participant, "attributes", {}) or {}
    <http://logger.info|logger.info>("Participant left: identity=%s attrs=%s", identity, attrs)

    reason = attrs.get("sip.call_end_reason") or attrs.get("call_end_reason") or attrs.get("reason")
    if reason == "max_call_duration_reached":
        <http://logger.info|logger.info>("Call ended due to max_call_duration for %s", identity)
        await handle_max_duration_cleanup(identity, participant, attrs)
    else:
        <http://logger.info|logger.info>("Participant %s left; reason=%s", identity, reason)
        # normal cleanup

# Example registration: if your room exposes `on`/`add_listener` style API:
def register_handlers(room):
    # common pattern: room.on(event_name, callback)
    # replace with your SDK's equivalent registration method
    room.on("participant_left", lambda p: asyncio.create_task(on_participant_left(p)))
    room.on("participant_removed", lambda p: asyncio.create_task(on_participant_left(p)))
    room.on("participant_attributes_changed", lambda changed, p: asyncio.create_task(on_attributes_changed(changed, p)))

# optional attributes-changed handler
async def on_attributes_changed(changed_attributes: dict, participant):
    if "sip.callStatus" in changed_attributes:
        status = changed_attributes["sip.callStatus"]
        identity = getattr(participant, "identity", "<unknown>")
        <http://logger.info|logger.info>("Participant %s sip.callStatus -> %s", identity, status)
        if status == "active":
            <http://logger.info|logger.info>("Call answered for %s", identity)
How to wire: • Call 
register_handlers(your_room)
after you obtain the room client (when your app joins or creates the room). • Many Python SDKs require callbacks to be synchronous; wrapping in 
asyncio.create_task
keeps your async handler running. B — JavaScript / Node (LiveKit Room SDK)
Copy code
// Example assumes `room` is a LiveKit Room instance from livekit-client (node/web)
// and uses EventEmitter style handlers.

const registerHandlers = (room) => {
  // participant removed/left
  room.on('participantDisconnected', (participant) => {
    // participant is a RemoteParticipant/Participant object
    const identity = participant.identity;
    const attrs = participant.metadata ? JSON.parse(participant.metadata) : {}; // if metadata used
    console.log('participantDisconnected', identity, attrs);

    const reason = attrs['sip.call_end_reason'] || attrs['call_end_reason'] || attrs['reason'];
    if (reason === 'max_call_duration_reached') {
      console.log(`Call for ${identity} ended due to max_call_duration`);
      // perform cleanup
    }
  });

  // participant metadata/attributes changed (if SDK exposes this)
  room.on('participantMetadataChanged', (participant) => {
    // some SDKs surface attributes/metadata changes via this event
    const identity = participant.identity;
    const attrs = participant.metadata ? JSON.parse(participant.metadata) : {};
    const status = attrs['sip.callStatus'];
    if (status) {
      console.log(`Participant ${identity} sip.callStatus -> ${status}`);
      if (status === 'active') {
        console.log(`Call answered for ${identity}`);
      }
    }
  });

  // for livekit-client browser SDK, `participantDisconnected` and `trackPublished` events are common.
};

module.exports = { registerHandlers };
Notes and adjustments • Exact event names vary by SDK/version: common names include 
participant_left
, 
participant_removed
, 
participantDisconnected
, 
participantDisconnected
, 
participant_metadata_changed
, or 
participant_attributes_changed
. If an event name differs in your library, use its documented name. • Participant attributes may live in 
participant.attributes
, 
participant.metadata
, or be provided as the 
changed_attributes
map on an attributes-changed event. Inspect the participant object logged at runtime to confirm keys. • Wrap async handlers with 
asyncio.create_task
(Python) or ensure Node handlers call async functions properly (no unhandled promises). • Keep webhook handling (server-side) as the reliable source of termination reason if you need guaranteed processing. If you share which SDK and version you’re using (Python server SDK, livekit-js in Node, or browser), I’ll adapt this snippet exactly to that library’s event API.
b
what details i can get from rtc.LocalParticipant
t
You can access these details from a 
rtc.LocalParticipant
(local user) in LiveKit: • Identity and IDs ◦ 
identity
— developer-provided identity string ◦ 
sid
— server-assigned participant SID • Display / descriptive info ◦ 
name
— display name ◦ 
metadata
— arbitrary string you can get/set ◦ 
attributes
— key/value map you can get/set • Connection & type ◦ 
state
— connection state (JOINING, JOINED, ACTIVE, DISCONNECTED) ◦ 
kind
/ 
kind_detail
— participant kind (STANDARD, SIP, AGENT, etc.) • Tracks & publications ◦ 
trackPublications
/ 
audioTrackPublications
/ 
videoTrackPublications
— published tracks and their 
TrackPublication
objects ◦ Each publication has 
sid
, 
name
, 
kind
, 
source
, and 
muted
status ◦ Methods to publish/unpublish tracks (e.g., 
publishTrack
, 
publishAudioTrack
) • Permissions & subscription control ◦ 
permissions
/ 
ParticipantPermission
— what the participant may do ◦ APIs to set subscription permissions (who can subscribe to your tracks) • Data & messaging ◦ 
publishData(...)
— send data packets to other participants • Timestamps and lifecycle ◦ 
joined_at
/ connection timestamps • Methods to update local state ◦ 
setMetadata(metadata)
— update metadata ◦ 
setAttributes(attributes)
— update attributes ◦ 
setTrackSubscriptionPermissions(...)
— control subscriptions • Events you can observe for the local participant ◦ Track publication changes, metadata/attributes changes, permission updates, and connection state transitions Key distinctions • LocalParticipant represents the current client and exposes publishing/data APIs and setters; RemoteParticipant represents others and is used to subscribe to their tracks and read their metadata. • Use event listeners to react to changes (metadata/attributes/track state/etc.). If you tell me which SDK and language you’re using (e.g., livekit-js in Node/browser, Python server SDK, or mobile SDK), I’ll provide exact property names and a short code example for that environment. Sources: Rooms, participants, and tracks | LiveKit docs | Participant attributes and metadata | LiveKit docs | Client SDKs | livekit/livekit_composite | DeepWiki | client-sdk-js | livekit/livekit_composite | DeepWiki | Data packets | LiveKit docs
b
i am creating the sip participant: participant = await ctx.api.sip.create_sip_participant( api.CreateSIPParticipantRequest( room_name=room_name, sip_trunk_id=trunk_id, sip_call_to=dial_number, participant_identity=participant_identity, participant_name=f"Customer_{call_context.customer_data.customer_name}", max_call_duration=timedelta(seconds=180), # 3 minutes max wait_until_answered=True ) ) and then in on_attributes_changed, i am doing: def on_participant_attributes_changed(changed_attrs, participant): """Track real-time attribute changes (call status, warnings, etc.)""" self.logger.info(f"$$$$$$$$$$ logging participant details in on_participant_attributes_changed:{participant}") identity = getattr(participant, "identity", "<unknown>") self.logger.info(f"%%%%%%%%% Logging identity:{identity} and target_participant_identity:{target_participant_identity} ") # Only log for target participant if target_participant_identity and identity != target_participant_identity: return the participant_identity is equal to target_participant_identity, is something wrong, because the participant identity will not match target_participant_identity
t
Thanks — quick checklist and concrete fixes so your attribute-change callback reliably matches the SIP participant you created. 1) Confirm what identity you passed • Log the value right after creation:
Copy code
<http://self.logger.info|self.logger.info>("created sip participant identity=%s result=%s", participant_identity, participant)
• Ensure 
participant_identity
is exactly the string you expect (unique per call). 2) Ensure you did NOT set hide_phone_number=True • If 
hide_phone_number=True
, LiveKit may generate an anonymized identity. Make sure your request either omits that field or sets it to False. 3) Wait for the participant to join before relying on events • The participant may not be fully present when you attach/compare. Wait for the participant to appear, then register the listener:
Copy code
# after creation
await ctx.wait_for_participant(identity=participant_identity)
register_attribute_listener()
4) Inspect the actual event payload (log everything) • Temporarily log every attribute-change event to see who triggers it:
Copy code
def on_participant_attributes_changed(changed_attrs, participant):
    <http://self.logger.info|self.logger.info>("attr-changed event: identity=%s attrs=%s", getattr(participant,"identity","<nil>"), changed_attrs)
    ...
• This reveals the identity string and keys you should compare. 5) Compare the correct field • Use 
participant.identity
(not participant.sid or participant.participant_identity). Example:
Copy code
if getattr(participant,"identity",None) != target_participant_identity:
    return
# handle changes for the target participant
6) Avoid race conditions • If you create multiple SIP participants, events for other participants will arrive too. Use unique identities (e.g., include call_id) so comparisons are unambiguous. If you want, paste: • the exact 
CreateSIPParticipantRequest
you send (including any optional fields), and • one full log line printed from the attribute-change handler and I’ll point out the exact mismatch and a minimal code change. Sources: Make outbound calls | LiveKit docs | Participant attributes and metadata | LiveKit docs | SIP APIs | LiveKit docs | Agent-assisted warm transfer | LiveKit docs | Rooms, participants, and tracks | LiveKit docs
b
participant = await ctx.api.sip.create_sip_participant( api.CreateSIPParticipantRequest( room_name=room_name, sip_trunk_id=trunk_id, sip_call_to=dial_number, participant_identity=participant_identity, participant_name=f"Customer_{call_context.customer_data.customer_name}", max_call_duration=timedelta(seconds=180), # 3 minutes max wait_until_answered=True ) )
t
Thanks — your 
CreateSIPParticipantRequest
looks correct (no 
hide_phone_number
shown). Next steps to find why the identity in your 
on_participant_attributes_changed
callback doesn't match the 
participant_identity
you passed: 1) Log the create result immediately
Copy code
<http://self.logger.info|self.logger.info>("create_sip_participant returned: %s", participant)
Check the returned object for the identity string (it’s 
participant.participant_identity
/ 
participant.identity
depending on client). Confirm it exactly equals 
participant_identity
. 2) Log every attributes-changed event (no filter) to see the actual identity delivered
Copy code
def on_participant_attributes_changed(changed_attrs, participant):
    <http://self.logger.info|self.logger.info>("attr-changed event: identity=%s attrs=%s", getattr(participant, "identity", None), changed_attrs)
This reveals whether the event is for a different participant or the identity string is different (extra whitespace, prefix, etc.). 3) Wait for participant join before comparing (avoid race) • After create, wait for the participant to appear, then attach listener or start checks:
Copy code
await ctx.wait_for_participant(identity=participant_identity)
# then rely on on_participant_attributes_changed comparisons
4) Compare the correct field and normalize strings • Use 
participant.identity
(or 
participant.participant_identity
if your returned object uses that). • Normalize both sides:
Copy code
if (getattr(participant,"identity","").strip() != participant_identity.strip()):
    return
5) Ensure uniqueness to avoid collisions • Use a unique identity per call (include call_id) to avoid matching a different session. 6) If still mismatched, capture exact values and paste here • Add this temp log right after creation and in the handler, then share the two lines: ◦ create log: "create returned identity=..." ◦ handler log: "attr-changed event identity=..." I’ll inspect and point out the exact mismatch and minimal fix. Which log lines can you paste here (create result and one sample attr-changed event)?
b
what is sid and is it unique?
t
Good question — short answer first: • sid is a server-assigned stable identifier (string) for a LiveKit resource (participant, track, room, etc.). • Yes — a participant’s 
sid
is unique for that participant instance and is safe to use as an unambiguous identifier within LiveKit. More detail: • Purpose: 
sid
is generated by LiveKit and used internally and by APIs to reference a specific resource regardless of developer-provided 
identity
or 
name
. • Uniqueness scope: each 
sid
is globally unique for that resource type within the LiveKit deployment (you won’t get two participants with the same participant 
sid
). • When to use 
sid
vs `identity`: ◦ Use 
sid
when you need an immutable, server-side identifier (logs, DB keys, API calls that target a specific participant instance). ◦ Use 
identity
when you want a human-meaningful or developer-provided label (login names, display mapping). Note that identities can be re-used across sessions and may be collided or replaced. • Lifespan: a participant 
sid
is tied to that connection/session. If the same logical user rejoins, they may get a new 
sid
. • Examples of use: finding the participant in room queries, correlating events/traces, or storing call records keyed to the server 
sid
. If you want, I can show how to log and compare 
sid
in your handlers instead of 
identity
, or provide a short snippet storing 
sid
in your DB for later lookups. Which would you prefer?
b
is the below correct:
Copy code
def on_participant_attributes_changed(changed_attrs, participant):
            """Track real-time attribute changes (call status, warnings, etc.)"""
            <http://self.logger.info|self.logger.info>(f"$$$$$$$$$$ logging participant details in on_participant_attributes_changed:{participant}")
            participant_sid = getattr(participant, "sid", "<unknown>")
            <http://self.logger.info|self.logger.info>(f"%%%%%%%%% Logging participant_sid:{participant_sid} and target_participant_identity:{target_participant_identity} ")
            # get participant details
            

            if target_participant_identity and participant_sid != get_participant_res.sid:
                return
            
            # Log SIP call status changes
            if "sip.callStatus" in changed_attrs:
                status = changed_attrs["sip.callStatus"]
                <http://self.logger.info|self.logger.info>(
                    "[%s] 📞 Call status changed for %s: %s",
                    room_name,
                    target_participant_identity,
                    status
                )
                if status == "active":
                    <http://self.logger.info|self.logger.info>("[%s] ✅ Call answered", room_name)
            
            # Log any timeout warnings or end reasons
            if "sip.call_end_reason" in changed_attrs:
                reason = changed_attrs["sip.call_end_reason"]
                self.logger.warning(
                    "[%s] ⚠️ Call end reason updated: %s",
                    room_name,
                    reason
                )
            
            # Log other relevant attribute changes
            relevant_attrs = {k: v for k, v in changed_attrs.items() 
                            if k.startswith("sip.") and k not in ["sip.callStatus", "sip.call_end_reason"]}
            if relevant_attrs:
                self.logger.debug(
                    "[%s] 🔄 SIP attributes changed for %s: %s",
                    room_name,
                    target_participant_identity,
                    relevant_attrs
                )
t
You're very close — a few fixes to make it correct and robust. Key points: • Compare the participant's 
sid
to a known 
sid
(not to 
get_participant_res.sid
unless that variable is defined and in scope). • If you want to target by identity, compare 
participant.identity
to 
target_participant_identity
. • Normalize and log the actual values to avoid surprises. Corrected handler (minimal changes):
Copy code
def on_participant_attributes_changed(changed_attrs, participant):
    """Track real-time attribute changes (call status, warnings, etc.)"""
    <http://self.logger.info|self.logger.info>("$$$$$$$$$$ logging participant details in on_participant_attributes_changed: %s", participant)

    participant_sid = getattr(participant, "sid", "<unknown>")
    participant_identity = getattr(participant, "identity", None)
    <http://self.logger.info|self.logger.info>("%%%%%%%%% Logging participant_sid:%s identity:%s target_participant_identity:%s",
                     participant_sid, participant_identity, target_participant_identity)

    # If you are matching by SID, ensure you have stored target_participant_sid earlier:
    if target_participant_sid and participant_sid != target_participant_sid:
        return

    # Or, if matching by identity:
    if target_participant_identity and participant_identity != target_participant_identity:
        return

    # Log SIP call status changes
    if "sip.callStatus" in changed_attrs:
        status = changed_attrs["sip.callStatus"]
        <http://self.logger.info|self.logger.info>("[%s] 📞 Call status changed for %s: %s",
                         room_name, participant_identity or participant_sid, status)
        if status == "active":
            <http://self.logger.info|self.logger.info>("[%s] ✅ Call answered", room_name)

    # Log any timeout warnings or end reasons
    if "sip.call_end_reason" in changed_attrs:
        reason = changed_attrs["sip.call_end_reason"]
        self.logger.warning("[%s] ⚠️ Call end reason updated for %s: %s",
                            room_name, participant_identity or participant_sid, reason)

    # Log other relevant attribute changes
    relevant_attrs = {k: v for k, v in changed_attrs.items()
                      if k.startswith("sip.") and k not in ["sip.callStatus", "sip.call_end_reason"]}
    if relevant_attrs:
        self.logger.debug("[%s] 🔄 SIP attributes changed for %s: %s",
                          room_name, participant_identity or participant_sid, relevant_attrs)
If you don't yet have 
target_participant_sid
, capture it when you create the SIP participant:
Copy code
res = await ctx.api.sip.create_sip_participant(...)
target_participant_sid = getattr(res, "participant_id", None) or getattr(res, "sid", None)
Want me to adapt the handler to only use SID (recommended) and show exactly where to store 
target_participant_sid
after creation?
b
but why i am getting the local participant and not the SIP participant?
t
Good question — here are the likely reasons your handler is seeing the local participant and how to fix it. Why this happens (most common causes) • Your event registration is on the local participant object or a room event that emits the local participant first. Some SDKs fire attribute events for the local participant when you set attributes or metadata locally. • You're comparing by identity but the local participant shares the same identity (if you used the same identity for the local client), or you're matching on the wrong field (e.g., comparing identity to a stored SID). • You attached the handler before the SIP participant joined, and early events come from the local participant lifecycle. How to confirm which participant you received • Log these values inside the handler: ◦ participant.kind (should be "SIP" for SIP participants) ◦ participant.sid ◦ participant.identity ◦ participant.attributes Example log line:
Copy code
<http://self.logger.info|self.logger.info>("attr-changed: kind=%s sid=%s identity=%s attrs=%s",
                 getattr(participant, "kind", None),
                 getattr(participant, "sid", None),
                 getattr(participant, "identity", None),
                 changed_attrs)
How to ensure you only handle the SIP participant • Filter by kind:
Copy code
if getattr(participant, "kind", "").upper() != "SIP":
    return
• Or filter by the server-assigned SID you saved when creating the SIP participant:
Copy code
# store target_participant_sid after create_sip_participant (res.participant_id or res.participant_sid)
if target_participant_sid and getattr(participant, "sid", None) != target_participant_sid:
    return
• Or filter by identity but ensure you compare normalized strings and wait for the SIP participant to join first:
Copy code
await ctx.wait_for_participant(identity=participant_identity)
# then attach listener or start checking events
Recommended minimal corrected handler
Copy code
def on_participant_attributes_changed(changed_attrs, participant):
    kind = getattr(participant, "kind", "").upper()
    sid = getattr(participant, "sid", None)
    identity = getattr(participant, "identity", None)
    <http://self.logger.info|self.logger.info>("attr-changed: kind=%s sid=%s identity=%s attrs=%s",
                     kind, sid, identity, changed_attrs)

    # only care about SIP participants (or use saved SID)
    if kind != "SIP":
        return

    # if you have stored the SIP participant sid, match that instead:
    if target_participant_sid and sid != target_participant_sid:
        return

    # handle sip.callStatus etc...
Action items to try now 1. After create, log the creation response and save the returned SID: ◦ res = await ctx.api.sip.create_sip_participant(...) ◦ save target_participant_sid = getattr(res, "participant_id", None) or getattr(res, "participant_sid", None) or getattr(res, "sid", None) 1. Use the handler above to filter by 
kind == "SIP"
or by 
target_participant_sid
. 2. If you still see the local participant, paste the logs from the create response and one attr-changed log line and I’ll point out the mismatch. Want me to produce the exact patch to your handler that saves the SID on creation and uses it to filter?
b
get participant in a room
t
Use the Room service GetParticipant API to fetch a participant by room + identity (server-side call). Example patterns — pick the language you use. Python (async):
Copy code
from livekit.api import RoomParticipantIdentity

res = await ctx.api.room.get_participant(
    RoomParticipantIdentity(room=room_name, identity=participant_identity)
)
participant = res.participant
# access fields:
sid = participant.sid
identity = participant.identity
kind = participant.kind
attrs = participant.attributes or {}
Node (server-side RoomServiceClient):
Copy code
const res = await roomService.getParticipant(roomName, participantIdentity);
const participant = res.participant;
const sid = participant.sid;
const identity = participant.identity;
const attrs = participant.attributes || {};
Go:
Copy code
res, err := roomClient.GetParticipant(ctx, &livekit.RoomParticipantIdentity{
  Room: roomName, Identity: participantIdentity,
})
participant := res.Participant
sid := participant.Sid
Notes: • You must call this from server-side code with appropriate roomAdmin privileges. • Response includes 
sid
, 
identity
, 
state
, 
metadata
, 
attributes
, 
tracks
, 
permission
, 
kind
. • Use 
participant.sid
as the immutable identifier; 
identity
is developer-provided and may be reused. • If you need to poll call state, read 
participant.attributes["sip.callStatus"]
. • If GetParticipant returns no participant, wait for join or ensure identity is correct/unique. Tell me which SDK/language you want a fuller runnable snippet for and I’ll provide it. Sources: Rooms, participants, and tracks | LiveKit docs | Managing participants | LiveKit docs
b
why is the local particiapnt returned?
Copy code
on_participant_attributes_changed:rtc.LocalParticipant(sid=PA_Mvisbiasdasd, identity=agent-AJ_CU7jHdaZm5F, name=)"
i am creating the participant like the following:
Copy code
participant = await ctx.api.sip.create_sip_participant(
                api.CreateSIPParticipantRequest(
                    room_name=room_name,
                    sip_trunk_id=trunk_id,
                    sip_call_to=dial_number,
                    participant_identity=participant_identity,
                    participant_name=f"Customer_{name}",
                    max_call_duration=timedelta(seconds=180), # 3 minutes max
                    wait_until_answered=True
                    )
                )
is any event fired after the call has gone beyond 3 minutes
t
You're seeing the local participant because the attributes-changed event is firing for the local participant as well. Quick answers and what to do next. Why that happens • The SDK emits attribute/metadata events for any participant whose attributes change — including the local participant. • Your local client is a LocalParticipant (kind may be STANDARD/AGENT), while the SIP leg is a separate participant (kind == "SIP" or a remote participant). Your handler receives both unless you filter. What happens when max_call_duration is reached • LiveKit will automatically end the call when the timer expires. • You will see the SIP participant leave (participant-left/participant-removed) and the SIP-specific attributes updated (e.g., 
sip.call_end_reason
or similar). • A 
sip_call_ended
webhook is also emitted with reason like 
max_call_duration_reached
. Immediate fixes (minimal changes) 1. Log identifying fields in your handler to confirm which participant you got:
Copy code
<http://self.logger.info|self.logger.info>("attr-changed: kind=%s sid=%s identity=%s attrs=%s",
                 getattr(participant,"kind",None),
                 getattr(participant,"sid",None),
                 getattr(participant,"identity",None),
                 changed_attrs)
1. Filter for the SIP participant (recommended) or by saved SID:
Copy code
# prefer SID if you saved it after creation
if target_participant_sid and getattr(participant,"sid",None) != target_participant_sid:
    return

# or filter by kind
if getattr(participant,"kind","").upper() != "SIP":
    return
1. Save the SIP participant SID immediately after creation:
Copy code
res = await ctx.api.sip.create_sip_participant(...)
# res is SIPParticipantInfo — extract server id
target_participant_sid = getattr(res, "participant_id", None) or getattr(res, "participant_sid", None) or getattr(res, "participant_id", None)
1. Handle max duration in your handlers: • In the attributes-changed handler, check 
sip.call_end_reason
or 
sip.callStatus == "hangup"
. • Also handle participant-left/participant-removed to run cleanup. • Use the webhook 
sip_call_ended
for reliable server-side processing (includes 
reason
). If you want, I can: • Patch your handler to save the SID at creation and use it to filter events, or • Provide a short runnable example that demonstrates creating the SIP participant, saving the returned SID, and reacting only to that SIP participant’s attribute changes and leave events. Which do you prefer?
b
Provide a short runnable example that demonstrates creating the SIP participant, saving the returned SID, and reacting only to that SIP participant’s attribute changes and leave events.
t
Below is a concise async Python example you can run (adapt imports/API client creation to your environment). It: • Creates a SIP participant with 
wait_until_answered=True
• Saves the returned SIP participant SID • Registers handlers that filter events by that SID and by 
kind == "SIP"
• Reacts to 
sip.callStatus
changes and participant-left events (including max duration termination) Adjust names for your LiveKit client objects and event wiring.
Copy code
import asyncio
from datetime import timedelta

# Replace these with your actual LiveKit API client/context objects
# ctx.api.sip.create_sip_participant(...) and ctx.api.room.get_participant(...) are used below.
# Also replace `room_client` / `room` with your runtime room object that emits events.

POLL_TIMEOUT = 30

async def create_sip_and_listen(ctx, room, room_name, trunk_id, dial_number, participant_identity, customer_name):
    # store target SID in outer scope so handlers can access it
    target_participant_sid = None

    # Handler: attribute changes
    def on_participant_attributes_changed(changed_attrs, participant):
        kind = getattr(participant, "kind", "").upper()
        sid = getattr(participant, "sid", None)
        identity = getattr(participant, "identity", None)

        # Debug log
        <http://ctx.logger.info|ctx.logger.info>("attr-changed event: kind=%s sid=%s identity=%s changed=%s",
                        kind, sid, identity, changed_attrs)

        # Only handle SIP participant with matching SID
        if kind != "SIP":
            return
        if target_participant_sid is None or sid != target_participant_sid:
            return

        # react to sip.callStatus changes
        if "sip.callStatus" in changed_attrs:
            status = changed_attrs["sip.callStatus"]
            <http://ctx.logger.info|ctx.logger.info>("SIP callStatus for %s (%s) -> %s", identity or sid, sid, status)
            if status == "active":
                <http://ctx.logger.info|ctx.logger.info>("Call answered for %s", identity or sid)
            if status == "hangup":
                <http://ctx.logger.info|ctx.logger.info>("Call ended (hangup) for %s", identity or sid)

        # react to explicit end reasons
        if "sip.call_end_reason" in changed_attrs:
            reason = changed_attrs["sip.call_end_reason"]
            ctx.logger.warning("Call end reason for %s: %s", identity or sid, reason)

    # Handler: participant left / removed
    def on_participant_left(participant):
        kind = getattr(participant, "kind", "").upper()
        sid = getattr(participant, "sid", None)
        identity = getattr(participant, "identity", None)
        <http://ctx.logger.info|ctx.logger.info>("participant left event: kind=%s sid=%s identity=%s", kind, sid, identity)

        if kind != "SIP":
            return
        if target_participant_sid is None or sid != target_participant_sid:
            return

        <http://ctx.logger.info|ctx.logger.info>("Target SIP participant left: %s (%s)", identity or sid, sid)
        # perform cleanup here (DB updates, notify, etc.)

    # Register handlers on the room (adjust to your SDK's registration API)
    # Wrap in tasks if the SDK expects sync callbacks
    room.on("participant_attributes_changed", lambda changed, p: asyncio.create_task(async_wrap(on_participant_attributes_changed, changed, p)))
    room.on("participant_left", lambda p: asyncio.create_task(async_wrap(on_participant_left, p)))
    room.on("participant_removed", lambda p: asyncio.create_task(async_wrap(on_participant_left, p)))

    # Helper to allow sync handler code to run as async task
    async def async_wrap(func, *args, **kwargs):
        return func(*args, **kwargs)

    # Create SIP participant (blocking until answered). Capture returned SID from response.
    try:
        res = await ctx.api.sip.create_sip_participant(
            api.CreateSIPParticipantRequest(
                room_name=room_name,
                sip_trunk_id=trunk_id,
                sip_call_to=dial_number,
                participant_identity=participant_identity,
                participant_name=f"Customer_{customer_name}",
                max_call_duration=timedelta(seconds=180),
                wait_until_answered=True,
            )
        )
    except api.TwirpError as e:
        ctx.logger.error("Failed to create SIP participant: %s metadata=%s", e, getattr(e, "metadata", None))
        raise

    # The create response is SIPParticipantInfo — extract the server participant id (SID)
    # Field names may vary by client; check returned object for participant_id/participant_id or sid fields
    target_participant_sid = getattr(res, "participant_id", None) or getattr(res, "participant_sid", None) or getattr(res, "sid", None) or getattr(res, "participant", None) and getattr(res.participant, "sid", None)
    <http://ctx.logger.info|ctx.logger.info>("Created SIP participant: identity=%s returned_res=%s target_sid=%s", participant_identity, res, target_participant_sid)

    # Optionally: poll once to confirm attributes
    # (use RoomService GetParticipant if you have room admin access)
    try:
        get_res = await ctx.api.room.get_participant(api.RoomParticipantIdentity(room=room_name, identity=participant_identity))
        p = get_res.participant
        attrs = p.attributes or {}
        <http://ctx.logger.info|ctx.logger.info>("Polled participant attrs after create: sid=%s identity=%s attrs=%s", getattr(p, "sid", None), getattr(p, "identity", None), attrs)
    except Exception:
        # get_participant may fail if not yet joined; that's OK
        <http://ctx.logger.info|ctx.logger.info>("Could not poll participant immediately; waiting for events")

    # Keep the function alive or return control to your app; handlers will react to changes while running
    return {"participant_identity": participant_identity, "participant_sid": target_participant_sid}
Notes and tips • Ensure your 
room.on(...)
uses the correct event names for your SDK (some SDKs name them slightly differently). Replace string names with SDK enums if available. • If your SDK's event callbacks are synchronous, wrapping them with 
asyncio.create_task
prevents blocking the event loop. • 
res
field names vary by language client — log the full 
res
object to confirm which property holds the SIP participant SID (common names: 
participant_id
, 
participant_sid
, or nested 
res.participant.sid
). • Filtering by 
sid
and 
kind == "SIP"
avoids handling local participant events. If you want, I can adapt this example to your exact SDK (livekit-python, livekit-server-sdk, or livekit-js) and show the precise registration calls for that library. Which SDK are you using?
b
python
t
Below is a concise, runnable Python example (async) using the LiveKit server & room APIs pattern you’ve been using. It: • Creates a SIP participant with 
wait_until_answered=True
• Extracts and saves the returned participant SID • Registers room event handlers that filter by 
kind == "SIP"
and the saved SID • Reacts to 
sip.callStatus
changes and participant leave events (including max duration termination) Adjust imports, client initialization, and event-registration calls to match your environment.
Copy code
import asyncio
from datetime import timedelta
from typing import Optional

# Replace these with your actual imports / client objects
# from livekit import api  # your generated SDK API module
# ctx.api should expose ctx.api.sip.create_sip_participant and ctx.api.room.get_participant
# `room` should be your runtime room client that emits events.

POLL_TIMEOUT = 30.0  # seconds

async def run_sip_call(ctx, room, room_name: str, trunk_id: str, dial_number: str, participant_identity: str, customer_name: str):
    # container for the target SIP participant SID (closure so handlers can read/write)
    target_participant_sid: Optional[str] = None

    # helper wrapper to run sync handler logic as async tasks
    async def async_wrap(func, *args, **kwargs):
        return func(*args, **kwargs)

    # attribute-changed handler
    def on_participant_attributes_changed(changed_attrs: dict, participant):
        kind = getattr(participant, "kind", "").upper()
        sid = getattr(participant, "sid", None)
        identity = getattr(participant, "identity", None)

        <http://ctx.logger.info|ctx.logger.info>("attr-changed event: kind=%s sid=%s identity=%s changed=%s",
                        kind, sid, identity, changed_attrs)

        # only care about SIP participant with matching SID
        if kind != "SIP":
            return
        if target_participant_sid is None or sid != target_participant_sid:
            return

        # react to sip.callStatus changes
        if "sip.callStatus" in changed_attrs:
            status = changed_attrs["sip.callStatus"]
            <http://ctx.logger.info|ctx.logger.info>("SIP callStatus for %s (%s) -> %s", identity or sid, sid, status)
            if status == "active":
                <http://ctx.logger.info|ctx.logger.info>("Call answered for %s", identity or sid)
            if status == "hangup":
                <http://ctx.logger.info|ctx.logger.info>("Call ended (hangup) for %s", identity or sid)

        # explicit end reason
        if "sip.call_end_reason" in changed_attrs:
            reason = changed_attrs["sip.call_end_reason"]
            ctx.logger.warning("Call end reason for %s: %s", identity or sid, reason)

    # participant-left / removed handler
    def on_participant_left(participant):
        kind = getattr(participant, "kind", "").upper()
        sid = getattr(participant, "sid", None)
        identity = getattr(participant, "identity", None)

        <http://ctx.logger.info|ctx.logger.info>("participant left event: kind=%s sid=%s identity=%s", kind, sid, identity)

        if kind != "SIP":
            return
        if target_participant_sid is None or sid != target_participant_sid:
            return

        <http://ctx.logger.info|ctx.logger.info>("Target SIP participant left: %s (%s)", identity or sid, sid)
        # cleanup actions here (DB, notify, etc.)

    # Register handlers on the room.
    # Replace these with your SDK's exact registration calls if names differ.
    room.on("participant_attributes_changed", lambda changed, p: asyncio.create_task(async_wrap(on_participant_attributes_changed, changed, p)))
    room.on("participant_left", lambda p: asyncio.create_task(async_wrap(on_participant_left, p)))
    room.on("participant_removed", lambda p: asyncio.create_task(async_wrap(on_participant_left, p)))

    # Create SIP participant (blocking until answered)
    try:
        res = await ctx.api.sip.create_sip_participant(
            api.CreateSIPParticipantRequest(
                room_name=room_name,
                sip_trunk_id=trunk_id,
                sip_call_to=dial_number,
                participant_identity=participant_identity,
                participant_name=f"Customer_{customer_name}",
                max_call_duration=timedelta(seconds=180),
                wait_until_answered=True,
            )
        )
    except api.TwirpError as e:
        ctx.logger.error("Failed to create SIP participant: %s metadata=%s", e, getattr(e, "metadata", None))
        raise

    # Extract the server-assigned participant SID from the response.
    # Field names vary by client; check the actual object you receive.
    target_participant_sid = (
        getattr(res, "participant_id", None)
        or getattr(res, "participant_sid", None)
        or getattr(res, "sid", None)
        or (getattr(res, "participant", None) and getattr(res.participant, "sid", None))
    )
    <http://ctx.logger.info|ctx.logger.info>("Created SIP participant: identity=%s returned_res=%s target_sid=%s", participant_identity, res, target_participant_sid)

    # Optionally poll once to confirm attributes (may not be immediately available)
    try:
        get_res = await ctx.api.room.get_participant(api.RoomParticipantIdentity(room=room_name, identity=participant_identity))
        p = get_res.participant
        attrs = p.attributes or {}
        <http://ctx.logger.info|ctx.logger.info>("Polled participant attrs after create: sid=%s identity=%s attrs=%s", getattr(p, "sid", None), getattr(p, "identity", None), attrs)
    except Exception:
        <http://ctx.logger.info|ctx.logger.info>("Could not poll participant immediately; waiting for events")

    # Keep the coroutine alive while events are handled by the registered callbacks.
    # In a real app you'd return control; here we sleep/poll until SIP participant leaves or timeout.
    timeout = 300.0
    start = asyncio.get_event_loop().time()
    while True:
        if asyncio.get_event_loop().time() - start > timeout:
            ctx.logger.warning("Timeout waiting for SIP lifecycle events")
            break
        await asyncio.sleep(1)

    return {"participant_identity": participant_identity, "participant_sid": target_participant_sid}
Notes • Replace 
room.on(...)
event names if your Python client uses different names or enums. • Log the full 
res
object once to confirm which property contains the participant SID; adapt extraction accordingly. • Filtering by 
sid
+ 
kind == "SIP"
avoids processing local participant events. • When 
max_call_duration
is reached you will see either: ◦ 
sip.callStatus == "hangup"
or a 
sip.call_end_reason
attribute update in the attribute-changed handler, and ◦ a 
participant_left
/ 
participant_removed
event for the SIP participant, and ◦ a 
sip_call_ended
webhook if you have webhooks enabled. If you want, I’ll adapt this to the exact livekit-python package API you have (show me a sample 
res
printed after create, and I’ll pin the correct 
sid
extraction line).
b
by default local participant is processed?
t
Yes — by default the SDK will emit participant events for all participants, including the local participant. Event handlers receive changes for any participant whose attributes/metadata change unless you filter. What to do • Filter in your handler by participant kind or SID/identity so you only process the SIP leg you care about. ◦ By kind:
Copy code
if getattr(participant, "kind", "").upper() != "SIP":
        return
• By SID (recommended, immutable):
Copy code
if target_participant_sid and getattr(participant, "sid", None) != target_participant_sid:
        return
• Log events initially (no filter) to confirm what the SDK emits:
Copy code
<http://self.logger.info|self.logger.info>("attr-changed: kind=%s sid=%s identity=%s attrs=%s",
                   getattr(participant,"kind",None),
                   getattr(participant,"sid",None),
                   getattr(participant,"identity",None),
                   changed_attrs)
• Register handlers after the SIP participant joins (or wait for it) to avoid processing early local-participant lifecycle events. Would you like a tiny patch to your handler that applies these filters?
b
Copy code
def on_participant_attributes_changed(changed_attrs, participant):
            """Track real-time attribute changes (call status, warnings, etc.)"""
            # Only process RemoteParticipant events (SIP callers), skip LocalParticipant (agent)
            if not isinstance(participant, rtc.RemoteParticipant):
                return
            
            participant_sid = getattr(participant, "sid", "<unknown>")
            
            # Only log for target participant
            if target_participant_identity and participant_sid != get_participant_res.sid:
                return
            
            <http://self.logger.info|self.logger.info>(
                "[%s] 🔄 SIP participant attributes changed: %s",
                room_name,
                {k: v for k, v in changed_attrs.items() if k.startswith("sip.")}
            )
            
            # Log SIP call status changes
            if "sip.callStatus" in changed_attrs:
                status = changed_attrs["sip.callStatus"]
                <http://self.logger.info|self.logger.info>(
                    "[%s] 📞 Call status changed for %s: %s",
                    room_name,
                    target_participant_identity,
                    status
                )
                if status == "active":
                    <http://self.logger.info|self.logger.info>("[%s] ✅ Call answered", room_name)
            
            # Log any timeout warnings or end reasons
            if "sip.call_end_reason" in changed_attrs:
                reason = changed_attrs["sip.call_end_reason"]
                self.logger.warning(
                    "[%s] ⚠️ Call end reason updated: %s",
                    room_name,
                    reason
                )
            
            # Log other relevant attribute changes
            relevant_attrs = {k: v for k, v in changed_attrs.items() 
                            if k.startswith("sip.") and k not in ["sip.callStatus", "sip.call_end_reason"]}
            if relevant_attrs:
                self.logger.debug(
                    "[%s] 🔄 SIP attributes changed for %s: %s",
                    room_name,
                    target_participant_identity,
                    relevant_attrs
                )
when i applied the above i did not see any logs because the particiapnt is local and not remote
t
You're filtering out the SIP participant because your code returns early when the event carries a LocalParticipant. Fix: capture the SIP participant SID when it connects, then filter attribute events by that SID and by kind == "SIP". Minimal runnable pattern you can drop into your Python app: 1) After creating the SIP participant, save the identity returned. 2) On participant_connected, detect the SIP participant, save its 
sid
. 3) In participant_attributes_changed and participant_left handlers, ignore LocalParticipant and only process events where 
kind == "SIP"
and 
sid == target_sid
. Concise example (merge into your async app; adapt room.on registration to your SDK):
Copy code
# assume ctx, room, room_name, participant_identity already defined

target_participant_sid = None

def on_participant_connected(participant):
    nonlocal target_participant_sid
    kind = getattr(participant, "kind", "").upper()
    identity = getattr(participant, "identity", None)
    sid = getattr(participant, "sid", None)
    <http://ctx.logger.info|ctx.logger.info>("connected: kind=%s identity=%s sid=%s", kind, identity, sid)

    if kind == "SIP" and identity == participant_identity:
        target_participant_sid = sid
        <http://ctx.logger.info|ctx.logger.info>("Saved target SIP participant SID=%s", target_participant_sid)

def on_participant_attributes_changed(changed_attrs, participant):
    # ignore local participant
    if isinstance(participant, rtc.LocalParticipant):
        return

    kind = getattr(participant, "kind", "").upper()
    sid = getattr(participant, "sid", None)
    identity = getattr(participant, "identity", None)
    <http://ctx.logger.info|ctx.logger.info>("attr-changed: kind=%s identity=%s sid=%s changed=%s", kind, identity, sid, changed_attrs)

    if kind != "SIP":
        return
    if target_participant_sid is None or sid != target_participant_sid:
        return

    # handle SIP attributes
    if "sip.callStatus" in changed_attrs:
        status = changed_attrs["sip.callStatus"]
        <http://ctx.logger.info|ctx.logger.info>("sip.callStatus -> %s for %s", status, identity or sid)
    if "sip.call_end_reason" in changed_attrs:
        reason = changed_attrs["sip.call_end_reason"]
        ctx.logger.warning("sip.call_end_reason -> %s for %s", reason, identity or sid)

def on_participant_left(participant):
    kind = getattr(participant, "kind", "").upper()
    sid = getattr(participant, "sid", None)
    identity = getattr(participant, "identity", None)
    <http://ctx.logger.info|ctx.logger.info>("left: kind=%s identity=%s sid=%s", kind, identity, sid)

    if kind == "SIP" and sid == target_participant_sid:
        <http://ctx.logger.info|ctx.logger.info>("Target SIP participant left: %s (%s)", identity or sid, sid)
        # cleanup

# register handlers (wrap with asyncio.create_task if handlers must be async)
room.on("participant_connected", lambda p: on_participant_connected(p))
room.on("participant_attributes_changed", lambda changed, p: on_participant_attributes_changed(changed, p))
room.on("participant_left", lambda p: on_participant_left(p))
room.on("participant_removed", lambda p: on_participant_left(p))

# create SIP participant (example)
res = await ctx.api.sip.create_sip_participant(
    api.CreateSIPParticipantRequest(
        room_name=room_name,
        sip_trunk_id=trunk_id,
        sip_call_to=dial_number,
        participant_identity=participant_identity,
        participant_name=f"Customer_{name}",
        max_call_duration=timedelta(seconds=180),
        wait_until_answered=False,  # non-blocking recommended so you can get connected event
    )
)
<http://ctx.logger.info|ctx.logger.info>("create_sip_participant returned: %s", res)
Notes: • Use 
wait_until_answered=False
for non-blocking flow so you can capture the 
participant_connected
event and SID. If you must use 
wait_until_answered=True
, still capture the returned response and then poll or wait for 
participant_connected
to get the SID. • Filtering by 
sid
and 
kind == "SIP"
is the most reliable way to ignore local participant events. • When 
max_call_duration
elapses you will see 
sip.callStatus == "hangup"
, 
sip.call_end_reason
(reason like max duration), and a participant_left/participant_removed event for the SIP participant; webhook 
sip_call_ended
is also emitted if enabled. If you want, I can adapt this exact pattern to your existing function (show me where you currently register handlers and whether you use wait_until_answered=True).
b
{"level": "ERROR", "name": "livekit", "message": "failed to emit event participant_disconnected\nTraceback (most recent call last):\n  File \"/usr/local/lib/python3.11/site-packages/livekit/rtc/event_emitter.py\", line 58, in emit\n    callback(*callback_args)\n  File \"/home/app-user/services/call_orch_service.py\", line 387, in on_participant_disconnected\n    kind = getattr(participant, \"kind\", \"\").upper()\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAttributeError: 'int' object has no attribute 'upper'", "exc_info": "AttributeError: 'int' object has no attribute 'upper'", "pid": 94, "job_id": "AJ_AMkx4zxBUB4M", "room_id": "RM_Nevw9fqnQ7xb", "timestamp": "2025-12-17T12:41:48.786918+00:00"}
t
You're getting that AttributeError because the event callback received an integer instead of a Participant object. Fixes and a safe handler pattern: 1) Ensure your callback signature matches the SDK event (single participant arg). Register like:
Copy code
session.on("participant_disconnected", on_participant_disconnected)
2) Defensively handle unexpected types to avoid crashing the emitter:
Copy code
def on_participant_disconnected(participant):
    # defensive: ensure we have a participant-like object
    if not hasattr(participant, "kind"):
        ctx.logger.warning("participant_disconnected received unexpected payload: %r", participant)
        return

    kind = getattr(participant, "kind", "").upper()
    sid = getattr(participant, "sid", None)
    identity = getattr(participant, "identity", None)
    <http://ctx.logger.info|ctx.logger.info>("Participant disconnected: kind=%s sid=%s identity=%s", kind, sid, identity)

    if kind != "SIP":
        return

    # handle SIP participant disconnect (cleanup, logging, etc.)
    ...
3) Log the raw event payload once to diagnose why an int is emitted:
Copy code
ctx.logger.error("raw participant_disconnected payload: %r", participant)
If you paste that log line here I can help interpret it. 4) Common causes for non-object payloads: • Wrong event name or emitter used elsewhere that emits different data. • A wrapper or adapter that translates arguments differently. • Bug in an older SDK version — consider upgrading the livekit package. If you want, I can provide a full patched version of your disconnect handler (with SID-based filtering and cleanup) using your current handler code. Which do you prefer — the minimal defensive patch above or a full replacement? Sources: Handling events | LiveKit docs | Rooms, participants, and tracks | LiveKit docs
b
t
Thanks — that GitHub issue is relevant. It shows some SDK versions emit a numeric reason or different args for participant-disconnected/participant-removed events, which explains your TypeError. Quick actionable guidance: 1) Defensively handle different callback signatures • Accept either (participant) or (reason_int, participant) or (participant, reason) so your handler won't crash. Example (drop into your code):
Copy code
def on_participant_disconnected(*args):
    # normalize to get participant object if present
    participant = None
    for a in args:
        # prefer objects that look like participants
        if hasattr(a, "sid") or hasattr(a, "identity"):
            participant = a
            break
    if participant is None:
        ctx.logger.warning("participant_disconnected received unexpected args: %r", args)
        return

    kind = getattr(participant, "kind", "")
    sid = getattr(participant, "sid", None)
    identity = getattr(participant, "identity", None)
    <http://ctx.logger.info|ctx.logger.info>("Participant disconnected: kind=%s sid=%s identity=%s", kind, sid, identity)

    if str(kind).upper() != "SIP":
        return

    # handle SIP participant cleanup...
2) Prefer event names that consistently provide Participant objects • Use 
participant_connected
, 
participant_attributes_changed
, and `participant_removed`/`participant_left` handlers where your SDK reliably passes the Participant object. • If your SDK version is inconsistent, wrap all handlers with the defensive pattern above. 3) Log raw args once to diagnose • Add a one-time log to capture the unexpected payload and determine which SDK version/adapter is emitting it:
Copy code
ctx.logger.error("raw participant_disconnected payload: %r", args)
4) Short-term workaround vs long-term • Workaround: use the defensive handler and filter by SID/kind as shown. • Long-term: upgrade the livekit Python package to a version where event signatures are stable, or check the specific SIP repo issue discussion for the recommended fix. If you want, paste one raw log line from that error (the 
args
payload you log) and I’ll map exactly which argument is the participant vs reason and provide a fully patched handler you can drop in. Which log can you share? Sources: Rooms, participants, and tracks | LiveKit docs | SIP participant | LiveKit docs | Participant attributes and metadata | LiveKit docs | SIP APIs | LiveKit docs | Create and configure a Plivo SIP trunk | LiveKit docs
2 Views
Open in Slack
PreviousNext
Powered by