session.say()

session.generate_reply()

session.say()

session.generate_reply()

from livekit_agents import AgentSession, openai, function_tool, RunContext

# Configure the agent session with a realtime LLM and XAI TTS voice
session = AgentSession(
    llm=openai.realtime.RealtimeModel(
        model="openai.realtime.some-realtime-model",  # replace with your realtime model id
        modalities=["text"]
    ),
    tts="xai/think-fast-1.0"  # XAI Think Fast 1.0 TTS voice
)

# Example tool that speaks immediately while performing a long-running call
@function_tool()
async def lookup_order(context: RunContext, order_id: str):
    # Start spoken hold message without awaiting so it plays concurrently
    context.session.say("Let me look that up for you while I fetch the details.")
    # Simulate or call external async API
    result = await some_async_order_api.get(order_id)
    # Optionally generate a spoken summary after the result arrives
    await context.session.generate_reply(f"I found your order. Status: {result['status']}.")
    return result

# Example tool using generate_reply during the tool execution
@function_tool()
async def check_availability(context: RunContext, item_id: str):
    # Use generate_reply to synthesize a shorter spoken response during processing
    await context.session.generate_reply("Checking availability now.")
    avail = await inventory_api.check(item_id)
    return avail

# Register tools and start the agent session as your app requires
session.register_tool(lookup_order)
session.register_tool(check_availability)

# Start or run the session according to your app's lifecycle
# e.g., await session.start()

some_async_order_api

session.say()

session.generate_reply()

import os
import asyncio
from livekit.agents import AgentSession, function_tool, RunContext
from livekit.plugins import xai

# Ensure XAI API key is set in environment
# export XAI_API_KEY="your_xai_api_key"
# export LIVEKIT_API_KEY="your_livekit_api_key"   # if needed for LiveKit services

# Create an AgentSession using the xAI plugin realtime model and think-fast-1.0 voice
session = AgentSession(
    llm=xai.realtime.RealtimeModel(voice="think-fast-1.0"),  # xAI realtime model with Think Fast 1.0 voice
    # If you want separate TTS and realtime LLM, you can set tts="xai/think-fast-1.0" instead
)

# Example long-running tool that speaks a hold message using session.say()
@function_tool()
async def lookup_order(context: RunContext, order_id: str):
    # fire-and-forget spoken hold message so audio plays while we fetch data
    context.session.say("One moment — I’m pulling up your order details.")
    # simulate long-running external call
    await asyncio.sleep(2)  # replace with: result = await orders_api.get(order_id)
    result = {"order_id": order_id, "status": "shipped"}
    # speak a summary after the result arrives (await so reply completes before continuing)
    await context.session.generate_reply(f"I found order {order_id}. Status: {result['status']}.")
    return result

# Example tool that uses generate_reply() during processing and returns data
@function_tool()
async def check_availability(context: RunContext, item_id: str):
    # generate a short spoken message and receive a SpeechHandle
    handle = await context.session.generate_reply(f"Checking availability for item {item_id}.")
    # perform the check
    await asyncio.sleep(1)  # replace with: avail = await inventory_api.check(item_id)
    avail = {"item_id": item_id, "available": True}
    # optionally inspect handle.interrupted to decide next steps
    if getattr(handle, "interrupted", False):
        # user interrupted — optionally stop further speech or alter behavior
        return {"status": "interrupted"}
    # finalize with spoken confirmation
    await context.session.generate_reply(f"Item {item_id} is {'available' if avail['available'] else 'not available'}.")
    return avail

# Register tools with the session
session.register_tool(lookup_order)
session.register_tool(check_availability)

# Example entrypoint: start the session (adapt to your app runtime)
async def main():
    # Start or attach the session as your app requires
    await session.start()   # adjust per your app lifecycle (or await web server bootstrap)
    # Keep running for demo purposes
    try:
        while True:
            await asyncio.sleep(3600)
    finally:
        await session.stop()

if __name__ == "__main__":
    asyncio.run(main())

context.session.say(...)

generate_reply()

handle.interrupted

XAI_API_KEY

llm

tts="xai/think-fast-1.0"

AgentSession