feat: twilio connector warm transfer

examples/warm-transfer/support_agent.py

@@ -0,0 +1,134 @@
+import logging
+from collections.abc import Callable
+
+from livekit.agents import (
+    Agent,
+    AgentServer,
+    AgentSession,
+    JobContext,
+    cli,
+    room_io,
+)
+from livekit.agents.beta.workflows import WarmTransferResult
+from livekit.agents.llm import ToolError, function_tool
+from livekit.plugins import noise_cancellation
+
+logger = logging.getLogger("warm-transfer")
+
+
+class SupportAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(instructions=INSTRUCTIONS)
+
+    async def on_enter(self):
+        self.session.generate_reply()
+
+    @function_tool
+    async def transfer_to_human(self) -> None:
+        """Called when the user asks to speak to a human agent. This will put the user on
+           hold while the supervisor is connected.
+
+        Ensure that the user has confirmed that they wanted to be transferred. Do not start transfer
+        until the user has confirmed.
+        Examples on when the tool should be called:
+        ----
+        - User: Can I speak to your supervisor?
+        - Assistant: Yes of course.
+        ----
+        - Assistant: I'm unable to help with that, would you like to speak to a human agent?
+        - User: Yes please.
+        ----
+        """
+
+        logger.info("tool called to transfer to human")
+        await self.session.say(
+            "Please hold while I connect you to a human agent.", allow_interruptions=False
+        )
+        try:
+            result = await self._start_transfer()
+        except ToolError as e:
+            logger.error(f"failed to transfer to supervisor with tool error: {e}")
+            raise e
+        except Exception as e:
+            logger.exception("failed to transfer to supervisor")
+            raise ToolError(f"failed to transfer to supervisor with error: {e}") from e
+
+        logger.info(
+            "transfer to supervisor successful",
+            extra={"lk.pii.supervisor_identity": result.human_agent_identity},
+        )
+        await self.session.say(
+            "you are on the line with my supervisor. I'll be hanging up now.",
+            allow_interruptions=False,
+        )
+        self.session.shutdown()
+
+    # implemented per transport (SIP, Twilio connector, ...)
+    async def _start_transfer(self) -> WarmTransferResult:
+        raise NotImplementedError
+
+
+def run(create_agent: Callable[[], Agent]) -> None:
+    server = AgentServer()
+
+    @server.rtc_session(agent_name="sip-inbound")
+    async def entrypoint(ctx: JobContext) -> None:
+        session = AgentSession(
+            llm="openai/gpt-4.1-mini",
+            stt="deepgram/nova-3:en",
+            tts="cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
+        )
+        await session.start(
+            agent=create_agent(),
+            room=ctx.room,
+            room_options=room_io.RoomOptions(
+                audio_input=room_io.AudioInputOptions(
+                    # enable Krisp BVC noise cancellation
+                    noise_cancellation=noise_cancellation.BVCTelephony(),
+                ),
+                delete_room_on_close=False,  # keep the room open for the customer and supervisor
+            ),
+        )
+
+    # this example requires explicit dispatch using named agents
+    # supervisor will be placed in a separate room, and we do not want it to dispatch the default agent
+    cli.run_app(server)
+
+
+INSTRUCTIONS = """
+# Personality
+
+You are friendly and helpful, with a welcoming personality
+You're naturally curious, empathetic, and intuitive, always aiming to deeply understand the user's intent by actively listening.
+
+# Environment
+
+You are engaged in a live, spoken dialogue over the phone.
+There are no other ways of communication with the user (no chat, text, visual, etc)
+
+# Tone
+
+Your responses are warm, measured, and supportive, typically 1-2 sentences to maintain a comfortable pace.
+You speak with gentle, thoughtful pacing, using pauses (marked by "...") when appropriate to let emotional moments breathe.
+You naturally include subtle conversational elements like "Hmm," "I see," and occasional rephrasing to sound authentic.
+You actively acknowledge feelings ("That sounds really difficult...") and check in regularly ("How does that resonate with you?").
+You vary your tone to match the user's emotional state, becoming calmer and more deliberate when they express distress.
+
+# Identity
+
+You are a customer support agent for LiveKit.
+
+# Transferring to a human
+
+In some cases, the user may ask to speak to a human agent. This could happen when you are unable to answer their question.
+When such is requested, you would always confirm with the user before initiating the transfer.
+"""
+
+
+SUMMARY_INSTRUCTIONS = """
+Introduce the conversation from your perspective as the AI assistant who participated in this call:
+
+WHO you're talking to (name, role, company if mentioned)
+WHY they contacted you (goal, problem, request)
+WHY a human agent is requested or needed at this point
+Brief summary in 100-200 characters from a first-person perspective"""

examples/warm-transfer/twilio_connector_warm_transfer.py

@@ -0,0 +1,41 @@
+import os
+
+from dotenv import load_dotenv
+from support_agent import SUMMARY_INSTRUCTIONS, SupportAgent, run
+
+from livekit.agents.beta.workflows import TwilioConnectorWarmTransferTask, WarmTransferResult
+
+load_dotenv()
+
+# LiveKit credentials (read from env by the SDK): LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET
+# Twilio REST credentials + caller ID:
+TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")  # "ACxxxx..."
+TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")  # "xxxx..."
+TWILIO_FROM_NUMBER = os.getenv(
+    "TWILIO_FROM_NUMBER"
+)  # "+15005006000" - your Twilio number, shown to supervisor
+SUPERVISOR_PHONE_NUMBER = os.getenv("LIVEKIT_SUPERVISOR_PHONE_NUMBER")  # "+12003004000"
+# This example places the outbound call with the Twilio REST SDK (pip install twilio):
+# connect_twilio_call returns a connect_url that we hand to Twilio in <Stream url=...>
+# TwiML; Twilio then streams the supervisor's call audio back to the connector.
+
+
+class TwilioSupportAgent(SupportAgent):
+    async def _start_transfer(self) -> WarmTransferResult:
+        assert SUPERVISOR_PHONE_NUMBER is not None
+        assert TWILIO_FROM_NUMBER is not None
+        return await TwilioConnectorWarmTransferTask(
+            SUPERVISOR_PHONE_NUMBER,
+            twilio_from_number=TWILIO_FROM_NUMBER,
+            twilio_account_sid=TWILIO_ACCOUNT_SID,
+            twilio_auth_token=TWILIO_AUTH_TOKEN,
+            chat_ctx=self.chat_ctx,
+            # give up if the supervisor doesn't pick up within 25s:
+            # ringing_timeout=25,
+            # add extra instructions for summarization
+            extra_instructions=SUMMARY_INSTRUCTIONS,
+        )
+
+
+if __name__ == "__main__":
+    run(TwilioSupportAgent)

examples/warm-transfer/warm_transfer.py

@@ -1,21 +1,9 @@
-import logging
 import os
 
 from dotenv import load_dotenv
+from support_agent import SUMMARY_INSTRUCTIONS, SupportAgent, run
 
-from livekit.agents import (
-    Agent,
-    AgentServer,
-    AgentSession,
-    JobContext,
-    cli,
-    room_io,
-)
-from livekit.agents.beta.workflows import WarmTransferTask
-from livekit.agents.llm import ToolError, function_tool
-from livekit.plugins import noise_cancellation
-
-logger = logging.getLogger("warm-transfer")
+from livekit.agents.beta.workflows import WarmTransferResult, WarmTransferTask
 
 load_dotenv()
 
@@ -25,136 +13,24 @@
 SIP_NUMBER = os.getenv("LIVEKIT_SIP_NUMBER")  # "+15005006000" - caller ID shown to supervisor
 
 
-class SupportAgent(Agent):
-    def __init__(self) -> None:
-        super().__init__(instructions=INSTRUCTIONS)
-
-    async def on_enter(self):
-        self.session.generate_reply()
-
-    @function_tool
-    async def transfer_to_human(self) -> None:
-        """Called when the user asks to speak to a human agent. This will put the user on
-           hold while the supervisor is connected.
-
-        Ensure that the user has confirmed that they wanted to be transferred. Do not start transfer
-        until the user has confirmed.
-        Examples on when the tool should be called:
-        ----
-        - User: Can I speak to your supervisor?
-        - Assistant: Yes of course.
-        ----
-        - Assistant: I'm unable to help with that, would you like to speak to a human agent?
-        - User: Yes please.
-        ----
-        """
-
-        logger.info("tool called to transfer to human")
-        await self.session.say(
-            "Please hold while I connect you to a human agent.", allow_interruptions=False
-        )
-        try:
-            assert SIP_TRUNK_ID is not None
-            assert SUPERVISOR_PHONE_NUMBER is not None
-
-            result = await WarmTransferTask(
-                sip_call_to=SUPERVISOR_PHONE_NUMBER,
-                sip_trunk_id=SIP_TRUNK_ID,
-                sip_number=SIP_NUMBER,
-                chat_ctx=self.chat_ctx,
-                # to reach an extension behind an IVR, pass DTMF tones to send once
-                # answered, e.g. dtmf="wwww1234#" (each `w` pauses ~0.5s):
-                # dtmf=SUPERVISOR_EXTENSION,
-                # give up if the supervisor doesn't pick up within 25s:
-                # ringing_timeout=25,
-                # add extra instructions for summarization
-                # you can also customize the entire instructions by overriding the `get_instructions` method
-                extra_instructions=SUMMARY_INSTRUCTIONS,
-            )
-        except ToolError as e:
-            logger.error(f"failed to transfer to supervisor with tool error: {e}")
-            raise e
-        except Exception as e:
-            logger.exception("failed to transfer to supervisor")
-            raise ToolError(f"failed to transfer to supervisor with error: {e}") from e
-
-        logger.info(
-            "transfer to supervisor successful",
-            extra={"supervisor_identity": result.human_agent_identity},
+class SIPSupportAgent(SupportAgent):
+    async def _start_transfer(self) -> WarmTransferResult:
+        assert SIP_TRUNK_ID is not None
+        assert SUPERVISOR_PHONE_NUMBER is not None
+        return await WarmTransferTask(
+            sip_call_to=SUPERVISOR_PHONE_NUMBER,
+            sip_trunk_id=SIP_TRUNK_ID,
+            sip_number=SIP_NUMBER,
+            chat_ctx=self.chat_ctx,
+            # to reach an extension behind an IVR, pass DTMF tones to send once
+            # answered, e.g. dtmf="wwww1234#" (each `w` pauses ~0.5s):
+            # dtmf=SUPERVISOR_EXTENSION,
+            # give up if the supervisor doesn't pick up within 25s:
+            # ringing_timeout=25,
+            # add extra instructions for summarization
+            extra_instructions=SUMMARY_INSTRUCTIONS,
         )
-        await self.session.say(
-            "you are on the line with my supervisor. I'll be hanging up now.",
-            allow_interruptions=False,
-        )
-        self.session.shutdown()
-
-
-server = AgentServer()
-
-
-@server.rtc_session(agent_name="sip-inbound")
-async def entrypoint(ctx: JobContext):
-    session = AgentSession(
-        llm="openai/gpt-4.1-mini",
-        stt="deepgram/nova-3:en",
-        tts="cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
-    )
-
-    support_agent = SupportAgent()
-
-    await session.start(
-        agent=support_agent,
-        room=ctx.room,
-        room_options=room_io.RoomOptions(
-            audio_input=room_io.AudioInputOptions(
-                # enable Krisp BVC noise cancellation
-                noise_cancellation=noise_cancellation.BVCTelephony(),
-            ),
-            delete_room_on_close=False,  # keep the room open for the customer and supervisor
-        ),
-    )
-
-
-INSTRUCTIONS = """
-# Personality
-
-You are friendly and helpful, with a welcoming personality
-You're naturally curious, empathetic, and intuitive, always aiming to deeply understand the user's intent by actively listening.
-
-# Environment
-
-You are engaged in a live, spoken dialogue over the phone.
-There are no other ways of communication with the user (no chat, text, visual, etc)
-
-# Tone
-
-Your responses are warm, measured, and supportive, typically 1-2 sentences to maintain a comfortable pace.
-You speak with gentle, thoughtful pacing, using pauses (marked by "...") when appropriate to let emotional moments breathe.
-You naturally include subtle conversational elements like "Hmm," "I see," and occasional rephrasing to sound authentic.
-You actively acknowledge feelings ("That sounds really difficult...") and check in regularly ("How does that resonate with you?").
-You vary your tone to match the user's emotional state, becoming calmer and more deliberate when they express distress.
-
-# Identity
-
-You are a customer support agent for LiveKit.
-
-# Transferring to a human
-
-In some cases, the user may ask to speak to a human agent. This could happen when you are unable to answer their question.
-When such is requested, you would always confirm with the user before initiating the transfer.
-"""
-
-
-SUMMARY_INSTRUCTIONS = """
-Introduce the conversation from your perspective as the AI assistant who participated in this call:
-
-WHO you're talking to (name, role, company if mentioned)
-WHY they contacted you (goal, problem, request)
-WHY a human agent is requested or needed at this point
-Brief summary in 100-200 characters from a first-person perspective"""
 
 
 if __name__ == "__main__":
-    # this example requires explicit dispatch using named agents
-    # supervisor will be placed in a separate room, and we do not want it to dispatch the default agent
-    cli.run_app(server)
+    run(SIPSupportAgent)

livekit-agents/livekit/agents/beta/workflows/__init__.py

@@ -7,7 +7,7 @@
 from .phone_number import GetPhoneNumberResult, GetPhoneNumberTask
 from .task_group import TaskCompletedEvent, TaskGroup, TaskGroupResult
 from .utils import WorkflowInstructions
-from .warm_transfer import WarmTransferResult, WarmTransferTask
+from .warm_transfer import TwilioConnectorWarmTransferTask, WarmTransferResult, WarmTransferTask
 
 __all__ = [
     "GetEmailTask",
@@ -29,5 +29,6 @@
     "TaskGroup",
     "TaskGroupResult",
     "WarmTransferTask",
+    "TwilioConnectorWarmTransferTask",
     "WarmTransferResult",
 ]