.

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

So how to get final all status, end reason, call start time , end time and duration ?

Thanks — you're asking how to get the final call status, end reason, start time, end time, and duration. Below are the practical ways to collect each piece and a concise approach you can implement. What to capture and where • Call start time: Record the timestamp from the first relevant event — typically

participant_joined

(or

participant_connected

) for the SIP participant. • Call end time: Record the timestamp from

participant_left

,

participant_disconnected

, or

room_finished

(whichever indicates the call ended). • Duration: Calculate as

end_time - start_time

. • Final status & end reason: Read the SIP participant attribute

sip.callStatus

(values include

active

,

hangup

,

failed

). Combine that with the leaving webhook/event to determine the final status and reason (e.g.,

sip.callStatus = hangup

→ participant hangup). • Room-level termination: If room was explicitly deleted,

room_finished

or the

delete_room

action indicates a final end triggered by room deletion. Recommended implementation (webhooks + per-participant state) 1. Enable webhooks for these events:

participant_joined

, `participant_left`/`participant_disconnected`,

participant_attributes_changed

, and

room_finished

. 2. On `participant_joined`: ◦ Save

start_time

and participant id (and mark participant kind = SIP if applicable). 1. On `participant_attributes_changed`: ◦ Watch for

sip.callStatus

updates; update the participant’s current status and store the latest value. 1. On

participant_left

or `participant_disconnected`: ◦ Save

end_time

, store the final

sip.callStatus

if present, compute

duration = end_time - start_time

, and persist a call record: {start_time, end_time, duration, final_status, end_reason}. 1. On `room_finished`: ◦ If you rely on room lifecycles, use this as a safeguard to finalize any in-progress calls and ensure records are closed. 1. Persist records in your DB/logging so you can query final summaries later. Example logic sketch • start_time = timestamp from

participant_joined

• keep updating current_status from

participant_attributes_changed

(

sip.callStatus

) • when

participant_left

or

room_finished

arrives: ◦ end_time = event.timestamp ◦ final_status = last known

sip.callStatus

(or derive from the leaving event) ◦ duration = end_time - start_time ◦ end_reason = map final_status to a reason (e.g.,

hangup

→ user hangup,

failed

→ connection failure,

room_finished

→ room deleted) Questions to help me tailor exact code • Which SDK/platform are you using (browser, iOS, Android, or server)? • Do you prefer webhook-based server-side recording or client-side event reporting? • Do you already have

sip.callStatus

available in your attributes for these calls? Tell me the platform and preferred approach and I’ll provide a small code example you can drop in. Sources: SIP Lifecycle Management Agent | LiveKit docs | Server-side | livekit/livekit_composite | DeepWiki | Agents telephony integration | LiveKit docs

how to extract the direction of call whether inbound or outbound

You're asking how to determine whether a call is inbound or outbound. Two reliable ways: • Inspect

sip.callStatus

on the SIP participant: ◦ If the first status you observe is

ringing

, treat the call as inbound. ◦ If the first status you observe is

dialing

, treat the call as outbound. ◦ Watch

participant_attributes_changed

webhooks/events to catch the initial status. • If you create the SIP participant programmatically with

create_sip_participant

, that call is outbound by definition (since your server initiated it). Which platform do you want a code example for (server webhook handler, browser SDK, Android, or iOS)? Sources: SIP Lifecycle Management Agent | LiveKit docs

Inspect

sip.callStatus

on the SIP participant: ◦ If the first status you observe is

ringing

, treat the call as inbound. ◦ If the first status you observe is

dialing

, treat the call as outbound. is this method reliable for production ?

Yes — using the initial

sip.callStatus

value to determine direction (first observed

ringing

= inbound, first observed

dialing

= outbound) is reliable for production, provided you follow a few best practices: • Use event-driven detection: read the first

sip.callStatus

seen in

participant_attributes_changed

rather than polling the attribute at arbitrary times. • Persist the first status timestamp so you don’t miss the initial state if later events arrive. • Handle late-joining observers: if your agent joins after the initial states, fall back to known context (e.g., if you called

create_sip_participant

, mark that call outbound) or infer from current state plus metadata. • Combine signals when possible: use the API path (outbound calls you create) and webhook events (

participant_joined

,

participant_left

,

room_finished

) alongside

sip.callStatus

to eliminate ambiguity. • Map states to reasons and log them (e.g.,

ringing

→ inbound,

dialing

→ outbound; `hangup`/`failed` → end reasons). If you tell me your platform (server/webhook, browser, iOS, or Android) I’ll provide a concise code snippet to capture the first

sip.callStatus

and persist the direction reliably. Sources: SIP Lifecycle Management Agent | LiveKit docs | Agents telephony integration | LiveKit docs | Accepting calls overview | LiveKit docs | examples/frontdesk | livekit/livekit_composite | DeepWiki

can yuo implement this logic and give me whole code

I can do that — which platform and language do you want the full implementation for? • Server webhook (Node.js/Express or Python/Flask) • Server-side LiveKit RTC handler (Python) • Browser (JavaScript) • Android (Kotlin) • iOS (Swift) Also tell me: • Do you already use webhooks or prefer in-room event handlers? • Where should the call records be stored (simple file/console, database like Postgres, or another store)? Pick one platform and storage option and I’ll provide a complete, ready-to-run example.

• Server-side LiveKit RTC handler (Python) python

Great — I'll provide a complete, ready-to-run Python server-side LiveKit RTC handler that implements the production-safe logic we've discussed: detects call direction (inbound/outbound) using the first observed

sip.callStatus

(with fallbacks), captures start/end timestamps, computes duration, and persists/logs a final call record. Before I send the code, one quick clarification so I tailor it exactly to your environment: • Do you want persistence to a simple local SQLite DB or to an in-memory/log-only solution (you can switch to Postgres/Redis later)? Tell me which persistence option you prefer and I’ll return the full Python file and run instructions. Sources: SIP Lifecycle Management Agent | LiveKit docs | sip | livekit/livekit_composite | DeepWiki | Agents telephony integration | LiveKit docs | Outbound Calling Script | LiveKit docs

simple print it

Below is a complete Python LiveKit server-side RTC handler that prints call direction (inbound/outbound), start time, end time, duration, and final status to the console. It's the simplified, print-only version you asked for. Save as, e.g.,

sip_call_tracker.py

.

import asyncio
import logging
import time
from datetime import datetime

from livekit.agents import AgentServer, JobContext, AgentSession, function_tool
from livekit import rtc
from livekit.agents.inference import STT, LLM, TTS

logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("sip-call-tracker")


class CallTracker:
    def __init__(self, room_name: str):
        self.room_name = room_name
        self.direction = None  # "inbound" or "outbound"
        self.start_time = None
        self.end_time = None
        self.duration = None
        self.last_status = None
        self.phone_number = None
        self.sip_call_id = None
        self.trunk_id = None
        self._first_status_recorded = False

    def detect_direction_from_identity(self, identity: str):
        # Simple identity heuristic: identities that start with "sip_" are inbound
        if identity and identity.startswith("sip_"):
            self.direction = "inbound"
        else:
            self.direction = "outbound"
        <http://logger.info|logger.info>(f"Detected direction (identity heuristic): {self.direction}")

    def record_first_status(self, status: str):
        if not self._first_status_recorded and status:
            # Use first observed sip.callStatus to determine direction if not already set
            if status == "ringing":
                self.direction = "inbound"
            elif status == "dialing":
                self.direction = "outbound"
            self._first_status_recorded = True
            <http://logger.info|logger.info>(f"First sip.callStatus observed: {status} -> direction: {self.direction}")

    def update_status(self, status: str):
        self.last_status = status
        <http://logger.info|logger.info>(f"Call status update: {status}")
        # Start time at first active
        if status == "active" and not self.start_time:
            self.start_time = time.time()
            <http://logger.info|logger.info>(f"Call started at {datetime.fromtimestamp(self.start_time).isoformat()}")

    def end_call(self):
        self.end_time = time.time()
        if self.start_time:
            self.duration = self.end_time - self.start_time
        <http://logger.info|logger.info>("Call ended.")
        self.print_summary()

    def print_summary(self):
        summary = {
            "room_name": self.room_name,
            "direction": self.direction,
            "phone_number": self.phone_number,
            "trunk_id": self.trunk_id,
            "sip_call_id": self.sip_call_id,
            "started_at": datetime.fromtimestamp(self.start_time).isoformat() if self.start_time else None,
            "ended_at": datetime.fromtimestamp(self.end_time).isoformat() if self.end_time else None,
            "duration_seconds": round(self.duration, 2) if self.duration else None,
            "final_status": self.last_status,
        }
        <http://logger.info|logger.info>("===== Final call summary =====")
        for k, v in summary.items():
            <http://logger.info|logger.info>(f"{k}: {v}")
        <http://logger.info|logger.info>("==============================")


# In-memory map of trackers per room (single-process). Use DB/Redis for multi-worker.
trackers: dict[str, CallTracker] = {}


class SIPCallAgent:
    def __init__(self, ctx: JobContext):
        self.ctx = ctx
        self.room_name = ctx.room.name
        identity = ctx.participant.identity if ctx.participant else None
        tracker = CallTracker(self.room_name)
        tracker.detect_direction_from_identity(identity)
        trackers[self.room_name] = tracker
        self.tracker = tracker

    async def on_participant_connected(self, participant: rtc.RemoteParticipant):
        if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
            return
        attrs = participant.attributes or {}
        self.tracker.phone_number = attrs.get("sip.phoneNumber") or self.tracker.phone_number
        self.tracker.sip_call_id = attrs.get("sip.callID") or self.tracker.sip_call_id
        self.tracker.trunk_id = attrs.get("sip.trunkID") or self.tracker.trunk_id

        # If sip.callStatus present on connect, record it (may be first status)
        call_status = attrs.get("sip.callStatus")
        if call_status:
            self.tracker.record_first_status(call_status)
            self.tracker.update_status(call_status)

        <http://logger.info|logger.info>(f"SIP participant connected (room={self.room_name}, phone={self.tracker.phone_number})")

    async def on_attributes_changed(self, changed: dict[str, str], participant: rtc.Participant):
        if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
            return
        if "sip.callStatus" in changed:
            status = changed["sip.callStatus"]
            self.tracker.record_first_status(status)
            self.tracker.update_status(status)
        if "sip.phoneNumber" in changed and not self.tracker.phone_number:
            self.tracker.phone_number = changed["sip.phoneNumber"]

    async def on_participant_disconnected(self, participant: rtc.RemoteParticipant):
        if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
            return
        # Capture last status if provided on disconnect
        attrs = participant.attributes or {}
        final_status = attrs.get("sip.callStatus")
        if final_status:
            self.tracker.update_status(final_status)
        self.tracker.end_call()


@function_tool()
async def get_call_info(ctx: any = None):
    t = trackers.get(ctx.room.name)
    if not t:
        return {"error": "no call tracked"}
    return {
        "direction": t.direction,
        "duration_seconds": round(t.duration, 2) if t.duration else None,
        "status": t.last_status,
    }


@AgentServer.rtc_session()
async def entrypoint(ctx: JobContext):
    ctx.log_context_fields = {"room": ctx.room.name}

    session = AgentSession(
        stt=STT(model="deepgram/nova-3", language="en"),
        llm=LLM(model="openai/gpt-4.1-mini"),
        tts=TTS(model="elevenlabs/eleven_multilingual_v2"),
        vad=None,
        tools=[get_call_info],
    )

    agent = SIPCallAgent(ctx)

    await session.start(agent=agent, room=ctx.room)
    await ctx.connect(auto_subscribe=rtc.AutoSubscribe.AUDIO_ONLY)

    ctx.room.on("participant_connected", lambda p: asyncio.create_task(agent.on_participant_connected(p)))
    ctx.room.on("participant_attributes_changed", lambda changed, p: asyncio.create_task(agent.on_attributes_changed(changed, p)))
    ctx.room.on("participant_disconnected", lambda p: asyncio.create_task(agent.on_participant_disconnected(p)))

    <http://logger.info|logger.info>(f"Agent attached to room {ctx.room.name}")


server = AgentServer()

if __name__ == "__main__":
    import sys
    from livekit.agents.cli import run_app

    run_app(server)

How to run 1. Install dependencies: ◦ pip install "livekit-agents[deepgram,openai,elevenlabs]" python-dotenv 1. Provide your LiveKit credentials via environment (.env) or environment variables as required by your LiveKit agent runtime. 2. Start agent: ◦ python sip_call_tracker.py console ◦ Or use `dev`/other run modes as your environment requires. If you want, I can also: • Add a small SQLite persistence snippet, • Replace the identity heuristic with a metadata-based detection, • Or give a tiny unit test to simulate attribute events. Which would you prefer next?