When i create a call using the sip.create_sip_part...
# ask-ai
w
When i create a call using the sip.create_sip_participant. How can i destinguish call statuses like e.g. "Busy", "Invalid Number", "No Connection"?
t
Thanks for reaching out, Lukas. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — you're asking how to distinguish outcomes like "Busy", "Invalid Number", or "No Connection" when creating a call with
sip.create_sip_participant
. Here's how you can detect those statuses. • Monitor the participant attributes for SIP state changes (listen for the
participant_attributes_changed
event). • Check
sip.callStatus
for lifecycle states (
dialing
,
ringing
,
active
,
hangup
). • When
sip.callStatus
becomes
hangup
, read
sip.hangupCode
and
sip.hangupReason
on the participant to determine why the call ended. Common mappings you can use: • Busy: SIP code
486
(or
600
) • Invalid number: SIP code
404
(or
484
) • No connection / timeout: SIP code
408
(or
504
) Example logic (pseudo/Python):
Copy code
if changed_attributes.get("sip.callStatus") == "hangup":
    code = participant.attributes.get("sip.hangupCode", "")
    if code == "486":
        result = "Busy"
    elif code in ("404", "484"):
        result = "Invalid Number"
    elif code in ("408", "504"):
        result = "No Connection"
    else:
        result = f"Hangup ({code})"
If you’d like, tell me: • which SDK/language you’re using and • an example of the hangup attributes you’re seeing —I can provide a ready-to-use event handler for your environment. Sources: SIP Lifecycle Management Agent | LiveKit docs
w
I am using the Python SDK. Please show me a full example
t
Thanks — below is a complete, runnable Python example using the LiveKit Python SDK that: • creates an outbound SIP participant with
create_sip_participant
• listens for
participant_attributes_changed
• checks
sip.callStatus
and, on
hangup
, inspects
sip.hangupCode
/
sip.hangupReason
• maps common SIP codes/reasons to user-friendly statuses (Busy, Invalid Number, No Connection) Save this as e.g.
sip_call_example.py
, populate your .env, and run.
Copy code
import asyncio
import logging
import os
from dotenv import load_dotenv

from livekit import api
from livekit.agents import AgentContext, JobContext, Worker
from livekit import rtc

load_dotenv()
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("sip_example")

# Helpers: map SIP codes to friendly statuses
def interpret_sip_hangup(code: str, reason: str) -> str:
    if not code and not reason:
        return "Unknown"
    if code in ("486", "600") or reason in ("USER_BUSY", "BUSY"):
        return "Busy"
    if code in ("404", "484") or reason in ("INVALID_NUMBER", "INCOMPATIBLE_DESTINATION", "NUMBER_MOVED"):
        return "Invalid Number"
    if code in ("408", "504") or reason in ("PROGRESS_TIMEOUT", "NO_ANSWER", "NO_USER_RESPONSE"):
        return "No Connection"
    return f"Hangup ({code or reason})"

# Event handler: participant attribute changes
async def on_participant_attributes_changed(ctx: JobContext, participant: rtc.RemoteParticipant, changed_attributes: dict):
    # Only handle SIP participants
    if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
        return

    # Log attribute changes
    for k, v in changed_attributes.items():
        <http://logger.info|logger.info>(f"Participant {participant.identity} attribute changed: {k} = {v}")

    # Inspect call status changes
    sip_status = changed_attributes.get("sip.callStatus")
    if sip_status:
        <http://logger.info|logger.info>(f"sip.callStatus => {sip_status}")
        if sip_status == "dialing":
            <http://logger.info|logger.info>("Call is dialing")
        elif sip_status == "ringing":
            <http://logger.info|logger.info>("Remote is ringing")
        elif sip_status == "active":
            <http://logger.info|logger.info>("Call is active")
        elif sip_status == "automation":
            <http://logger.info|logger.info>("Answered by automation/IVR")
        elif sip_status == "hangup":
            # When hangup occurs, look up hangup code/reason from participant attributes
            hangup_code = participant.attributes.get("sip.hangupCode", "")
            hangup_reason = participant.attributes.get("sip.hangupReason", "")
            friendly = interpret_sip_hangup(hangup_code, hangup_reason)
            <http://logger.info|logger.info>(f"Call ended — code: {hangup_code}, reason: {hangup_reason}, interpreted: {friendly}")

            # Optionally publish data or take action in the room
            try:
                await ctx.room.local_participant.publish_data(f"Call result: {friendly}", subject="sip/result")
            except Exception:
                # publish_data may fail if not connected; ignore in example
                pass

# Initiate outbound SIP call via API
async def create_outbound_sip(room_name: str, sip_trunk_id: str, phone_number: str, participant_identity: str = "outbound_sip"):
    lk_api = api.LiveKitAPI(
        url=os.getenv("LIVEKIT_URL"),
        api_key=os.getenv("LIVEKIT_API_KEY"),
        api_secret=os.getenv("LIVEKIT_API_SECRET"),
    )

    req = api.CreateSIPParticipantRequest(
        sip_trunk_id=sip_trunk_id,
        sip_call_to=phone_number,
        room_name=room_name,
        participant_identity=participant_identity,
        play_dialtone=True,
        hide_phone_number=False,
    )

    try:
        <http://logger.info|logger.info>(f"Creating SIP participant to dial {phone_number} in room {room_name}")
        await lk_api.sip.create_sip_participant(request=req)
        <http://logger.info|logger.info>("SIP participant creation request sent")
    except Exception as e:
        logger.error(f"Failed to create SIP participant: {e}")
    finally:
        await lk_api.aclose()

# Worker entrypoint for LiveKit agent-style flow (keeps a room connection to receive events)
@Worker.register_acio_job
async def entrypoint(ctx: JobContext):
    ctx.log_info("Agent started and joined room", extra={"room": ctx.room.name})

    # Register event listener for participant attribute changes
    ctx.room.on("participant_attributes_changed", lambda args: asyncio.create_task(
        on_participant_attributes_changed(ctx, *args)
    ))

    # Kick off placing the outbound SIP call (use env vars or replace with literals)
    asyncio.create_task(
        create_outbound_sip(
            room_name=ctx.room.name,
            sip_trunk_id=os.getenv("SIP_TRUNK_ID", ""),
            phone_number=os.getenv("TARGET_PHONE_NUMBER", ""),
            participant_identity="outbound_sip_call",
        )
    )

    # Connect agent's RTC connection so room events are received
    await ctx.connect()

# If running standalone (not via Worker CLI), provide a simple runner that simulates agent behavior.
if __name__ == "__main__":
    # .env should provide LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, SIP_TRUNK_ID, TARGET_PHONE_NUMBER
    # Run via LiveKit agents runner if you have it set up:
    # python sip_call_example.py dev
    from livekit.agents.cli import run_app
    run_app()
.env example:
Copy code
LIVEKIT_URL=<https://your-livekit.example>
LIVEKIT_API_KEY=key
LIVEKIT_API_SECRET=secret
SIP_TRUNK_ID=ST_...
TARGET_PHONE_NUMBER=+1234567890
Notes: • The code listens for
participant_attributes_changed
and inspects
sip.callStatus
,
sip.hangupCode
, and
sip.hangupReason
. • Common SIP mappings used: ◦ Busy: 486 or 600 / reason USER_BUSY or BUSY ◦ Invalid number: 404 or 484 / reason INVALID_NUMBER, INCOMPATIBLE_DESTINATION ◦ No connection/timeout: 408 or 504 / reason PROGRESS_TIMEOUT, NO_ANSWER • If you want a version that runs outside the agents framework (pure script + websocket/SDK room connection), tell me and I’ll provide that alternative. Sources: SIP Lifecycle Management Agent | LiveKit docs | Outbound Calling Script | LiveKit docs