https://livekit.io logo
Join Slack
Powered by
I want to write a transfer_to_human function tool ...
# ask-ai
c
I want to write a transfer_to_human function tool with error handling (human is unable to join) should be conveyed to the user.
t
Thanks for reaching out, Subham. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
You're asking for a transfer_to_human function that dials a human agent and informs the user if the human is unable to join — I'll provide a concise, ready-to-use pattern you can adapt and a few short notes on where to customize. Python (async) example — warm transfer with error handling and user notification:
Copy code
from livekit.agents import function_tool, get_job_context

@function_tool()
async def transfer_to_human(self, context) -> str:
    job_ctx = get_job_context()
    if not job_ctx:
        return "Cannot transfer: agent context is unavailable."

    # Customize these
    SIP_TRUNK_ID = "your_sip_trunk_id"
    HUMAN_AGENT_NUMBER = "tel:+15105550123"
    ROOM_NAME = job_ctx.room.name

    try:
        # Let the user know transfer is starting
        await self.session.generate_reply(
            instructions="Please hold while I connect you to a human agent."
        )

        # Attempt to dial the human and wait until answered
        await job_ctx.api.sip.create_sip_participant(
            # CreateSIPParticipantRequest proto expected here
            sip_trunk_id=SIP_TRUNK_ID,
            sip_call_to=HUMAN_AGENT_NUMBER,
            room_name=ROOM_NAME,
            wait_until_answered=True,
        )

        return "Connected to a human agent."

    except Exception as e:
        # Inform the user about the failure and provide next steps
        await self.session.generate_reply(
            instructions="Sorry — I couldn't connect you to a human agent right now. Please try again later or leave a message and we'll follow up."
        )
        # Return a short diagnostic string for logs/response context
        return f"Failed to connect to human agent: {str(e)}"
Notes and customization points • Use 
wait_until_answered=True
so the API waits for the human to answer; if unanswered or fails, the call will raise an exception which you catch. • Replace 
SIP_TRUNK_ID
and 
HUMAN_AGENT_NUMBER
with your trunk and agent contact. • Adjust user-facing messages in 
generate_reply()
for tone/next steps (e.g., scheduling a callback, leaving a voicemail). • If you want retry logic, add a loop with a limited number of attempts and escalate messaging after retries fail. • For Node.js the pattern is similar: call 
sipClient.createSIPParticipant(...)
and handle promise rejection with try/catch or .catch to send the user-friendly message. Would you like a version with retry attempts or a callback scheduling flow included? Sources: Agent-assisted warm transfer | LiveKit docs | Agents telephony integration | LiveKit docs
c
Copy code
@function_tool
    async def transfer_call(self, context: RunContext, reason: str | None = None):
        """
        Transfer the conversation to a human agent when appropriate.
        Call this function when:
        - Direct requests: "I want to speak to a human", "Connect me with a representative", "Transfer me to an agent"
        - Technical limitations: The user needs something beyond your capabilities
        - Complex issues: Problems requiring human judgment or access to systems beyond your reach

        Before calling this function, try to assist the user first if the request is within your capabilities.

        CRITICAL: Do NOT call this function if:
        - A transfer attempt has already been made in this conversation (check conversation history)
        - A transfer is currently in progress
        - You can answer basic information requests yourself
        - You can complete simple tasks within your capabilities
        - The user hasn't explicitly requested human assistance for complex issues

        If a transfer was already attempted and failed, continue helping the user with their request instead of trying to transfer again.

        EXAMPLE FLOWS:

        Example 1 (explicit request):
        User: "I want to talk to a real person instead of an AI. Please connect me to a human agent."
        Assistant: "I understand you'd prefer to speak with a human agent. I'm arranging that connection for you now, and a human representative will assist you shortly."
        [transfer_to_number function called ONCE]

        Example 2 (technical limitation):
        User: "I need to change my billing address and payment method for my order #12345."
        Assistant: "I understand you need to update your billing information. Since this requires accessing your account details, I'll connect you with a human representative who can help you with that change safely."
        [transfer_to_number function called ONCE]

        Example 3 (DO NOT call):
        User: "What's your refund policy?"
        Assistant: "Our standard refund policy allows returns within 30 days of purchase with a receipt. Would you like more specific information about a particular product or situation?"

        Example 4 (DO NOT call again after failed transfer):
        User: "Transfer me to a human"
        Assistant: [calls transfer_call, transfer fails]
        Assistant: "I'm sorry, I couldn't transfer the call right now. How else can I help you?"
        User: "Try again" or "Transfer me"
        Assistant: "I understand you'd like to speak with someone, but I wasn't able to complete the transfer. Let me help you with your request instead. What can I assist you with?"
        [DO NOT call transfer_call again]
        """

        # Prevent duplicate transfer attempts
        if self._transfer_in_progress:
            <http://logger.info|logger.info>("Transfer already in progress, ignoring duplicate call")
            await context.session.say("The transfer is in progress. Please hold on.")
            return

        # Prevent calling if transfer was already attempted recently (within 30 seconds)
        # This prevents rapid repeated calls while allowing retry after some time
        if self.human_transfer_vars.get("attempted"):
            attempted_at = self.human_transfer_vars.get("attempted_at")
            if attempted_at:
                time_since_attempt = (datetime.now(timezone.utc) - attempted_at).total_seconds()
                if time_since_attempt < 30:
                    <http://logger.info|logger.info>(f"Transfer attempted {time_since_attempt:.1f}s ago, ignoring duplicate call")
                    # await context.session.say("I've already attempted to transfer your call recently. Let me help you with your request instead. What can I assist you with?")
                    return
            else:
                # If attempted but no timestamp, still prevent to be safe
                <http://logger.info|logger.info>("Transfer already attempted previously, ignoring duplicate call")
                # await context.session.say("I've already attempted to transfer your call. Let me help you with your request instead. What can I assist you with?")
                return

        transfer_to = self.dial_info["transfer_to"]
        if not transfer_to:
            await context.session.say("I'm sorry, I cannot transfer the call right now. I can help schedule a callback for you.")
            # self._closing_task = asyncio.create_task(self.session.aclose())
            raise Exception("human agent number not provided")

        <http://logger.info|logger.info>(f"transferring call to {transfer_to}")

        # Set flag to prevent duplicate calls
        self._transfer_in_progress = True

        # let the message play fully before transferring
        await context.session.say("I'll be transferring you to my team. Please hold on while I connect you.")
        job_ctx = get_job_context()
        try:
            self.human_transfer_vars["attempted"] = True
            self.human_transfer_vars["attempted_at"] = datetime.now(timezone.utc)
            await job_ctx.api.sip.create_sip_participant(
                api.CreateSIPParticipantRequest(
                        room_name=job_ctx.room.name,
                        sip_trunk_id=self.dial_info["sip_trunk_id"],
                        sip_call_to=transfer_to,
                        participant_identity=transfer_to,
                        # function blocks until user answers the call, or if the call fails
                        wait_until_answered=True,
                )
            )

            participant = await job_ctx.wait_for_participant(identity=transfer_to)
            <http://logger.info|logger.info>(f"participant joined: {participant.identity}")
            self.human_transfer_vars["handover_at"] = datetime.now(timezone.utc)
            # await job_ctx.api.room.remove_participant(api.RoomParticipantIdentity(
            #     room=job_ctx.room.name,
            #     identity=job_ctx.room.local_participant.identity
            # ))
            <http://logger.info|logger.info>(f"Agent disconnected")
            self.human_transfer_vars["agent_disconnected_at"] = datetime.now(timezone.utc)
            self.human_transfer_vars["attempt_status"] = "success"
            <http://logger.info|logger.info>(f"transferred call to {transfer_to}")
            # Transfer successful, session will close, but reset flag for completeness
            self._transfer_in_progress = False
            self._closing_task = asyncio.create_task(self.session.aclose())
        except Exception as e:
            logger.error(f"error transferring call to {transfer_to}: {e}")
            self.human_transfer_vars["error_message"] = str(e)
            self.human_transfer_vars["attempt_status"] = "failed"
            # Reset flag to allow retry if needed
            self._transfer_in_progress = False
            await context.session.say("I'm sorry, I couldn't transfer the call right now. Can I arrange a callback for you?")
            # Don't close the session - continue the conversation
why does this close the session in case of failure to transfer the call?
t
You're right to check this — looking at your code, a failed transfer should NOT close the session. The session is closed only on the successful transfer path where you set 
_closing_task = asyncio.create_task(self.session.aclose())
. In the except block you explicitly reset 
_transfer_in_progress
, set attempt status to "failed", and call 
await context.session.say(...)
without calling 
session.aclose()
, so the agent remains in the room and the conversation continues. If you are seeing the session close on failure, please tell me: • Did you observe the session closing immediately after the exception, or after the follow-up message? • Any logs around the error (the logger.error line output or the value of 
self.human_transfer_vars["error_message"]
)? • Is there any other code that watches 
attempt_status
or 
agent_disconnected_at
and triggers 
session.aclose()
elsewhere? With those details I can point to the exact cause or suggest a minimal change to ensure the session stays open on failure. Sources: Agents telephony integration | LiveKit docs | Job lifecycle | LiveKit docs
c
Copy code
2025-12-15 12:48:40,455 - INFO agent - [room: playground-5l3O-krDv] -- Participant disconnected: +916364276271 (PA_u2jz7QSVGYD3) {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:40,455 - INFO agent - [room: playground-5l3O-krDv] -- remote participant disconnected from the call {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:40,455 - INFO agent - [room: playground-5l3O-krDv] -- SIP transfer target disconnected, transfer failed, continuing conversation with original caller {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:40,455 - INFO agent - [room: playground-5l3O-krDv] -- agent is present, continuing conversation with original caller {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:40,456 - ERROR agent - error transferring call to +916364276271: TwirpError(code=unavailable, message=twirp error unknown: unexpected status from INVITE response: sip status: 503: Service Unavailable, status=503, metadata={'error_details': 'CA4STHVuZXhwZWN0ZWQgc3RhdHVzIGZyb20gSU5WSVRFIHJlc3BvbnNlOiBzaXAgc3RhdHVzOiA1MDM6IFNlcnZpY2UgVW5hdmFpbGFibGUaQQoldHlwZS5nb29nbGVhcGlzLmNvbS9saXZla2l0LlNJUFN0YXR1cxIYCPcDEhNTZXJ2aWNlIFVuYXZhaWxhYmxl', 'sip_status': 'Service Unavailable', 'sip_status_code': '503'}) {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:42,170 - INFO livekit.agents - TTS metrics {"room": "playground-5l3O-krDv", "model_name": "sonic-3", "model_provider": "Cartesia", "ttfb": 0.24830645900510717, "audio_duration": 4.83, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:43,898 - INFO livekit.agents - LLM metrics {"room": "playground-5l3O-krDv", "model_name": "openai/gpt-oss-120b", "model_provider": "<http://api.groq.com|api.groq.com>", "ttft": 0.2, "prompt_tokens": 3959, "prompt_cached_tokens": 3840, "completion_tokens": 31, "tokens_per_second": 131.91, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:45,101 - INFO livekit.agents - TTS metrics {"room": "playground-5l3O-krDv", "model_name": "sonic-3", "model_provider": "Cartesia", "ttfb": 0.22831529199902434, "audio_duration": 3.2, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:45,440 - INFO livekit.agents - STT metrics {"room": "playground-5l3O-krDv", "model_name": "nova-3", "model_provider": "Deepgram", "audio_duration": 5.05, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:45,538 - DEBUG livekit.agents - tools execution completed {"room": "playground-5l3O-krDv", "speech_id": "speech_ecce35618500", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:45,538 - INFO agent - [room: playground-5l3O-krDv] --function tools executed: ['transfer_call'] {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:50,489 - INFO livekit.agents - STT metrics {"room": "playground-5l3O-krDv", "model_name": "nova-3", "model_provider": "Deepgram", "audio_duration": 5.05, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:51,494 - DEBUG livekit.agents - received user transcript {"room": "playground-5l3O-krDv", "user_transcript": "Hello.", "language": "en-IN", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:51,496 - INFO livekit.agents - EOU metrics {"room": "playground-5l3O-krDv", "end_of_utterance_delay": 0.64, "transcription_delay": 0.64, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:51,832 - INFO livekit.agents - LLM metrics {"room": "playground-5l3O-krDv", "model_name": "openai/gpt-oss-120b", "model_provider": "<http://api.groq.com|api.groq.com>", "ttft": 0.31, "prompt_tokens": 4029, "prompt_cached_tokens": 3840, "completion_tokens": 68, "tokens_per_second": 203.3, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:52,640 - INFO livekit.agents - TTS metrics {"room": "playground-5l3O-krDv", "model_name": "sonic-3", "model_provider": "Cartesia", "ttfb": 0.21637066599942045, "audio_duration": 2.74, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:55,491 - INFO livekit.agents - STT metrics {"room": "playground-5l3O-krDv", "model_name": "nova-3", "model_provider": "Deepgram", "audio_duration": 5.0, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:57,995 - DEBUG livekit.agents - received user transcript {"room": "playground-5l3O-krDv", "user_transcript": "Yes.", "language": "en-IN", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:57,997 - INFO livekit.agents - EOU metrics {"room": "playground-5l3O-krDv", "end_of_utterance_delay": 0.74, "transcription_delay": 0.74, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:48:58,527 - INFO livekit.agents - LLM metrics {"room": "playground-5l3O-krDv", "model_name": "openai/gpt-oss-120b", "model_provider": "<http://api.groq.com|api.groq.com>", "ttft": 0.42, "prompt_tokens": 4052, "prompt_cached_tokens": 0, "completion_tokens": 69, "tokens_per_second": 130.82, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:00,492 - INFO livekit.agents - STT metrics {"room": "playground-5l3O-krDv", "model_name": "nova-3", "model_provider": "Deepgram", "audio_duration": 5.0, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,399 - DEBUG livekit.agents - received user transcript {"room": "playground-5l3O-krDv", "user_transcript": "We were just having the conversation.", "language": "en-IN", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,400 - WARNING livekit.agents - skipping user input, speech scheduling is paused {"room": "playground-5l3O-krDv", "user_input": "We were just having the conversation.", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,417 - DEBUG livekit.agents - input stream detached {"room": "playground-5l3O-krDv", "participant": "identity-WOie", "source": "SOURCE_MICROPHONE", "accepted_sources": ["SOURCE_MICROPHONE"], "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,421 - INFO agent - [room: playground-5l3O-krDv] -- Session closed, reason: CloseReason.USER_INITIATED, error: None {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,421 - INFO agent - [room: playground-5l3O-krDv] -- Session closed, but not end of call, continuing conversation with original caller {"room": "playground-5l3O-krDv", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,422 - DEBUG livekit.agents - stream closed {"room": "playground-5l3O-krDv", "participant": "identity-WOie", "source": "SOURCE_MICROPHONE", "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
2025-12-15 12:49:01,423 - DEBUG livekit.agents - session closed {"room": "playground-5l3O-krDv", "reason": "user_initiated", "error": null, "pid": 18990, "job_id": "AJ_Lzt6sCiFf4jn"}
t
Thanks — I looked through your logs and code. Summary and next steps: Why the session closed • Your transfer code does not call 
session.aclose()
on transfer failure — the except block only resets flags and sends a failure message. • The logs show the session closed with reason: 
CloseReason.USER_INITIATED
at 124901.421. That means the caller (original participant) disconnected their client, which caused the agent session to close. • The SIP transfer itself failed earlier (503 Service Unavailable). The log lines show the transfer target disconnected and transfer failed, then the agent continued the conversation. That matches your code path for failures. What the sequence shows 1. create_sip_participant failed with SIP 503 — transfer marked failed. 2. Agent stays in room and speaks (TTS/LLM entries appear). 3. The original caller later disconnected (input stream detached, then session closed with user_initiated). 4. Agent session closed because the user hung up, not because of your error-handling path. What you can do if you want to prevent session closure when the caller hangs up • If your goal is to keep the agent session alive after the caller drops (e.g., to log, retry, or schedule a callback), detect 
CloseReason.USER_INITIATED
elsewhere and avoid tearing down background tasks you need, or re-open/hold the session state for callback scheduling. • If you want the agent to proactively offer a retry before the user can hang up, shorten cooldowns or immediately prompt with follow-up options after the failure message (e.g., “Would you like a callback in 10 minutes?”) so the user is less likely to drop. • To capture why the caller disconnected, log additional participant disconnect metadata and correlate timestamps with the transfer failure to decide automated follow-up actions. If you’d like, I can: • Suggest a minimal code change to preserve state or schedule a callback when 
attempt_status == "failed"
. • Draft a retry flow (limited attempts + exponential backoff) or a callback scheduling snippet. Which of those would you prefer?
c
The original caller didn't disconnect, the new sip participant failed, but the original sip participant stayed. Can I configure the new addition to not close session upon failure?
t
You're right to want the original caller to stay connected when the SIP transfer fails. Yes — you can ensure the new SIP participant's failure does not close the session. From your logs and code, the framework closes sessions only when you explicitly call 
session.aclose()
or when the original participant actually disconnects; the SIP transfer failure itself does not need to force the original caller off the call. Concrete checks and changes to prevent session closure on SIP transfer failure: 1. Ensure you only call 
session.aclose()
on the successful handover path • You already do this in the success branch. Do not call 
session.aclose()
anywhere in the exception path or in any participant-disconnect handler that triggers for the SIP target. 1. Handle participant-disconnect events without tearing down the agent session • If you have a global participant-disconnect handler, make it conditional: ◦ If the disconnected identity == transfer target -> mark attempt failed, keep agent session open. ◦ If the disconnected identity == original caller -> treat as user hangup and close session. • Use identity or participant metadata to distinguish original caller vs. SIP target. 1. Avoid implicit session shutdowns from "input stream detached" • Your logs show "input stream detached" and later "session closed" with reason 
user_initiated
. Ensure you aren't interpreting the SIP target disconnect event as the original caller's input stream detaching. • Confirm the participant identity used for the original caller remains attached to microphone/source events. 1. Guard background cleanup tasks • If you run cleanup tasks on participant disconnect (e.g., cancelling jobs, closing session), guard them to run only when the original caller disconnects or when handover succeeded. 1. Optional: explicit reconnection/hold behavior • After transfer failure, explicitly re-attach or resume the original caller's input stream if your agent paused speech scheduling or detached the input. In your logs you saw "skipping user input, speech scheduling is paused" — resume scheduling after a failed transfer so the agent continues to process the original caller audio. 1. Add conservative retry / callback logic • Instead of immediately tearing down on any disconnect, record 
attempt_status = "failed"
and offer retry or schedule callback — do not close session automatically. If you want, I can produce a minimal code patch showing: • a participant-disconnect handler that distinguishes identities and avoids closing the session for transfer-target disconnects, • and a small change to resume speech scheduling after a failure. Which snippet would you like first: the participant-disconnect handler or the flow to resume the original caller's audio/speech scheduling?
c
Copy code
async def entrypoint(ctx: JobContext):
    """Entrypoint for the agent"""
    ctx.log_context_fields = {
        "room": ctx.room.name,
    }
    <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- connecting to room")
    await ctx.connect()
    room_id = await ctx.room.sid
    <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- connected to room, room_id: {room_id}")
    dial_info = {
        "room_name": ctx.room.name,
        "transfer_to": "+916364276271",
        "sip_trunk_id": 'ST_sic2q5UZsU64',
    }
    _is_agent_present = True

    room=ctx.room

    session = AgentSession(
        vad=ctx.proc.userdata["vad"],
        false_interruption_timeout=1,  # Wait 1 second before resuming
        resume_false_interruption=True,   # Enable auto-resume
        min_interruption_duration=0.5,
        min_interruption_words=2,
    )

    usage_collector = metrics.UsageCollector()

    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        metrics.log_metrics(ev.metrics)
        usage_collector.collect(ev.metrics)
 
    @session.on("function_tools_executed")
    def _on_function_tools_executed(ev: FunctionToolsExecutedEvent):
        # filename = write_event_jsonl(ev.model_dump())
        data = ev.model_dump()
        function_call_output = data.get("function_call_outputs", [])
        tool_name = [fc.get("name") for fc in function_call_output if fc.get("name")]
        <http://logger.info|logger.info>(f"[room: {ctx.room.name}] --function tools executed: {tool_name}")
    
    @session.on("close")
    def _on_close(ev: CloseEvent):
    #     filename = write_event_jsonl(ev.model_dump())
        # <http://logger.info|logger.info>(f"Close: {filename}")
        <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- Session closed, reason: {ev.reason}, error: {ev.error}")
        _is_agent_present = False
        if agent._end_of_call:
            ctx.delete_room()
        else:
            <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- Session closed, but not end of call, continuing conversation with original caller")

    
    @room.on("participant_disconnected")
    def _on_participant_disconnected(participant: rtc.Participant):
        <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- Participant disconnected: {participant.identity} ({participant.sid})")
        if participant.sid == room.local_participant.sid:
            <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- agent disconnected from the call")
        else:
            <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- remote participant disconnected from the call")
            # Check if this is the SIP transfer target disconnecting (transfer failed)
            transfer_to = dial_info.get("transfer_to")
            human_transfer_vars = agent.get_human_transfer_vars()
            
            # If this is the transfer target disconnecting and transfer was attempted, don't delete room
            # Continue the conversation with the original caller
            if transfer_to and participant.identity == transfer_to:
                if human_transfer_vars.get("attempt_status") == "success":
                    <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- SIP transfer target disconnected, transfer successful, deleting room")
                    ctx.delete_room()
                else:
                    <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- SIP transfer target disconnected, transfer failed, continuing conversation with original caller")
                    if _is_agent_present:
                        <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- agent is present, continuing conversation with original caller")
                        return
                    else:
                        <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- agent is not present, deleting room")
                        ctx.delete_room()
                    # Don't delete room - let the conversation continue
                    return
            else:
                # Remote participant disconnected, not the SIP transfer target, delete the room
                <http://logger.info|logger.info>(f"[room: {ctx.room.name}] -- Remote participant disconnected, not the SIP transfer target, deleting room")
                ctx.delete_room()

    agent = Agent(dial_info=dial_info, enter_instructions=ENTER_INSTRUCTIONS, exit_instructions=EXIT_INSTRUCTIONS, instructions=INSTRUCTIONS)

    await session.start(
        agent=agent,
        room=ctx.room,
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation.BVCTelephony(),
        ),
    )

if __name__ == "__main__":
    # cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="livekit_master_inbound_agent", job_memory_warn_mb=1024, job_memory_limit_mb=1024))
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, job_memory_warn_mb=1024, job_memory_limit_mb=1024))
t
You're very close — the disconnect handler and _is_agent_present scope are the two places to fix so the original caller stays connected when the SIP target fails. Key changes: 1) Make _is_agent_present writable from the handlers (use nonlocal or mutable container). 2) Only delete the room / call ctx.delete_room() when the original caller (local participant identity) disconnects or when transfer succeeded. 3) Treat transfer-target disconnects as non-fatal when attempt_status != "success": do not call ctx.delete_room(), do not set agent-present false, and resume speech scheduling if you had paused it. 4) Ensure you don't interpret the SIP target disconnect as input-stream detached for the original participant. Minimal edits to apply: • Use a mutable flag so handlers update it:
Copy code
_is_agent_present = {"val": True}
# in _on_close:
    _is_agent_present["val"] = False
# in disconnect handler:
    if _is_agent_present["val"]:
        ...
• Tighten the participant_disconnected logic:
Copy code
if participant.sid == room.local_participant.sid:
    # original caller disconnected -> tear down
    <http://logger.info|logger.info>(...); ctx.delete_room()
    return

# else remote participant disconnected
transfer_to = dial_info.get("transfer_to")
if transfer_to and participant.identity == transfer_to:
    # SIP target disconnected
    if agent.get_human_transfer_vars().get("attempt_status") == "success":
        <http://logger.info|logger.info>("transfer succeeded then disconnected"); ctx.delete_room()
    else:
        <http://logger.info|logger.info>("transfer failed; keep room open and resume input/scheduling")
        # resume scheduling if paused, do NOT delete room
    return

# other remote participant disconnected -> delete
ctx.delete_room()
• If you paused speech scheduling or detached input during transfer, explicitly resume it after failure so the original caller continues to be processed. If you'd like, I can produce a full patched entrypoint showing these edits and an example resume call for speech scheduling. Which would you prefer?
c
but does any of it decide the session close logic? Why does the agent close the session?
t
You're right to focus on what actually triggers a session close. Short answer: the session closes only when something explicitly tells it to close (either an explicit 
session.aclose()
or the room being deleted / the original caller disconnecting). A SIP transfer failure by itself does not force the agent session to close. What controls session close in your setup • Explicit close call: your code calls 
self.session.aclose()
(or schedules it) on the successful handover path — that directly closes the agent session. • Original caller hangup: when the original participant disconnects (or the room is deleted with 
ctx.delete_room()
), the underlying connection and session end. Your logs show CloseReason.USER_INITIATED, which means the participant or client caused the close. • Room deletion: calling 
ctx.delete_room()
removes the room and ends the session for everyone still connected. • Event handlers only react: your 
@session.on("close")
handler logs and changes 
_is_agent_present
, but that handler does not itself trigger the close — it runs after the close happens. Why your agent closed in the logs • The SIP target disconnected earlier (503), transfer failed, and your code correctly did not call 
session.aclose()
in the except block. • Later, logs show the session closed with CloseReason.USER_INITIATED — that is the original caller hanging up (or their client triggered a disconnect), not an automatic close caused by the SIP failure. • Your participant_disconnected handler can call 
ctx.delete_room()
for certain disconnects; if that handler mistakenly treated the original caller as "remote" or called 
ctx.delete_room()
when it shouldn't, that would also close the session. In your current handler you already try to avoid deleting the room when the transfer target disconnects and attempt_status != "success", which is the right approach. How to ensure SIP transfer failures never close the agent session • Never call 
session.aclose()
in the transfer failure code path. Only call it when handover truly succeeded. • Make sure 
ctx.delete_room()
is only called when the original caller disconnects or when you intentionally want to end the call. In your participant_disconnected handler: ◦ Distinguish identities (original caller vs SIP target) using 
participant.sid
or 
participant.identity
. ◦ If the disconnected identity == transfer target and 
attempt_status != "success"
, do NOT call 
ctx.delete_room()
and do NOT set agent-present false. • Ensure any global cleanup tasks that run on participant disconnect are guarded to run only for the original caller disconnect or a successful handover. • If you paused/resumed speech scheduling or detached streams during transfer, explicitly resume them after a failed transfer so the agent continues processing the original caller audio. Checks to run now 1. Confirm you never call 
session.aclose()
in the failure path. 2. Confirm the participant_disconnected handler does not call 
ctx.delete_room()
for the original caller or for failed transfer-target disconnects. 3. Verify the identity/sid comparisons are correct so you don't confuse the SIP target with the original caller. 4. Inspect logs around the CloseReason — if it's 
user_initiated
, correlate timestamps to see whether the original caller actually disconnected or your code or cleanup removed the room. If you'd like, I can: • Provide a small patched snippet of your participant_disconnected handler that strictly guards deletion and preserves the agent session on transfer failure. • Or add an explicit resume call after a failed transfer to ensure the original caller's input is reattached. Which would help you next?
c
My agent is doing the same tool call twice, why is it happening?
t
Thanks — I can help diagnose why 
transfer_call
is being invoked twice. A few likely causes and the targeted details I need to pinpoint which one applies: Possible causes • 
_transfer_in_progress
or the "attempted" flag is not reliably shared/updated (scoping issue, not using nonlocal/mutable container), so duplicate calls aren't prevented. • The agent triggers the tool from two code paths (LLM function call + explicit code) or the LLM re-generates the same function call after it didn't see the earlier call succeed. • Concurrency: two overlapping user inputs or events (e.g., repeated user utterances, STT producing multiple EOU events) cause the agent loop to invoke the tool twice. • A retry/timeout path triggers the tool again (e.g., cooldown logic not applied or timestamp check failing). • Event handlers (participant/stream events) fire and call the transfer path in addition to the model-driven tool call. What I need from you • The logs showing both 
transfer_call
tool invocations (timestamps around both calls). • Where 
_transfer_in_progress
and 
human_transfer_vars
are declared (scope) and how they’re updated. • Whether the tool is called only from the LLM (function tool) or also from other code paths. • Any STT/EOU spikes or repeated user transcripts around the time of duplicate calls. • If you have turn-level events showing function tool execution (you had a handler for "function_tools_executed" — the output around both calls helps). Quick checks you can run immediately • Ensure 
_transfer_in_progress
is in a mutable scope accessible to handlers (use a dict or declare nonlocal) so updates are visible everywhere. • Log when you set the flags (
attempted
, 
attempted_at
, 
_transfer_in_progress
) and when you early-return because a flag prevented a duplicate call. • Add a short cooldown: if 
attempted
and now - 
attempted_at
< 30s, reject further transfer attempts and log that branch. • Confirm the function_tools_executed event shows only one function call — if it shows two, the LLM made two calls; if it shows one but your transfer code ran twice, another code path triggered it. If you share the logs around the duplicate calls and the variable scope for 
_transfer_in_progress
/ 
human_transfer_vars
, I’ll point to the exact line or race causing the double invocation and suggest a minimal patch. Which logs can you paste?
c
can I make the agent call only one tool at a time, or pass the variables to agent's context to stop it from doing another tool call?
t
You're on the right track—yes, you can ensure the agent runs only one tool at a time and share state so the LLM or other code paths won't trigger a second transfer. Below are concise, practical patterns you can apply to your code: use a shared flag with proper scope, an asyncio.Lock to serialize tool calls, update 
human_transfer_vars
for cross-path visibility, and optionally add a guard in the function tool wrapper so the model can't re-invoke the tool while a transfer is in progress. 1) Use a mutable, shared state for flags (safe for inner handlers) • Problem: module/local booleans can be shadowed; handlers/closures see stale values. • Fix: use a mutable container (dict) or declare nonlocal, so all closures see updates. Example:
Copy code
state = {"transfer_in_progress": False, "attempted": False, "attempted_at": None}
2) Serialize tool calls with an asyncio.Lock • Ensures only one transfer call executes at a time even if triggered concurrently. Example:
Copy code
state["transfer_lock"] = asyncio.Lock()

async def safe_transfer_call(...):
    async with state["transfer_lock"]:
        if state["transfer_in_progress"]:
            # early return or inform user
            return "Transfer already in progress"
        state["transfer_in_progress"] = True
        try:
            # perform transfer
            ...
        finally:
            state["transfer_in_progress"] = False
3) Update human_transfer_vars (agent context) immediately and read it everywhere • Write attempt metadata before initiating the SIP call so other paths see it. • Use timestamps to enforce cooldowns. Example:
Copy code
human_transfer_vars["attempted"] = True
human_transfer_vars["attempted_at"] = datetime.now(timezone.utc)
human_transfer_vars["attempt_status"] = "in_progress"
On success/failure set 
attempt_status
to 
"success"
/ 
"failed"
. 4) Add an early-return guard in the function tool wrapper • Prevents the model from invoking the tool again while state indicates an in-progress or recent attempt. Example decorator sketch:
Copy code
def function_tool_guard(fn):
    async def wrapper(self, context, *args, **kwargs):
        if state.get("transfer_in_progress"):
            await context.session.say("A transfer is already in progress. Please wait.")
            return
        return await fn(self, context, *args, **kwargs)
    return wrapper
5) Ensure all code paths use the same state (LLM-driven calls, event handlers, manual calls) • If the LLM calls the tool, the function_tool wrapper and the lock will block duplicates. • If an event handler might call the same function, have it check 
state["transfer_in_progress"]
or 
human_transfer_vars
before calling. 6) Prevent the model from re-calling the tool by reflecting state in the agent prompt/context • After you set 
human_transfer_vars["attempted"]
, inject a short assistant message or context update that tells the model: "Transfer already in progress" so the LLM won’t attempt a second call before the state updates are visible. • Example: immediately produce a follow-up assistant turn like "Attempting to connect you now — please hold." This helps the model avoid repeating the function call. 7) Logging and telemetry • Log when you set/clear flags and when early-returning due to guard checks. That makes duplicates easy to spot in logs. 8) Example minimal integration (putting it together)
Copy code
state = {"transfer_lock": asyncio.Lock(), "transfer_in_progress": False}
human_transfer_vars = agent.get_human_transfer_vars()

@function_tool()
@function_tool_guard
async def transfer_call(self, context, ...):
    async with state["transfer_lock"]:
        if state["transfer_in_progress"]:
            await context.session.say("Transfer already in progress.")
            return
        state["transfer_in_progress"] = True
        human_transfer_vars["attempted"] = True
        human_transfer_vars["attempted_at"] = datetime.now(timezone.utc)
        human_transfer_vars["attempt_status"] = "in_progress"
        try:
            await job_ctx.api.sip.create_sip_participant(..., wait_until_answered=True)
            human_transfer_vars["attempt_status"] = "success"
            await context.session.generate_reply(instructions="Connected to a human agent.")
            # close session on success
            await self.session.aclose()
        except Exception as e:
            human_transfer_vars["attempt_status"] = "failed"
            human_transfer_vars["error_message"] = str(e)
            await context.session.say("I couldn't connect you to a human right now. Would you like a callback?")
        finally:
            state["transfer_in_progress"] = False
9) Cooldown & debounce • Enforce a cooldown by checking 
attempted_at
and refusing calls within N seconds:
Copy code
if human_transfer_vars.get("attempted") and (now - attempted_at).total_seconds() < 30:
    return "Transfer recently attempted; please wait."
10) If duplicates persist, instrument: • Log stack traces or function-call IDs when 
function_tools_executed
fires. • Capture model function-call outputs to see if the LLM generated two calls or if a second path triggered it. If you want, I can: • Provide a full patched snippet integrated with your entrypoint and disconnect handler. • Add a decorator that both acquires the lock and updates agent context, keeping the model informed. Which would you like next: the full patched entrypoint, or a ready-to-drop decorator + handler integration?
c
Explain CreateSIPParticipantRequest, what can I add to differentiate between participants, what does the input parameters to the function mean?
t
You're asking what each field in CreateSIPParticipantRequest means and how to use them to distinguish participants. Below is a concise breakdown, what each parameter does, and recommended fields to reliably differentiate participants in the room. Core fields and meaning • 
sip_trunk_id
(string): Which outbound SIP trunk to use (selects provider/caller ID). Required for placing the call. • 
sip_call_to
(string): The phone number to dial (E.164, e.g., 
+15105550100
). Required. • 
sip_number
(string): Caller ID (ANI) presented to the callee. Use this to preserve the original caller’s number during transfers; provider may override it. • 
room_name
(string): LiveKit room the SIP participant will join. Required — makes the SIP leg part of that room's call. • 
participant_identity
(string): Unique identity for the SIP participant in the room. Primary way to identify that participant in event handlers, logs, and code. • 
participant_name
(string): Human-readable label for UI/logs (not used for identity checks). • 
participant_metadata
/ 
participant_attributes
(string / map): Custom data you attach (e.g., 
{ "role": "agent", "transfer_type":"warm" }
). Useful for programmatic classification and decision logic. • 
dtmf
(string): DTMF digits to send after connect (supports 
w
for delays). • 
play_dialtone
(bool): Play a dialtone in the room while dialing. • 
wait_until_answered
(bool): If true, the API call blocks until the callee answers (or it fails). Useful when you need to know success/failure synchronously. • 
krisp_enabled
(bool): Enable provider noise cancellation for that SIP participant. • 
media_encryption
(enum): Controls SRTP behavior (
DISABLE
, 
ALLOW
, 
REQUIRE
). Which fields to use to reliably differentiate participants • 
participant_identity
— make this unique and deterministic (e.g., 
sip:+916364276271
, 
agent-123
, 
caller-<call-id>
). Use this in your participant_disconnected and other handlers to tell which participant left. • 
participant_attributes
/ 
participant_metadata
— add structured tags (role, origin, caller_id, transfer_attempt_id). These let your handlers treat SIP target vs original caller vs internal agent differently without guessing from identity strings. • 
sip_number
— set to original caller’s number when doing warm transfers so the agent or callee sees the correct caller ID and you can use it in logs/attributes. • 
room_name
— use predictable room naming to ensure you don’t mix participants from other calls. Practical patterns / best practices • Set 
participant_identity
explicitly when creating the SIP participant. Do not rely on auto-generated identities. • Immediately write state into your agent context (e.g., 
human_transfer_vars
) including 
attempt_id
, 
participant_identity
, 
attempt_status
, and 
attempted_at
. This lets event handlers and the LLM see transfer state atomically. • Use 
wait_until_answered=True
if you need the create call to return success/failure synchronously; otherwise, listen for participant join/disconnect events and correlate by 
participant_identity
. • Use 
participant_attributes
to store 
{ "role": "transfer_target", "attempt_id": "…", "original_caller": "+1…" }
so your disconnect handler can decide whether to delete the room or keep it. • In participant event handlers, compare 
participant.identity
(or 
participant.sid
) to the 
participant_identity
you set on creation; that is the most robust check. • If you need to resume the original caller after a failed transfer, make sure you resume speech scheduling / re-attach input for the original caller and do not treat the SIP target disconnect as the original caller disconnect. Example (Python sketch)
Copy code
req = CreateSIPParticipantRequest(
  sip_trunk_id="ST_abc",
  sip_call_to="+19175550100",
  sip_number="+14155550100",               # preserve caller ID
  room_name="room-xyz",
  participant_identity="transfer-<attempt_id>",
  participant_name="Support Agent",
  participant_attributes={"role":"transfer_target","attempt_id":"<attempt_id>"},
  wait_until_answered=True
)
When to rely on which mechanism • Need immediate success/failure? Use 
wait_until_answered=True
and check the RPC result/exception. • Prefer async event-driven handling? Use 
participant_identity
+ 
participant_attributes
and watch participant join/disconnect events, correlating by identity or attempt_id. • Want the model/event handlers to avoid duplicate actions? Set 
human_transfer_vars["attempted"]
and 
human_transfer_vars["participant_identity"]
before calling create, and have handlers check those fields. If you want, I can produce a small code snippet that: • Builds a deterministic 
participant_identity
and attributes for a transfer attempt, • Shows how to set 
human_transfer_vars
before calling 
create_sip_participant
, • And shows the minimal participant_disconnected handler that uses those fields to decide whether to keep or delete the room. Which snippet would you prefer? Sources: Make outbound calls | LiveKit docs | SIP APIs | LiveKit docs | Preserving Caller ID During SIP Transfers in LiveKit Agents | Enhanced noise cancellation | LiveKit docs | Agents telephony integration | LiveKit docs
3 Views
Open in Slack
PreviousNext
Powered by