I have a voicemail message thats supposed to be said (the generate reply is within the tool call), and then we hangup immediately after that. If the voicemail message is interrupted I need to make sure that we don't hang up after and that we handle the interruption gracefully.

Thanks for reaching out, Aumesh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

Provide the full tool call

Thanks — I understand you want the full tool call (complete, ready-to-use implementation) that plays a voicemail, detects interruptions, and only hangs up if the message completed. Below is a self-contained Python implementation you can adapt to your environment. It uses the same pattern: call

generate_reply()

(or

say()

in this example), wait for playout with interruption detection, and only call the hangup routine when the speech completed without interruption. There are two usage patterns shown: a blocking flow (awaiting playout) and a non-blocking callback flow. Please adapt names like

ctx.agent_session.say

,

ctx.job_context.api.room.delete_room

, and

hangup_call()

to match your actual SDK methods if they differ. Full tool call (complete example)

import asyncio
import logging
from typing import Optional
from livekit.agents import AgentContext, llm, tools
from livekit import api

logger = logging.getLogger("voicemail-tool")
logger.setLevel(<http://logging.INFO|logging.INFO>)


class VoicemailTool(tools.FunctionContext):
    def __init__(self, ctx: AgentContext) -> None:
        super().__init__()
        self._ctx = ctx
        self._current_task: Optional[asyncio.Task] = None

    @tools.context_tool
    async def detect_and_handle_voicemail(self) -> str:
        """
        Play a voicemail message, detect interruptions, and hang up only if
        the message completed without interruption.
        """
        <http://logger.info|logger.info>("Starting voicemail flow...")

        # The voicemail text to play
        voicemail_message = "I'm sorry I can't reach you right now. I'll call back later."

        # --- Blocking approach: generate/play voicemail and wait for playout ---
        try:
            # Generate/play the voicemail and allow interruptions so we can detect them
            handle = await self._ctx.agent_session.generate_reply(
                instructions=voicemail_message,
                allow_interruptions=True
            )

            # Wait until playout completes or is interrupted
            try:
                # Wait for playout to finish. The SDK will mark handle.interrupted if it was interrupted.
                await handle.wait_for_playout()

                # If playout completed without interruption => likely voicemail
                <http://logger.info|logger.info>("Voicemail playback completed without interruption. Proceeding to hang up.")

                # Hang up: example using room delete API (adapt to your hangup method)
                try:
                    await self._ctx.job_context.api.room.delete_room(
                        api.DeleteRoomRequest(room=self._ctx.job_context.room.name)
                    )
                    return "Voicemail left successfully; call ended."
                except Exception as e:
                    logger.error(f"Failed to end call after voicemail: {e}")
                    return "Voicemail left but failed to end call."

            except Exception:
                # If an exception occurs while waiting, check interrupted flag
                if getattr(handle, "interrupted", False):
                    <http://logger.info|logger.info>("Voicemail playback was interrupted by the callee. Continuing the session.")
                    # Continue the conversation — do not hang up
                    await self._ctx.agent_session.generate_reply(
                        instructions="I see you're there — how can I help you today?"
                    )
                    return "Playback interrupted; switched to live conversation."
                else:
                    # Unknown error while waiting
                    logger.exception("Error while waiting for playout.")
                    return "Error while waiting for voicemail playout."

        except Exception as e:
            logger.exception(f"Error during voicemail playback: {e}")
            return f"Error playing voicemail: {str(e)}"

    @tools.context_tool
    async def detect_and_handle_voicemail_nonblocking(self) -> str:
        """
        Non-blocking variant: attach a done-callback to the speech handle and return immediately.
        """

        voicemail_message = "I'm sorry I can't reach you right now. I'll call back later."
        try:
            handle = await self._ctx.agent_session.generate_reply(
                instructions=voicemail_message,
                allow_interruptions=True
            )
        except Exception as e:
            logger.exception(f"Failed to start voicemail playback: {e}")
            return f"Failed to start playback: {str(e)}"

        # Define async callback to run when speech completes/interrupts
        async def on_speech_done(h):
            try:
                if getattr(h, "interrupted", False):
                    <http://logger.info|logger.info>("Non-blocking callback: playback interrupted; engaging live session.")
                    await self._ctx.agent_session.generate_reply(
                        instructions="I see you're there — how can I help you?"
                    )
                else:
                    <http://logger.info|logger.info>("Non-blocking callback: playback completed; hanging up.")
                    try:
                        await self._ctx.job_context.api.room.delete_room(
                            api.DeleteRoomRequest(room=self._ctx.job_context.room.name)
                        )
                        <http://logger.info|logger.info>("Call ended after voicemail.")
                    except Exception as e2:
                        logger.error(f"Failed to end call in callback: {e2}")
            except Exception as cb_exc:
                logger.exception(f"Error in on_speech_done callback: {cb_exc}")

        # The SDK may expose add_done_callback or similar; adapt if method name differs.
        try:
            handle.add_done_callback(lambda h: asyncio.create_task(on_speech_done(h)))
            return "Voicemail playback started (non-blocking)."
        except Exception as e:
            logger.exception(f"Failed to attach callback: {e}")
            return "Playback started but failed to attach completion callback."

How this satisfies your requirement • It sets

allow_interruptions=True

when generating the voicemail so interruption is detectable. • The blocking variant awaits

handle.wait_for_playout()

and checks

handle.interrupted

to decide whether to hang up. • The non-blocking variant attaches an async callback that checks

handle.interrupted

and branches accordingly. • Hangup is only called when playback completed without interruption. Small implementation notes / adaptions you may need • Replace

self._ctx.agent_session.generate_reply

with

say()

if your environment uses that naming. • Replace room deletion with your actual hangup routine if you hang up differently (e.g.,

hangup_call()

). • Ensure

handle.wait_for_playout()

,

handle.interrupted

, and

handle.add_done_callback()

match your SDK's method/property names. • Add a timeout around

wait_for_playout()

if you want to guard against indefinite waits. Would you like a version of this full tool call adjusted to a specific SDK or runtime (for example: exactly

say()

instead of

generate_reply()

, or a Node/TypeScript variant)? If so, tell me which language/SDK and whether you prefer the blocking or non-blocking pattern and I’ll produce that exact snippet. Sources: SIP Lifecycle Management Agent | LiveKit Documentation | Agents telephony integration | LiveKit Documentation | How to implement call greeting retries in voice agents | LiveKit | How to detect when an agent has finished speaking | LiveKit

Why can't I just do

@function_tool
def detect_answering_machine:
    handle = await self.session.generate_reply(...., allow_interruptions=True)
    await handle.waitForPlayout()
    if not handle.interrupted:
         hangup

Good question — your simplified snippet expresses the correct high-level idea, but there are practical reasons you should be careful and a few important details to handle to make it reliable in production. I’ll explain the pitfalls and show a minimal but safe pattern you can use. Why the naive version can fail • Race conditions: If you immediately await

wait_for_playout()

and then act, there can be timing windows where the interruption state changes (depending on the SDK/platform). The SDK may set

interrupted

only after certain internal callbacks complete — so checking it immediately after

wait_for_playout()

without handling SDK semantics can be unreliable. • Exceptions vs flags: Some SDKs raise an exception when playout is interrupted instead of simply returning; you must handle that exception and still inspect

handle.interrupted

. If you only rely on the success path you might miss the interrupted case (or crash). • Blocking behavior and concurrency: A straight blocking wait ties up the current task. If your environment expects the session to remain responsive to other events (DTMF, incoming audio, transfers), blocking may prevent those events from being processed in time. • Timeouts and stuck playouts: Network or SDK issues can make

wait_for_playout()

hang. You should guard with a timeout so the call doesn’t hang forever. • Non-blocking architectures: If you use an event-driven runtime, using a blocking wait may be undesirable; attaching a completion callback is safer and integrates with the event loop. Key things to do instead • Allow interruptions: set

allow_interruptions=True

(you already have this). • Handle both the normal completion and the interrupted completion explicitly: ◦ Use try/except around

wait_for_playout()

because interruption may be surfaced as an exception. ◦ After the wait (or in the exception handler), check

handle.interrupted

to decide whether to hang up. • Use a timeout around the wait to avoid indefinite blocking. • Prefer a non-blocking callback pattern if you need the session to stay responsive. Minimal robust pattern (conceptual) 1. start playback with interruptions allowed 2. wait for playout with a timeout and catch exceptions 3. check

handle.interrupted

in both success and exception paths 4. only hang up when

handle.interrupted

is False Example (concise, robust)

handle = await self.session.generate_reply(..., allow_interruptions=True)

try:
    # guard with a timeout to avoid hanging forever
    await asyncio.wait_for(handle.wait_for_playout(), timeout=10.0)
    # playback finished normally
except Exception:
    # playback may have been interrupted or another error occurred
    # fall through to check the flag below

# Inspect the flag in both cases
if getattr(handle, "interrupted", False):
    # someone answered — continue the session
    await self.session.generate_reply("I see you're there—how can I help?")
else:
    # likely voicemail — hang up
    await hangup_call()

When to use a non-blocking callback • If you need to keep processing other events while the message plays, attach a done callback and handle hangup/continuation from that callback. This avoids blocking the main flow and reduces race windows. Summary • Your snippet captures the right intent, but make it robust by handling exceptions, adding a timeout, and being mindful of blocking vs. non-blocking behavior. Check

handle.interrupted

in both normal and exceptional flows — and prefer callbacks if your runtime requires responsiveness. If you tell me which SDK/language and whether you prefer blocking or non-blocking, I can give you a final exact snippet tailored to your environment.

I use the python SDK

Thanks — since you use the Python SDK, your simple idea is essentially correct: generate the voicemail with interruptions allowed, wait for playout, and only hang up if

handle.interrupted

is False. The important part is to make that flow robust in production. Below is a concise, production-ready Python snippet and a short explanation of the key safety checks you should keep. Full, robust pattern (Python SDK)

import asyncio
import logging

logger = logging.getLogger("voicemail")
logger.setLevel(<http://logging.INFO|logging.INFO>)

async def detect_answering_machine_and_maybe_hangup(session, job_ctx):
    # Start playback and allow interruptions so we can detect a live answer
    handle = await session.generate_reply(
        instructions="Hi, this is Acme. I'm calling about your order. I'll call back later. Goodbye.",
        allow_interruptions=True
    )

    try:
        # Guard the wait so we don't block forever if something goes wrong
        await asyncio.wait_for(handle.wait_for_playout(), timeout=15.0)

    except asyncio.TimeoutError:
        logger.warning("wait_for_playout timed out; treating as error and not hanging up automatically.")
        # Decide policy: here we avoid hanging up automatically
        return "timeout"

    except Exception as e:
        # Some SDKs surface interruptions as exceptions; fall through to checking flag
        <http://logger.info|logger.info>(f"wait_for_playout raised: {e}; will inspect handle.interrupted")

    # Inspect the interruption flag in both success and exception paths
    if getattr(handle, "interrupted", False):
        <http://logger.info|logger.info>("Playback was interrupted — do NOT hang up. Continue the session.")
        # Example follow-up prompt to transition to live conversation
        await session.generate_reply(instructions="I see you're there — how can I help you?")
        return "interrupted"
    else:
        <http://logger.info|logger.info>("Playback completed without interruption — hang up the call.")
        # Use your hangup method (room.disconnect / delete_room / hangup_call)
        try:
            await job_ctx.room_context().room.disconnect()
            return "hung_up"
        except Exception as e:
            logger.error(f"Failed to hang up cleanly: {e}")
            return "hangup_failed"

Why include the extra bits (timeouts, exception handling)? • Some SDKs raise exceptions when playback is interrupted; catching exceptions and then checking

handle.interrupted

ensures you don't mis-handle that case. • A timeout prevents your coroutine from hanging forever if the SDK or network misbehaves. • Checking

handle.interrupted

after the wait (in both normal and exception paths) is the reliable way to know whether a human answered. • If your runtime must remain responsive, use a non-blocking callback instead of awaiting (I can provide that if you prefer). If you want, I can: • Turn this into a non-blocking callback example using

handle.add_done_callback

. • Replace

job_ctx.room_context().room.disconnect()

with whatever exact hangup API you use. • Add optional behaviors (e.g., DTMF detection, retries, recording). Which of those would you like next? Sources: How to detect when an agent has finished speaking | LiveKit | Bringing AI avatars to voice agents | OpenAI and LiveKit partner to turn Advanced Voice into an API | VAD and turn detection configuration guide | LiveKit | Agents telephony integration | LiveKit Documentation