How could I limite the max time per session?

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

Python sdk

Thanks — I’m responding to your request about limiting the max time per session using the Python SDK. You can enforce a session timeout in your agent logic by starting an asyncio background task when the session begins, sending an optional warning, and calling

ctx.shutdown()

when the timer expires. Make sure to cancel the timer if the user disconnects early so you don’t attempt to shut down an already-closed session. Example pattern (Python / asyncio):

import asyncio

class MyAgent:
    def __init__(self):
        self._session_timer = None

    async def on_enter(self, userdata, ctx, max_seconds: int = 180):
        # Start the session timer when the session starts
        self._session_timer = asyncio.create_task(self._session_timeout(ctx, max_seconds))

    async def _session_timeout(self, ctx, max_seconds):
        # Optional: warn user 30s before end
        warning_seconds = 30
        if max_seconds > warning_seconds:
            await asyncio.sleep(max_seconds - warning_seconds)
            await self.session.say(f"{warning_seconds} seconds left in this session.")
            await asyncio.sleep(warning_seconds)
        else:
            await asyncio.sleep(max_seconds)

        await self.session.say("Session time is up. Thanks for joining!")
        await asyncio.sleep(1.0)
        ctx.shutdown("Session timer expired")

    async def on_exit(self, userdata, ctx):
        # Cancel the timer if the session ends early
        if self._session_timer and not self._session_timer.done():
            self._session_timer.cancel()
            try:
                await self._session_timer
            except asyncio.CancelledError:
                pass

Notes and best practices • Use a friendly closing message and consider a short warning (e.g., 30s left). • Make

max_seconds

configurable per room/agent so different sessions can have different limits. • Keep the timer cancellation robust to avoid race conditions (check

.done()

and handle

CancelledError

). • If you want the room removed when the session ends, combine this with

delete_room_on_close=True

in

RoomOptions

. If you’d like, I can: • Provide a shorter immediate-shutdown variant (no warning), • Convert this example to the exact agent class structure you’re using, or • Show a JavaScript example for cross-language parity. Which would help you most? Sources: Build a salary negotiation coach with a LemonSlice avatar | LiveKit | Build a salary negotiation coach with a LemonSlice avatar | LiveKit | Job lifecycle | LiveKit Documentation | Agent session | LiveKit Documentation, with interactive elements

This is my current code: ############################### INICIALIZACIÓN DE SERVIDOR LIVEKIT Y LIVEAVATAR ############################### class Assistant(Agent): def __init__(self) -> None: super().__init__( instructions="", ) _session_timer = None server = AgentServer(port=os.getenv("PRODUCTION_PORT")) #Configuración voz TTS setVoices = elevenlabs.VoiceSettings(stability=0.68,similarity_boost=0.54,style=0.47,speed=0.9,use_speaker_boost=True); @breezy-insurance-66772.rtc_session() async def my_agent(ctx: JobContext): # Logging setup # Add any other context you want in all log entries here ctx.log_context_fields = { "room": ctx.room.name, } # Set up a voice AI pipeline using OpenAI, Cartesia, AssemblyAI, and the LiveKit turn detector session = AgentSession( tts=elevenlabs.TTS( model="eleven_flash_v2_5", voice_id="ERYLdjEaddaiN9sDjaMX", voice_settings=setVoices, language="es" ), userdata={} ) @session.on("") #Cuando el usuario se desconecte, se detiene la sesión y el avatar @session.on("close") def on_close(close_event): print("Sesión cerrada, deteniendo avatar y borrando conversationID", session) # # Add a virtual avatar to the session, if desired # # For other providers, see https://docs.livekit.io/agents/models/avatar/ avatar = liveavatar.AvatarSession( avatar_id=os.getenv('LIVEAVATAR_AVATAR_ID'), # See https://docs.livekit.io/agents/models/avatar/plugins/hedra ) # # Start the avatar and wait for it to join await avatar.start(session, room=ctx.room) # Start the session, which initializes the voice pipeline and warms up the models await session.start( agent=Assistant(), room=ctx.room, room_options=room_io.RoomOptions( text_input=room_io.TextInputOptions(text_input_cb=custom_text_input_handler_tts), audio_input=False, audio_output=True ), ) await asyncio.sleep(2) #Saludamos al principio print("Audios de espera precargados: ", list(precarga_audios.keys())) await play_audio_once(session, saludo_text) await ctx.connect() if name == "__main__": print("Iniciando servidor de agentes...") cli.run_app(server)

Thanks — I’m responding to your request about adding a max session time to the Python code you shared. You can add an asyncio timer that starts when the session begins and calls

ctx.shutdown()

when time expires. Place the timer creation just after session start (after

await session.start(...)

) and cancel it in your session close handler so you don’t try to shut down an already-closed session. Below is a focused, minimal integration showing where to add the timer to your code pattern and how to cancel it when the user disconnects. I kept variable names similar to yours so you can drop it into your file. 1) Add a timer task to the session scope and start it after session start. 2) Warn the user (optional) before shutdown. 3) Cancel the task on close. Example integration (adapt to your indentation/style):

import asyncio

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="")
        self._session_timer = None

@FlashCall Server.rtc_session()
async def my_agent(ctx: JobContext):
    # ... your existing setup ...
    session = AgentSession(...)

    # Start avatar, start session as you already do
    avatar = liveavatar.AvatarSession(avatar_id=os.getenv('LIVEAVATAR_AVATAR_ID'))
    await avatar.start(session, room=ctx.room)

    await session.start(
        agent=Assistant(),
        room=ctx.room,
        room_options=room_io.RoomOptions(
            text_input=room_io.TextInputOptions(text_input_cb=custom_text_input_handler_tts),
            audio_input=False,
            audio_output=True,
        ),
    )

    # Start session timer (example: 3 minutes = 180s)
    MAX_SECONDS = 180
    WARNING_SECONDS = 30  # optional warning before end

    # create a background task and attach to session/assistant for cancellation
    async def _session_timeout_task():
        try:
            if MAX_SECONDS > WARNING_SECONDS:
                await asyncio.sleep(MAX_SECONDS - WARNING_SECONDS)
                await session.say(f"{WARNING_SECONDS} seconds left in this session.")
                await asyncio.sleep(WARNING_SECONDS)
            else:
                await asyncio.sleep(MAX_SECONDS)

            await session.say("Session time is up. Thanks for joining!")
            await asyncio.sleep(1.0)
            ctx.shutdown("Session timer expired")
        except asyncio.CancelledError:
            # Timer cancelled because session ended early; just exit
            return

    # save task so it can be cancelled on close
    ctx._session_timer = asyncio.create_task(_session_timeout_task())

    # continue with the rest of your flow
    await asyncio.sleep(2)
    await play_audio_once(session, saludo_text)
    await ctx.connect()

    # Register close handler to cancel timer and cleanup
    @session.on("close")
    def on_close(close_event):
        # cancel timer if still running
        timer = getattr(ctx, "_session_timer", None)
        if timer and not timer.done():
            timer.cancel()
        print("Sesión cerrada, deteniendo avatar y borrando conversationID", session)

Notes and tips • Put the timer on

ctx

(or on the Agent instance) so your close handler can reliably find and cancel it. • Use a short delay (like 1s) after the final message before calling

ctx.shutdown()

so audio finishes playing. • If you prefer no warning, set

WARNING_SECONDS = 0

or remove that block. • Adjust

MAX_SECONDS

per session by passing it in or reading a config/env var if you need flexibility. • Ensure your

on_close

handler runs in the right context — if it needs to be

async def

instead of sync, adapt cancel logic accordingly (await cancellation handling as needed). Would you like me to: • Insert this directly into your full posted code with correct indentation and handlers, or • Provide a variant that immediately deletes the room on close using

RoomOptions(delete_room_on_close=True)

? Which option helps you most? , with interactive elements

Okay, now I'm trying to stop the AgentSession.say from my python server when I press my speak button from my React client. I'm handling my client inputs this way:

#Función que recibe un mensaje y actua como TTS

async def custom_text_input_handler_tts(session: AgentSession, event: room_io.TextInputEvent) -> None:

try:

payload_bytes = event.text

print("Estado actual del agente: ",session.agent_state)

#Si desde el cliente se ha enviado un mensaje que no esté vacio

if payload_bytes != "" or not(not payload_bytes):

payload_split = payload_bytes.split("::")

userMessage = payload_bytes.split("::")[0] if len(payload_split) > 0  else payload_bytes

userKey = payload_bytes.split("::")[1] if len(payload_split) > 1 else "unknown"

userID = payload_bytes.split("::")[2] if len(payload_split) > 2 else "unknown"

participant = session.room_io.linked_participant

participant_identity = participant.sid if participant else "unknown"

print(f"Mensaje recibido de {participant_identity}: {userMessage}")

#Elegimos el audio de espera con la frase que toca

frase_espera = handle_waiting_phrases()

await play_audio_once(session,frase_espera)

session._update_agent_state("thinking")

print("Frase de espera acabada")

if "assistantKey" not in session.userdata:

session.userdata["assistantKey"] = userKey

#Miramos si ya hay creado un hilo de conversación para este usuario, si no lo hay, se crea uno nuevo y se guarda en la sesión

if "threadID" not in session.userdata or session.userdata["threadID"] == "" :

session.userdata["threadID"] = await getThreadID(session.userdata["assistantKey"])

print ("Se ha obtenido el siguiente ID: ", session.userdata["threadID"])

#Si se ha incluido un código de usuario lo añadimos al mensaje

if userID != "unknown":

userMessage = userMessage + ". Además, mi identificador de usuario es el siguiente: " + userID

#Petición al servidor openAI en node

respuestaChatBot = await askChatbot(userMessage,session.userdata["threadID"],session.userdata["assistantKey"])

async def safe_way(text: str):

#await session.interrupt()

#session._update_agent_state("listening")

handle = session.say(text, allow_interruptions=True)

handle.add_done_callback(lambda _: print("Se ha acabado la frase"))

await handle

session._update_agent_state("listening")

print("Estado actual del agente: ",session.agent_state)

if(respuestaChatBot != "" or not(not respuestaChatBot)):

respuestaChatBot = respuestaChatBot.replace(":", ",")

respuestaChatBot = re.sub(r"【.*?】", "", respuestaChatBot)

asyncio.create_task(safe_way(respuestaChatBot))

else:

respuestaChatBot = "Lo siento, pero no he podido procesar correctamente la solicitud. ¡Inténtalo de nuevo!"

asyncio.create_task(safe_way(respuestaChatBot))

else:

text_to_speak = "Lo siento, pero no te he podido entender correctamente. ¡Inténtalo de nuevo!"

session._update_agent_state("listening")

await session.say(respuestaChatBot, allow_interruptions=True)

except Exception as e:

text_to_speak = "Lo siento, pero no he podido procesar correctamente la solicitud. ¡Inténtalo de nuevo!"

async def safe_way(text: str):

await session.interrupt()

session._update_agent_state("listening")

session.say(text, allow_interruptions=True)

asyncio.create_task(safe_way(text_to_speak))

Thanks — I’m responding to your question about stopping/interrupting the AgentSession speech from your Python server when you press the speak button on your React client. I looked at your handler and the pattern you’re using to call

session.say(...)

. Short answer • Use

session.interrupt()

to stop the current speech and any in-progress

say

call. • Keep and manage the asyncio Task/handle returned by

session.say(...)

so you can cancel it directly if needed. • Make sure your React client sends a clear interrupt message (e.g., "interrupt" or a JSON action) that your

custom_text_input_handler_tts

recognizes and uses to call

session.interrupt()

or cancel the task. What to change in your code (practical steps) 1) Make the

say

call cancellable and keep the Task/handle on the session or ctx so other code can access it: • When you call

session.say(...)

, capture the returned future/handle. • Store that handle in

session.userdata

or

ctx._session_say_handle

so the interrupt path can find it. 2) On client speak/interrupt event, either: • Send a special text input/command from React that your text handler interprets (e.g., "ACTION::interrupt" or "CMD::stop"), or • Send a message with a field that differentiates normal text vs. interrupt. 3) In the handler, call

session.interrupt()

and cancel the stored handle if present: •

await session.interrupt()

will stop playback and allow new audio to begin. • If you created the say Task yourself (via

asyncio.create_task(session.say(...))

), cancel that task as well. Concrete example (adapt to your style). Key ideas: store the handle, call

session.interrupt()

, cancel task if needed.

# When starting speech, store the handle
async def start_speaking(session: AgentSession, text: str):
    # session.say returns a future-like handle; keep reference so we can cancel later
    handle = session.say(text, allow_interruptions=True)
    # Save so other parts (or interrupt) can access it
    session.userdata["_current_say_handle"] = handle
    try:
        await handle
    except asyncio.CancelledError:
        # handle canceled externally
        pass
    finally:
        session.userdata.pop("_current_say_handle", None)

# To interrupt (from your input handler when receiving an interrupt command)
async def interrupt_speech(session: AgentSession):
    # Preferred: call interrupt to stop audio pipeline
    try:
        await session.interrupt()
    except Exception:
        # continue to task-cancel fallback
        pass

    # If we started say as a Task, cancel it
    handle = session.userdata.get("_current_say_handle")
    if handle and not getattr(handle, "done", lambda: False)():
        try:
            handle.cancel()
            # if handle is awaitable, optionally await to let it finish cancelling
            await asyncio.sleep(0)  # yield so cancel can propagate
        except Exception:
            pass
    # Update agent state if you use it
    session._update_agent_state("listening")

Where to call interrupt based on React input • If your React speak button triggers sending the text to the server, add a separate control or payload to indicate "stop" (for example send

::STOP

or

{ action: "stop" }

). • In

custom_text_input_handler_tts

, detect that payload and call

await interrupt_speech(session)

immediately rather than starting a new TTS. Example logic in your `custom_text_input_handler_tts`: • If payload is an interrupt command: call

await session.interrupt()

and cancel the current handle; return. • Otherwise: proceed to enqueue/play the waiting audio and then call

asyncio.create_task(start_speaking(...))

. Other tips and gotchas • Use

allow_interruptions=True

in

session.say()

so the pipeline permits interruptions. • If you use

asyncio.create_task(session.say(...))

rather than awaiting it directly, keep the Task reference so you can cancel it later. •

session.interrupt()

is the recommended method to stop backend audio generation/playback; canceling the Task is a fallback if needed. • Ensure the interrupt command from the client arrives on the same room/session context so you call interrupt on the correct

AgentSession

. If you want, I can: • Insert these changes directly into the exact handler you posted (with correct indentation and where to detect the interrupt from your payload format), or • Provide a short React example showing how to send a "stop" command to your server. Which would you prefer? , with interactive elements