i linked twilio with LK , so how can i get the cal...
# ask-ai
b
i linked twilio with LK , so how can i get the callSid
t
Thanks for reaching out, Aman. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
If you’re looking to grab the
callSid
while using Twilio with LiveKit, here’s a simple way to do it: 1. First, check if the participant is a SIP participant by seeing if
participant.kind == ParticipantKind.SIP
. 2. Then, you can get the
callSid
from
participant.attributes['sip.twilio.callSid']
. Here’s a quick example to make it clearer:
Copy code
if (participant.kind === ParticipantKind.SIP) {
    const callSid = participant.attributes['sip.twilio.callSid'];
    // Use the callSid as you need
    console.log(callSid);
}
This approach works smoothly for both inbound and outbound calls when Twilio is set up with LiveKit.
b
import ParticipantKind from where
t
Hey there! To import
ParticipantKind
in LiveKit, here’s a quick guide depending on the programming language you’re using: • JavaScript/TypeScript:
import { ParticipantKind } from '@livekit/client-sdk-js';
Python:
from livekit.rtc import participant_pb2 as participant
(then you can access it as
participant.ParticipantKind
) • Swift:
import LiveKit
Dart/Flutter:
import 'package:livekit_client/livekit_client.dart';
Rust:
use livekit::proto::ParticipantKind;
If you have any questions or need a hand with anything else, just let me know—I’m here to help! Sources: src/index.ts | livekit-protocol/livekit/protocol/models.pyi | Sources/LiveKit/Participant/Participant+Kind.swift | lib/src/ui/builder/participant/participant_kind.dart | livekit/src/room/participant/mod.rs
b
import asyncio import os from dotenv import load_dotenv import json import re from livekit.rtc.participant_pb2 import ParticipantKind from livekit import agents from livekit.agents import AgentSession, Agent, RoomInputOptions, cli from livekit.plugins import ( noise_cancellation, google ) from livekit.agents import ( Agent, AgentSession, JobContext, WorkerOptions, cli, function_tool ) from prompt_services import prompts from tool_services import tools from resources.constants import get_constant load_dotenv() import time from db_implementations.mongo_implementation import MongoDBClient from prompt_services.prompt_services import get_bot_system_prompt # Global variables to store context and session for call termination _ctx = None _session = None _call_sid = None # Store the extracted Call SID client = MongoDBClient("teliphoney_bot_v1_db") def extract_call_sid_from_participant(participant, room_name): """ Extract Call SID from various sources in order of preference: 1. Participant metadata (JSON format) 2. Participant metadata (raw string) 3. Participant identity 4. Room name pattern 5. SIP headers (if available) """ call_sid = None print(f"🔍 Attempting to extract Call SID for participant: {participant.identity}") # Method 1: Try parsing metadata as JSON if participant.metadata: print(f"📋 Raw metadata: '{participant.metadata}'") try: metadata = json.loads(participant.metadata) print(f"📋 Parsed metadata: {metadata}") # Check various possible key names for Call SID possible_keys = ['callSid', 'call_sid', 'CallSid', 'CALL_SID', 'twilio_call_sid', 'sid'] for key in possible_keys: if key in metadata: call_sid = metadata[key] print(f"🔥 CALL SID found in metadata['{key}']: {call_sid}") return call_sid except json.JSONDecodeError: print("⚠️ Metadata is not valid JSON, trying as raw string") # Method 2: Search for Call SID pattern in raw metadata string # Twilio Call SIDs typically start with 'CA' followed by 32 hex characters call_sid_pattern = r'CA[a-f0-9]{32}' match = re.search(call_sid_pattern, participant.metadata, re.IGNORECASE) if match: call_sid = match.group(0) print(f"🔥 CALL SID found in raw metadata: {call_sid}") return call_sid # Method 3: Check participant identity for Call SID if participant.identity: print(f"🆔 Checking participant identity: {participant.identity}") call_sid_pattern = r'CA[a-f0-9]{32}' match = re.search(call_sid_pattern, participant.identity, re.IGNORECASE) if match: call_sid = match.group(0) print(f"🔥 CALL SID found in participant identity: {call_sid}") return call_sid # Method 4: Extract from room name pattern # Assuming room name might contain Call SID print(f"🏠 Checking room name: {room_name}") call_sid_pattern = r'CA[a-f0-9]{32}' match = re.search(call_sid_pattern, room_name, re.IGNORECASE) if match: call_sid = match.group(0) print(f"🔥 CALL SID found in room name: {call_sid}") return call_sid # Method 5: Check for any other patterns that might be Call SID # Sometimes the last part of room name could be Call SID room_parts = room_name.split('_') for part in room_parts: if len(part) >= 30: # Call SIDs are typically 34 characters print(f"🔍 Potential Call SID candidate: {part}") # You can add more validation here if part.startswith('CA') or len(part) == 34: call_sid = part print(f"🔥 POTENTIAL CALL SID from room parts: {call_sid}") return call_sid print(" No Call SID found in any source") return None async def entrypoint(ctx: agents.JobContext): global _ctx, _session, _call_sid _ctx = ctx # Store context globally for end_call function system_prompt = """ you are a helpful assistant your name is Summer instructions:- before ending the call say it was nice talking to you and then use the end_call function when the conversation is complete or the caller wants to hang up, always call the end_call function """ print("room_name", ctx.room) print("building prompt") parts = str(ctx.room.name).split("_") service_id = parts[1] if len(parts) > 1 else "unknown" caller_number = parts[2] if len(parts) > 2 else "unknown" await ctx.connect() # Enhanced participant event listener def on_participant_connected(participant): global _call_sid print(f"👤 Participant connected: {participant.identity}") print(f"📞 Participant kind: {participant.kind}") print(f"📋 Participant metadata: '{participant.metadata}'") # Check if it's a SIP participant if participant.kind == 3: # SIP participant type print("🔌 SIP Participant detected!") # Extract Call SID using our comprehensive method _call_sid = extract_call_sid_from_participant(participant, ctx.room.name) if _call_sid: print(f" SUCCESSFULLY EXTRACTED CALL SID: {_call_sid}") # Store Call SID for later use (e.g., in tools for ending call) # You can save this to database or pass to your tools else: print(" Could not extract Call SID from any source") print("🔧 Debug info:") print(f" - Room name: {ctx.room.name}") print(f" - Participant identity: {participant.identity}") # print(f" - ParticipantKind.SIP: {participant.attributes['sip.twilio.callSid']}") print(f" - Participant metadata: '{participant.metadata}'") print(f" - Participant kind: {participant.kind}") else: print(f"👤 Non-SIP participant (kind: {participant.kind})") # Method to get current Call SID def get_current_call_sid(): return _call_sid # Register the event listener ctx.room.on("participant_connected", on_participant_connected) session_ctx = tools.SessionContext(ctx) # You can also pass the Call SID getter to your tools session_ctx.get_call_sid = get_current_call_sid class AssistantWithNotes(Agent): def __init__(self, prompt="you are a helpful assistant") -> None: super().__init__( instructions=prompt, tools=[ # session_ctx.end_call ] ) # Create session and store globally _session = AgentSession( llm=google.beta.realtime.RealtimeModel( model="gemini-2.0-flash-exp", voice="Aoede", temperature=0.8, instructions="start with a greeting", ), ) print("session created") await _session.start( room=ctx.room, agent=AssistantWithNotes(prompt=system_prompt), room_input_options=RoomInputOptions( noise_cancellation=noise_cancellation.BVC(), ), ) print("Room created:", ctx.room) await _session.generate_reply( instructions="start with a greeting and say that you have reached <org name>", ) import logging if name == "__main__": # Configure logging for better debugging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) # Run the agent with the name that matches your dispatch rule cli.run_app(WorkerOptions( entrypoint_fnc=entrypoint, agent_name="telephony_agent", num_idle_processes=3 )) i just want to print the partecipant callSid so i can use it to terminate call from twilio itself
t
To print the participant's
callSid
so you can use it to end a call from Twilio, here’s a simple way to do it: 1. First, make sure the participant is a SIP participant by checking
participant.kind == ParticipantKind.SIP
. 2. Then, grab the
callSid
from the participant’s attributes with
participant.attributes['sip.twilio.callSid']
. 3. Finally, print out the
callSid
to double-check you’ve got it. Here’s a quick example you can try:
Copy code
if participant.kind == ParticipantKind.SIP:
    call_sid = participant.attributes.get('sip.twilio.callSid')
    if call_sid:
        print(f"Call SID: {call_sid}")
    else:
        print("Call SID not found in participant attributes.")
If you want me to walk you through anything else or have questions, just let me know—I’m here to help!
b
Traceback (most recent call last): File "/home/amansharma/projects/voice_agent/app/teliphoney_agent_v2.py", line 6, in <module> from livekit.rtc.participant_pb2 import ParticipantKind ModuleNotFoundError: No module named 'livekit.rtc.participant_pb2'
t
It looks like the
participant_pb2
module is missing, which is causing that error. No worries though—here’s a friendly step-by-step to help you fix it: 1. *Reinstall the LiveKit RTC package:*`pip uninstall livekit-rtc pip install livekit-rtc` 2. Check if protobuf files are generated: If you’re working from the source code, just run:`cd livekit-rtc ./generate_proto.sh` 3. Verify the package structure: Make sure the
participant_pb2.py
file is actually in
livekit/rtc/_proto/
. You can quickly check with this snippet:`import os import livekit.rtc package_dir = os.path.dirname(livekit.rtc.file) proto_dir = os.path.join(package_dir, '_proto') participant_pb2_path = os.path.join(proto_dir, 'participant_pb2.py') print(f"Proto directory exists: {os.path.exists(proto_dir)}") print(f"participant_pb2.py exists: {os.path.exists(participant_pb2_path)}")` 4. *Install protobuf dependencies:*`pip install protobuf>=4.21.0` 5. Try using a virtual environment: It’s a clean way to avoid conflicts—just create one and reinstall like this:`python -m venv livekit_env source livekit_env/bin/activate # On Windows: livekit_env\Scripts\activate pip install livekit-rtc` If you hit any bumps or want me to walk you through any of these steps, just give me a shout—I’m here to help! Sources: livekit-rtc/livekit/rtc/_proto/room_pb2.pyi
b
(.venv) amansharma@test-MS-7D70:~/projects/voice_agent$ pip install livekit-rtc ERROR: Could not find a version that satisfies the requirement livekit-rtc (from versions: none) ERROR: No matching distribution found for livekit-rtc
t
Hey! Just a quick heads-up—the right package to install is
livekit
, not
livekit-rtc
. Could you give this command a try?
Copy code
pip install livekit
If you run into any more trouble, feel free to reach out—I’m here to help! Sources: livekit-protocol/livekit/protocol/models.py
looks like the assistant got a little confused.
ParticipantKind
can be found here:
Copy code
from livekit.rtc import ParticipantKind
then you can access it directly as noted in the earlier example
👍 1