https://livekit.io logo
Join Slack
Powered by
Hi how we can transfer call handoff first to Staff...
# ask-ai
r
Hi how we can transfer call handoff first to Staff member A if he didn't pickup then try to transfer to staff member B
t
Thanks for reaching out, Umair. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Hey there! If you want to set up call handoff to multiple staff members one after the other, here’s a simple way to do it: 1. Set up DTMF Handling: You can assign DTMF digits to each staff member. For example, pressing "1" connects to Staff Member A, and pressing "2" connects to Staff Member B. 2. Sequential Handoff Logic: Try transferring the call to Staff Member A first. If that doesn’t work, don’t worry—automatically try Staff Member B next. 3. Use SIP REFER: The LiveKit API lets you transfer calls using SIP REFER, so you can manage these transfers smoothly through your code. 4. Error Handling: Make sure to include some error handling to either retry or move on to the next staff member if the first transfer doesn’t go through. If you want, I’m happy to walk you through more detailed steps or even share some code examples—just let me know!
r
So if llm agent decided to transfer then it will happen How we transfer and check if transfer was successful? If failed try to do second one
t
Hey there! If you want to transfer a call and make sure it went through smoothly, here’s a simple way to do it: 1. Start the Transfer: Use the 
TransferSIPParticipant
API call to send the call over to the first staff member. 2. Check How It Went: The API will give you a status or an error. If the transfer doesn’t work, no worries—you can catch that and move on. 3. Try Again or Move to the Next Person: If the transfer fails, you can either try again or transfer to the next staff member on your list. It’s a good idea to add a little pause between attempts. Here’s a quick example to show you how this might look in code:
Copy code
async def transfer_call_with_retry(participant_identity, transfer_to_list):
    for transfer_to in transfer_to_list:
        try:
            transfer_request = proto_sip.TransferSIPParticipantRequest(
                participant_identity=participant_identity,
                room_name=self.context.room.name,
                transfer_to=transfer_to,
                play_dialtone=True
            )
            await self.livekit_api.sip.transfer_sip_participant(transfer_request)
            return  # Exit if transfer is successful
        except Exception as e:
            await asyncio.sleep(1)  # Wait before trying the next transfer
If anything’s unclear or you want me to walk you through more examples, just let me know—I’m here to help! Sources: SIP troubleshooting guide | LiveKit Docs | livekit/livekit_sip.pb.go | sipclient.go | pkg/sip/outbound.go
r
provide code with correct imports and
t
Here's the complete Python code for transferring calls with retry functionality in LiveKit. I've made sure it's clear and easy to follow, but feel free to ask if anything needs more explanation!
Copy code
import os
import asyncio
import logging
from dotenv import load_dotenv
from livekit import api, rtc
from livekit.agents import (
    AutoSubscribe,
    JobContext,
    WorkerOptions,
    cli,
    function_tool,
)
from livekit.protocol import sip as proto_sip

# Load environment variables
load_dotenv(dotenv_path=".env.local")

# Initialize logging
logger = logging.getLogger("call-transfer-agent")
logger.setLevel(<http://logging.INFO|logging.INFO>)

class CallTransferAgent:
    def __init__(self, context: JobContext):
        self.context = context
        self.livekit_api = None
        self.max_retries = 3
        self.retry_delay = 1  # seconds

    async def initialize_api(self):
        if self.livekit_api is None:
            livekit_url = os.getenv('LIVEKIT_URL')
            api_key = os.getenv('LIVEKIT_API_KEY')
            api_secret = os.getenv('LIVEKIT_API_SECRET')

            if not all([livekit_url, api_key, api_secret]):
                raise ValueError("LiveKit credentials not provided")

            self.livekit_api = api.LiveKitAPI(
                url=livekit_url,
                api_key=api_key,
                api_secret=api_secret
            )

    async def transfer_call_with_retry(self, participant_identity: str, transfer_to: str) -> None:
        """Transfer call with retry logic"""
        await self.initialize_api()
        
        for attempt in range(self.max_retries):
            try:
                request = proto_sip.TransferSIPParticipantRequest(
                    participant_identity=participant_identity,
                    room_name=self.context.room.name,
                    transfer_to=f"tel:{transfer_to}",
                    play_dialtone=True
                )
                
                await self.livekit_api.sip.transfer_sip_participant(request)
                <http://logger.info|logger.info>(f"Call transferred successfully to {transfer_to}")
                return

            except Exception as e:
                logger.error(f"Call transfer failed (attempt {attempt + 1}/{self.max_retries}): {e}")
                
                if attempt < self.max_retries - 1:
                    logger.info(f"Retrying in {self.retry_delay} seconds...")
                    await asyncio.sleep(self.retry_delay)
                else:
                    logger.error("Maximum retry attempts reached")

    async def on_dtmf_received(self, event: rtc.SipDTMF):
        """Handle DTMF input to trigger transfer"""
        digit = event.digit
        if not digit:
            return

        # Mapping digits to phone numbers
        department_mapping = {
            "1": "+12345678901",  # Billing
            "2": "+12345678902",  # Technical Support
            "3": "+12345678903",   # Customer Service
        }

        transfer_number = department_mapping.get(digit)
        if transfer_number:
            await self.transfer_call_with_retry(event.participant.identity, transfer_number)
        else:
            logger.info(f"Invalid digit '{digit}' received")

async def entrypoint(context: JobContext) -> None:
    agent = CallTransferAgent(context)
    
    # Connect to room and wait for participants
    await context.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    
    # Set up DTMF handler
    @context.room.on("sip_dtmf_received")
    def handle_dtmf(event: rtc.SipDTMF):
        asyncio.create_task(agent.on_dtmf_received(event))
    
    # Keep the agent running
    await asyncio.Event().wait()

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
### Prerequisites: • Make sure your 
.env.local
file is set up with your LiveKit credentials like this:
Copy code
LIVEKIT_URL=your-livekit-url
LIVEKIT_API_KEY=your-api-key
LIVEKIT_API_SECRET=your-api-secret
• You'll also want to install the required packages if you haven't already: 
pip install livekit-python livekit-agents python-dotenv
If you have any questions or want me to walk you through any part of this, just let me know — I'm here to help!
Open in Slack
PreviousNext
Powered by