Thanks — I understand you want a concise example using the LiveKit Agents Python SDK that ends a session when a time limit is reached. Below is a simple, ready-to-use example that: • Starts an agent session. • Launches a background timeout task that waits the configured number of seconds. • Sends a polite closing message, waits for playout, then shuts down the session. • Cancels the timeout if the session ends before the timer. Replace the STT/LLM/TTS placeholders with your actual implementations.

import asyncio
import json
from livekit.agents import AgentSession, AgentServer, JobContext, Agent
# import your STT/LLM/TTS plugins, e.g.:
# from livekit.plugins import deepgram, openai, cartesia, silero

server = AgentServer()

# Default timeout in seconds (change as needed)
DEFAULT_SESSION_TIMEOUT = 300  # 5 minutes

@server.rtc_session(agent_name="time_limited_agent")
async def entrypoint(ctx: JobContext):
    """
    Entrypoint for a time-limited agent session.
    The timeout can be overridden via job metadata: {"session_timeout": 120}
    """
    # Read timeout from job metadata if present
    try:
        meta = json.loads(ctx.job.metadata) if ctx.job and ctx.job.metadata else {}
        session_timeout = int(meta.get("session_timeout", DEFAULT_SESSION_TIMEOUT))
    except (ValueError, TypeError, json.JSONDecodeError):
        session_timeout = DEFAULT_SESSION_TIMEOUT

    # Create your AgentSession with real plugins
    session = AgentSession(
        # Example placeholders — replace with your actual plugin instances
        # vad=None,
        # stt=deepgram.STT(),
        # llm=openai.LLM(model="gpt-4o"),
        # tts=cartesia.TTS()
    )

    async def timeout_monitor():
        try:
            await asyncio.sleep(session_timeout)
            print(f"Session time limit of {session_timeout} seconds reached. Ending session.")

            # Generate a polite farewell before shutting down
            await session.generate_reply(
                instructions="Inform the user the session time limit has been reached and finish politely."
            )

            # Wait until any TTS playout finishes
            await session.wait_for_playout()

            # Gracefully end the session
            session.shutdown()
        except asyncio.CancelledError:
            # Timeout cancelled because session ended earlier — nothing to do
            pass
        except Exception as e:
            # Log unexpected errors in the timeout handler
            print(f"Timeout handler error: {e}")

    # Start the timeout monitor task
    timeout_task = asyncio.create_task(timeout_monitor())

    # Ensure timeout is cancelled when session ends normally
    def on_shutdown():
        if not timeout_task.done():
            timeout_task.cancel()

    ctx.add_shutdown_callback(on_shutdown)

    # Start the AgentSession (example Agent instructions shown)
    await session.start(
        room=ctx.room,
        agent=Agent(instructions="You are a helpful assistant with a limited session duration.")
    )

    # Normal session work continues here until session.shutdown() is called or externally ended.

If you’d like, I can: • Show a version that forcibly disconnects participants instead of using

session.shutdown()

. • Provide a minimal example with specific STT/LLM/TTS plugin setup for your environment. Which would help you more? Sources: How to detect and handle silence during calls | LiveKit, with interactive elements