from livekit import rtc
from livekit.agents import Agent, RunContext, function_tool

# Pre-synthesize a hold message once at startup
HOLD_FRAMES: list[rtc.AudioFrame] = []

async def preload_hold_message(tts) -> None:
    global HOLD_FRAMES
    async for event in tts.synthesize("Let me check that for you."):
        HOLD_FRAMES.append(event.frame)

class MyAgent(Agent):
    @function_tool()
    async def check_order_status(
        self,
        context: RunContext,
        order_id: str,
    ) -> str:
        """Check the status of an order.

        Args:
            order_id: The order ID to look up.
        """
        async def cached_audio():
            for frame in HOLD_FRAMES:
                yield frame

        # Play the hold message concurrently — don't await
        hold_handle = context.session.say(
            "Let me check that for you.",
            audio=cached_audio(),
            add_to_chat_ctx=False,
        )

        # Call the external API (runs while the hold message plays)
        result = await fetch_order_status(order_id)

        # If the API returned before the hold message finished, cancel it
        if not hold_handle.interrupted and not hold_handle.done():
            hold_handle.interrupt()

        return result

i found this code in this doc https://docs.livekit.io/agents/multimodality/audio/#caching-tts Can anyone confirm is it possible to play a music from a file here ?

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

AUDIO_FILE = Path(file).resolve().parent.parent / "assets/audio/hold_music.wav" if function_name == "addServiceNowCommentAndWaitForResponse": context.disallow_interruptions() # await self._announce_tool_message(context, tool_def, "request-start") # Load music async def hold_music(): async for frame in audio_frames_from_file(AUDIO_FILE): yield frame # Start playing music (DO NOT await) hold_handle = context.session.say( "Please stay on the line while I work on your case and get the further updates. This may take a moment", audio=hold_music(), add_to_chat_ctx=False, ) try: # Run long API result = await execute_vapi_server_tool( tool_def["server"], args, tool_def, tool_call_id ) # Stop music if still playing if not hold_handle.done(): hold_handle.interrupt() await self._announce_tool_message(context, tool_def, "request-complete") if isinstance(result, dict): return result.get("text") or result.get("content") or str(result) return str(result) except Exception as e: logger.error(f"{function_name} failed: {str(e)}") if not hold_handle.done(): hold_handle.interrupt() await self._announce_tool_message(context, tool_def, "request-failed") return f"Tool '{function_name}' failed: {str(e)}" # Announce request start await self._announce_tool_message(context, tool_def, "request-start") try: if "server" in tool_def and "url" in tool_def["server"]: logger.info(f"Using Vapi-style server endpoint for {function_name}") result = await execute_vapi_server_tool(tool_def["server"], args, tool_def, tool_call_id) # Announce request completion await self._announce_tool_message(context, tool_def, "request-complete") if isinstance(result, dict): if "text" in result: return result["text"] if "content" in result: return result["content"] return str(result) return str(result) else: raise RuntimeError(f"No server configuration found for tool '{function_name}'") except Exception as e: logger.error(f"Tool '{function_name}' failed: {str(e)}") # Announce request failure await self._announce_tool_message(context, tool_def, "request-failed") return f"Tool '{function_name}' failed: {str(e)}" can you check this part of code of mine is correctly implemented ?

Thanks — I reviewed the code you pasted that starts the hold music with

context.session.say(...)

, runs a long API call, then interrupts the music. Overall the approach is correct and follows the same pattern as the TTS example: create an async generator that yields frames, call

context.session.say(...)

without awaiting to play concurrently, run the long task, and interrupt the handle if still playing. A few concrete checks and suggestions to make it more robust: 1) Ensure

audio_frames_from_file

yields

rtc.AudioFrame

objects • The helper

audio_frames_from_file(AUDIO_FILE)

should return an async generator of

rtc.AudioFrame

. That matches your

hold_music()

wrapper — good. 2) Keep the async generator simple and lazy • Your implementation: async def hold_music(): async for frame in audio_frames_from_file(AUDIO_FILE): yield frame is correct: it yields frames lazily as playback consumes them. 3) Don’t await the

say()

call when you want concurrency — you already do this • You correctly call

hold_handle = context.session.say(..., audio=hold_music(), ...)

and then await the long API. This will let audio play while the API runs. 4) Interrupt logic • You check

if not hold_handle.done(): hold_handle.interrupt()

in both success and exception paths. That’s correct. A couple of small improvements: ◦ You may also want to check

if not hold_handle.interrupted and not hold_handle.done():

before calling

interrupt()

(your TTS example used that), but calling

interrupt()

when already interrupted is usually harmless. ◦ If you want to ensure abrupt stop immediately, call `hold_handle.interrupt()`; if you prefer polite stop (let it finish current frame), confirm what

interrupt()

semantics are for your SDK version. 5) disallow_interruptions() call • You call

context.disallow_interruptions()

for this function — that will prevent external interruptions while the tool runs. Make sure that is the intended UX (user won’t be able to stop the music/call flow while the long task is running). 6) Announcements ordering • You commented out

await self._announce_tool_message(context, tool_def, "request-start")

near the top and later call it after the long tool run. If you want the user to hear a start announcement before the long API runs (and before music), move the announce call to before starting the music/API, or keep it after depending on desired UX. 7) Resource & format considerations • Confirm the audio file sample rate and channels match what the helper expects (your

audio_frames_from_file

may accept sample_rate/num_channels options). If you need lower memory use, stream from disk rather than preloading all frames. 8) Error handling and return values • Your try/except returns useful messages and interrupts the hold handle on exception — good. • Small cleanup: in some branches you use several nested returns; ensure consistent formatting so the function always returns the expected string. 9) Looping music • If your hold music is shorter than the long operation and should loop, you’ll need to yield frames repeatedly (e.g., repeat the file or preload frames into a list and loop). Right now the music will play once and stop when frames are exhausted (unless your

audio_frames_from_file

implementation loops). 10) Example small tidy-up (optional) • You can simplify the audio generator inline: async def hold_music(): async for f in audio_frames_from_file(AUDIO_FILE): yield f If you’d like, I can: • Show a short example that loops the WAV seamlessly. • Verify the exact

audio_frames_from_file

signature and give the recommended options (sample rate / channels) for best results. • Check the precise

Handle

API (

done()

,

interrupted

,

interrupt()

) for your LiveKit SDK version and adjust the interrupt checks. Do you want me to provide a looping example or verify

audio_frames_from_file

arguments for your file? , with interactive elements

yes

Great — I’ll provide a short looping example plus guidance on

audio_frames_from_file

options and a couple of small safety tips to make the hold music robust. Looping hold-music example 1. This example uses

audio_frames_from_file()

to stream frames and loops them by preloading frames into memory once, then repeatedly yielding them while the long task runs.

from pathlib import Path
from livekit.agents.utils.audio import audio_frames_from_file

AUDIO_FILE = Path(__file__).resolve().parent.parent / "assets/audio/hold_music.wav"

async def preload_frames(path: Path):
    frames = []
    async for frame in audio_frames_from_file(path):
        frames.append(frame)
    return frames

async def looping_hold_music(frames):
    # Loop indefinitely until the say handle is interrupted
    while True:
        for frame in frames:
            yield frame

# Usage inside your function
frames = await preload_frames(AUDIO_FILE)

# Start playing (do NOT await) so it runs concurrently with the long task:
hold_handle = context.session.say(
    "Please stay on the line while I work on your case and get further updates.",
    audio=looping_hold_music(frames),
    add_to_chat_ctx=False,
)

try:
    # Run your long API call
    result = await execute_vapi_server_tool(tool_def["server"], args, tool_def, tool_call_id)
finally:
    # Ensure music stops regardless of success/exception
    if not hold_handle.done():
        hold_handle.interrupt()

Notes and recommendations • Preloading vs streaming: ◦ Preloading into

frames

makes looping seamless and avoids gaps between repeats. ◦ If the file is large and memory is a concern, stream from disk and implement a streaming loop that re-opens/rewinds the source each loop; preloading is preferred for short hold clips. •

audio_frames_from_file

arguments: ◦ It accepts common audio formats (WAV, MP3, OGG, MP4, etc.) and internally decodes via FFmpeg/PyAV. ◦ If needed, you can pass

sample_rate

and

num_channels

to ensure the frames match the session codec expectations (e.g.,

sample_rate=24000

,

num_channels=1

). Example:

async for frame in audio_frames_from_file(AUDIO_FILE, sample_rate=24000, num_channels=1):
        yield frame

• Interrupt semantics: ◦ Calling

hold_handle.interrupt()

in your success and except paths is correct. You can optionally check

if not hold_handle.interrupted and not hold_handle.done()

before calling

interrupt()

, but calling interrupt repeatedly is harmless. • UX:

context.disallow_interruptions()

prevents users interrupting during the operation. Keep it only if you want to block user-initiated interruptions. • Seamless looping: Preloading frames ensures contiguous playback without gaps between loops. If you stream from disk you may hear a small gap unless you carefully buffer frames. If you want, I can: • Convert this to a streaming loop (no preload) to save memory. • Show the exact

audio_frames_from_file

signature you have in your environment and recommend `sample_rate`/`num_channels` values for your LiveKit setup. Which would you prefer? , with interactive elements