Uninterruptable Agent | LiveKit Documentation

This example configures an agent to finish speaking even if the user talks over it by disabling interruptions. The agent also seeds the first user input so you can test the behavior immediately.

Prerequisites

Add a .env.local in this directory with your LiveKit credentials:

LIVEKIT_URL=your_livekit_url
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret

Install dependencies:

pip install "livekit-agents" python-dotenv

Load configuration and create the AgentServer

Load environment variables so the audio plugins can authenticate. Create an AgentServer to manage sessions.

from dotenv import load_dotenv
from livekit.agents import JobContext, AgentServer, cli, Agent, AgentSession, inference

load_dotenv(".env.local")

server = AgentServer()

Create a non-interruptable agent

Set allow_interruptions=False when constructing the agent. The agent class is lightweight — only instructions and the interruption setting are defined here.

class UninterruptableAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""
                You are a helpful assistant communicating through voice who is not interruptable.
            """,
            allow_interruptions=False
        )

    async def on_enter(self):
        self.session.generate_reply(user_input="Say something somewhat long and boring so I can test if you're interruptable.")

Create the RTC session entrypoint

Create an AgentSession with STT/LLM/TTS configured, start the session with the agent, and connect to the room.

@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    ctx.log_context_fields = {"room": ctx.room.name}

    session = AgentSession(
        stt=inference.STT(model="deepgram/nova-3-general"),
        llm=inference.LLM(model="google/gemma-4-31b-it"),

        tts=inference.TTS(model="inworld/inworld-tts-2", voice="Ashley"),

        preemptive_generation=True,
    )

    await session.start(agent=UninterruptableAgent(), room=ctx.room)
    await ctx.connect()

Run it

lk agent console uninterruptable.py

How it works

allow_interruptions=False keeps TTS playback intact even if new speech arrives.
on_enter seeds a first prompt so you can test the behavior without speaking first.
The rest of the media pipeline remains unchanged from a standard agent.
This setting is useful when you want to ensure an announcement completes before listening again.

Full example

from dotenv import load_dotenv
from livekit.agents import JobContext, AgentServer, cli, Agent, AgentSession, inference

load_dotenv(".env.local")

class UninterruptableAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""
                You are a helpful assistant communicating through voice who is not interruptable.
            """,
            allow_interruptions=False
        )

    async def on_enter(self):
        self.session.generate_reply(user_input="Say something somewhat long and boring so I can test if you're interruptable.")

server = AgentServer()

@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    ctx.log_context_fields = {"room": ctx.room.name}

    session = AgentSession(
        stt=inference.STT(model="deepgram/nova-3-general"),
        llm=inference.LLM(model="google/gemma-4-31b-it"),

        tts=inference.TTS(model="inworld/inworld-tts-2", voice="Ashley"),

        preemptive_generation=True,
    )

    await session.start(agent=UninterruptableAgent(), room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    cli.run_app(server)