warm_handoff.py

"""
---
title: Warm Handoff Agent
category: telephony
tags: [call-transfer, warm-handoff, sip, agent-to-human, function-tools, deepgram, openai, elevenlabs]
difficulty: intermediate
description: Agent demonstrating warm handoff functionality to transfer calls to human agents
demonstrates:
  - Call transfer to human agents via SIP
  - Warm handoff implementation patterns
  - SIP participant creation and management
  - Function tools for call operations
  - Multi-participant call handling
  - Professional call transfer announcements
---
"""

import asyncio
import os
import uuid
from dotenv import load_dotenv
from livekit.agents import JobContext, JobProcess, AgentServer, cli, Agent, AgentSession, RunContext, function_tool
from livekit import rtc
from livekit import api
from livekit.plugins import deepgram, openai, silero, elevenlabs

load_dotenv()

class WarmHandoffAgent(Agent):
    def __init__(self, job_context=None, vad=None) -> None:
        self.job_context = job_context
        super().__init__(
            instructions="""
                You are a helpful assistant communicating through voice. You're helping me test ... yourself ... since you're the AI agent.
                Don't use any unpronouncable characters.
            """,
            stt=deepgram.STT(),
            llm=openai.LLM(model="gpt-4o"),
            tts=elevenlabs.TTS(
                encoding="pcm_44100",
                model="eleven_multilingual_v2"
            ),
            vad=vad
        )

    @function_tool
    async def transfer_call(self, context: RunContext, phone_number: str):
        """
        Transfer the current call to a human agent at the specified phone number.

        Args:
            context: The call context
            phone_number: The phone number to transfer the call to
        """
        if not self.job_context:
            await self.session.say("I'm sorry, I can't transfer the call at this time.")
            return None, "Failed to transfer call: No job context available"

        room_name = os.environ.get('LIVEKIT_ROOM_NAME', self.job_context.room.name)
        identity = f"transfer_{uuid.uuid4().hex[:8]}"

        livekit_url = os.environ.get('LIVEKIT_URL')
        livekit_api_key = os.environ.get('LIVEKIT_API_KEY')
        livekit_api_secret = os.environ.get('LIVEKIT_API_SECRET')
        sip_trunk_id = os.environ.get('SIP_TRUNK_ID')

        try:
            print(f"Transferring call to {phone_number}")

            if self.job_context and hasattr(self.job_context, 'api'):
                response = await self.job_context.api.sip.create_sip_participant(
                    api.CreateSIPParticipantRequest(
                        sip_trunk_id=sip_trunk_id,
                        sip_call_to=phone_number,
                        room_name=room_name,
                        participant_identity=identity,
                        participant_name="Human Agent",
                        krisp_enabled=True
                    )
                )
            else:
                livekit_api = api.LiveKitAPI(
                    url=livekit_url,
                    api_key=livekit_api_key,
                    api_secret=livekit_api_secret
                )

                response = await livekit_api.sip.create_sip_participant(
                    api.CreateSIPParticipantRequest(
                        sip_trunk_id=sip_trunk_id,
                        sip_call_to=phone_number,
                        room_name=room_name,
                        participant_identity=identity,
                        participant_name="Human Agent",
                        krisp_enabled=True
                    )
                )

                await livekit_api.aclose()

            await self.session.say(f"I'm transferring you to a human agent now. Please hold while we connect you.")

            return None, f"I've transferred you to a human agent at {phone_number}. Please hold while we connect you."

        except Exception as e:
            print(f"Error transferring call: {e}")
            await self.session.say(f"I'm sorry, I couldn't transfer the call at this time.")
            return None, f"Failed to transfer call: {e}"

    async def on_enter(self):
        self.session.generate_reply()

server = AgentServer()

def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

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

    session = AgentSession()
    agent = WarmHandoffAgent(job_context=ctx, vad=ctx.proc.userdata["vad"])

    await session.start(agent=agent, room=ctx.room)
    await ctx.connect()

    def on_participant_connected_handler(participant: rtc.RemoteParticipant):
        asyncio.create_task(async_on_participant_connected(participant))

    async def async_on_participant_connected(participant: rtc.RemoteParticipant):
        await agent.session.say(f"Hi there! Is there anything I can help you with?")

    for participant in ctx.room.remote_participants.values():
        asyncio.create_task(async_on_participant_connected(participant))

    ctx.room.on("participant_connected", on_participant_connected_handler)

if __name__ == "__main__":
    cli.run_app(server)