Agents Py

LiveKit Agents Python SDK

Build voice AI agents with LiveKit's Python Agents SDK.

LiveKit MCP server tools

This skill works alongside the LiveKit MCP server, which provides direct access to the latest LiveKit documentation, code examples, and changelogs. Use these tools when you need up-to-date information that may have changed since this skill was created.

Available MCP tools:

• docs_search - Search the LiveKit docs site
• get_pages - Fetch specific documentation pages by path
• get_changelog - Get recent releases and updates for LiveKit packages
• code_search - Search LiveKit repositories for code examples
• get_python_agent_example - Browse 100+ Python agent examples

When to use MCP tools:

• You need the latest API documentation or feature updates
• You're looking for recent examples or code patterns
• You want to check if a feature has been added in recent releases
• The local references don't cover a specific topic

When to use local references:

• You need quick access to core concepts covered in this skill
• You're working offline or want faster access to common patterns
• The information in the references is sufficient for your needs

Use MCP tools and local references together for the best experience.

References

Consult these resources as needed:

• ./references/livekit-overview.md -- LiveKit ecosystem overview and how these skills work together
• ./references/agent-session.md -- AgentSession lifecycle, events, and configuration
• ./references/tools.md -- Function tools, RunContext, and tool results
• ./references/models.md -- STT, LLM, TTS model strings and plugin configuration
• ./references/workflows.md -- Multi-agent handoffs, Tasks, TaskGroups, and pipeline nodes

Installation

uv add "livekit-agents[silero,turn-detector]~=1.3" \
  "livekit-plugins-noise-cancellation~=0.2" \
  "python-dotenv"

Environment variables

Use the LiveKit CLI to load your credentials into a .env.local file:

lk app env -w

Or manually create a .env.local file:

LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret
LIVEKIT_URL=wss://your-project.livekit.cloud

Quick start

Basic agent with STT-LLM-TTS pipeline

from dotenv import load_dotenv
from livekit import agents, rtc
from livekit.agents import AgentSession, Agent, AgentServer, room_io
from livekit.plugins import noise_cancellation, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

load_dotenv(".env.local")

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            Keep responses concise, 1-3 sentences. No markdown or emojis.""",
        )

server = AgentServer()

@server.rtc_session()
async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        stt="assemblyai/universal-streaming:en",
        llm="openai/gpt-4.1-mini",
        tts="cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
                    if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                    else noise_cancellation.BVC(),
            ),
        ),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )

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

Basic agent with realtime model

from dotenv import load_dotenv
from livekit import agents, rtc
from livekit.agents import AgentSession, Agent, AgentServer, room_io
from livekit.plugins import openai, noise_cancellation

load_dotenv(".env.local")

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful voice AI assistant."
        )

server = AgentServer()

@server.rtc_session()
async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        llm=openai.realtime.RealtimeModel(voice="coral")
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
                    if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                    else noise_cancellation.BVC(),
            ),
        ),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )

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

Core concepts

Agent class

Define agent behavior by subclassing Agent:

from livekit.agents import Agent, function_tool

class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="Your system prompt here",
        )

    async def on_enter(self) -> None:
        """Called when agent becomes active."""
        await self.session.generate_reply(
            instructions="Greet the user"
        )

    async def on_exit(self) -> None:
        """Called before agent hands off to another agent."""
        pass

    @function_tool()
    async def my_tool(self, param: str) -> str:
        """Tool description for the LLM."""
        return f"Result: {param}"

AgentSession

The session orchestrates the voice pipeline:

session = AgentSession(
    stt="assemblyai/universal-streaming:en",
    llm="openai/gpt-4.1-mini",
    tts="cartesia/sonic-3:voice_id",
    vad=silero.VAD.load(),
    turn_detection=MultilingualModel(),
)

Key methods:

• session.start(room, agent) - Start the session
• session.say(text) - Speak text directly
• session.generate_reply(instructions) - Generate LLM response
• session.interrupt() - Stop current speech
• session.update_agent(new_agent) - Switch to different agent

Function tools

Use the @function_tool decorator:

from livekit.agents import function_tool, RunContext

@function_tool()
async def get_weather(self, context: RunContext, location: str) -> str:
    """Get the current weather for a location."""
    return f"Weather in {location}: Sunny, 72°F"

Running the agent

# Development mode with auto-reload
uv run agent.py dev

# Console mode (local testing)
uv run agent.py console

# Production mode
uv run agent.py start

# Download required model files
uv run agent.py download-files

LiveKit Inference model strings

Use model strings for simple configuration without API keys:

STT (Speech-to-Text):

• "assemblyai/universal-streaming:en" - AssemblyAI streaming
• "deepgram/nova-3:en" - Deepgram Nova
• "cartesia/ink" - Cartesia STT

LLM (Large Language Model):

• "openai/gpt-4.1-mini" - GPT-4.1 mini (recommended)
• "openai/gpt-4.1" - GPT-4.1
• "openai/gpt-5" - GPT-5
• "gemini/gemini-3-flash" - Gemini 3 Flash
• "gemini/gemini-2.5-flash" - Gemini 2.5 Flash

TTS (Text-to-Speech):

• "cartesia/sonic-3:{voice_id}" - Cartesia Sonic 3
• "elevenlabs/eleven_turbo_v2_5:{voice_id}" - ElevenLabs
• "deepgram/aura:{voice}" - Deepgram Aura

Best practices

1. Always use LiveKit Inference model strings as the default for STT, LLM, and TTS. This eliminates the need to manage individual provider API keys. Only use plugins when you specifically need custom models, voice cloning, Anthropic Claude, or self-hosted models. 2. Use adaptive noise cancellation with a lambda to detect SIP participants and apply appropriate noise cancellation (BVCTelephony for phone calls, BVC for standard participants). 3. Use MultilingualModel turn detection for natural conversation flow. 4. Structure prompts with Identity, Output rules, Tools, Goals, and Guardrails sections. 5. Test with console mode before deploying to LiveKit Cloud. 6. Use `lk app env -w` to load LiveKit Cloud credentials into your environment.

AgentSession reference

The AgentSession is the main orchestrator for your voice AI app.

Constructor options

from livekit.agents import AgentSession
from livekit.plugins import silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

session = AgentSession(
    # Models (use inference strings or plugin instances)
    stt="assemblyai/universal-streaming:en",
    llm="openai/gpt-4.1-mini",
    tts="cartesia/sonic-3:voice_id",
    
    # Voice activity detection
    vad=silero.VAD.load(),
    
    # Turn detection
    turn_detection=MultilingualModel(),  # or "vad", "stt", "manual"
    
    # Voice options
    allow_interruptions=True,
    min_interruption_duration=0.5,
    min_interruption_words=0,
    min_endpointing_delay=0.5,
    max_endpointing_delay=3.0,
    
    # User data
    userdata={"key": "value"},
)

Starting the session

from livekit import rtc
from livekit.agents import room_io
from livekit.plugins import noise_cancellation

await session.start(
    room=ctx.room,
    agent=my_agent,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(
            # Use adaptive noise cancellation based on participant type
            noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
                if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                else noise_cancellation.BVC(),
        ),
    ),
)

Key methods

Generate speech

# Generate LLM response
handle = session.generate_reply(
    instructions="Greet the user warmly",
    user_input="Hello!",  # Optional user message
    allow_interruptions=True,
)
await handle.wait_for_playout()

# Speak text directly
handle = session.say(
    "Hello! How can I help you today?",
    allow_interruptions=True,
)
await handle.wait_for_playout()

Interrupt and control

# Stop current speech
session.interrupt()

# Commit user turn manually (when turn_detection="manual")
session.commit_user_turn()

# Clear user turn
session.clear_user_turn()

Switch agents

# Switch to a different agent
session.update_agent(new_agent)

Access state

# Chat context
chat_ctx = session.chat_ctx

# Current agent state
state = session.agent_state  # "initializing", "listening", "thinking", "speaking"

# User data
data = session.userdata

Events

from livekit.agents import (
    UserStateChangedEvent,
    AgentStateChangedEvent,
    ConversationItemAddedEvent,
    MetricsCollectedEvent,
)

@session.on("user_state_changed")
def on_user_state_changed(ev: UserStateChangedEvent):
    # ev.new_state: "speaking", "listening", "away"
    print(f"User state: {ev.new_state}")

@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
    # ev.new_state: "initializing", "listening", "thinking", "speaking"
    print(f"Agent state: {ev.new_state}")

@session.on("conversation_item_added")
def on_conversation_item_added(ev: ConversationItemAddedEvent):
    print(f"New message: {ev.item}")

@session.on("metrics_collected")
def on_metrics_collected(ev: MetricsCollectedEvent):
    print(f"Metrics: {ev.metrics}")

@session.on("user_input_transcribed")
def on_user_input_transcribed(ev):
    print(f"User said: {ev.transcript}")

Turn detection modes

# Recommended: Turn detector model
from livekit.plugins.turn_detector.multilingual import MultilingualModel
session = AgentSession(
    turn_detection=MultilingualModel(),
    vad=silero.VAD.load(),
)

# VAD only
session = AgentSession(
    turn_detection="vad",
    vad=silero.VAD.load(),
)

# STT endpointing
session = AgentSession(
    turn_detection="stt",
    stt="assemblyai/universal-streaming:en",
    vad=silero.VAD.load(),
)

# Manual control
session = AgentSession(
    turn_detection="manual",
)

Voice options

Closing the session

# Graceful close
await session.close()

# Shutdown with options
session.shutdown(drain=True, reason="user_initiated")

LiveKit overview

LiveKit is a realtime communication platform for building AI-native applications with audio, video, and data streaming. This overview helps you understand the LiveKit ecosystem and how to use these skills effectively.

Platform components

LiveKit Cloud

LiveKit Cloud is a fully managed platform for building, deploying, and operating AI agent applications. It includes:

• Realtime media infrastructure - Global mesh of servers for low-latency audio, video, and data streaming
• Managed agent hosting - Deploy agents without managing servers or orchestration
• LiveKit Inference - Run AI models directly within LiveKit Cloud without API keys
• Native telephony - Provision phone numbers and connect PSTN calls directly to rooms
• Observability - Built-in analytics, logs, and quality metrics

Agents framework

The Agents framework lets you build Python or Node.js programs that join LiveKit rooms as realtime participants. Key capabilities:

• Voice pipelines - Stream audio through STT-LLM-TTS pipelines
• Realtime models - Use models like OpenAI Realtime API that handle speech directly
• Tool calling - Define functions the LLM can invoke during conversations
• Multi-agent workflows - Hand off between specialized agents
• Turn detection - State-of-the-art model for natural conversation flow

Architecture

┌─────────────┐     WebRTC      ┌─────────────┐     HTTP/WS      ┌─────────────┐
│   Frontend  │ ◄─────────────► │   LiveKit   │ ◄──────────────► │    Agent    │
│  (Web/App)  │                 │    Room     │                  │   Server    │
└─────────────┘                 └─────────────┘                  └─────────────┘
                                       │                                │
                                       │                                │
                                       ▼                                ▼
                                ┌─────────────┐                  ┌─────────────┐
                                │  Telephony  │                  │  AI Models  │
                                │    (SIP)    │                  │ (STT/LLM/TTS)│
                                └─────────────┘                  └─────────────┘

How these skills work together

The LiveKit skills cover the full stack for building voice AI applications:

Typical workflow:

1. Choose your backend - Use agents-py or agents-ts based on your team's preference 2. Build the frontend - Use agents-ui for React-based web interfaces 3. Connect via LiveKit - Both connect to the same LiveKit room for realtime communication

Using the skills effectively

When to use each skill

• Building a new voice agent? Start with agents-py or agents-ts for the backend logic
• Need a web interface? Add agents-ui for pre-built React components
• Full-stack project? Use both a backend skill and agents-ui together

Combining skills

The skills are designed to work together. A typical project structure:

my-voice-app/
├── agent/           # Use agents-py or agents-ts skill
│   └── agent.py     # or agent.ts
├── frontend/        # Use agents-ui skill
│   └── src/
│       └── app/
└── .env.local       # Shared LiveKit credentials

Environment setup

All skills require LiveKit credentials:

LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret
LIVEKIT_URL=wss://your-project.livekit.cloud

Get these from your LiveKit Cloud dashboard or self-hosted deployment.

Resources

• LiveKit documentation
• Agents framework guide
• LiveKit Cloud
• GitHub repositories

Models reference

LiveKit Inference is the recommended way to use AI models with LiveKit Agents. It provides access to leading models without managing individual provider API keys. LiveKit Cloud handles authentication, billing, and optimal provider selection automatically.

LiveKit Inference (recommended)

Use model strings to configure STT, LLM, and TTS in your AgentSession.

STT (speech-to-text)

session = AgentSession(
    stt="deepgram/nova-3:en",
)

Automatic model selection: Use "auto:language" to let LiveKit choose the best STT model for a language:

session = AgentSession(
    stt="auto:en",  # Best available English STT
    stt="auto:es",  # Best available Spanish STT
)

LLM (large language model)

session = AgentSession(
    llm="openai/gpt-4.1-mini",
)

TTS (text-to-speech)

session = AgentSession(
    tts="cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
)

Popular voices:

Realtime models

For speech-to-speech without separate STT/TTS pipelines:

OpenAI Realtime

from livekit.plugins import openai

session = AgentSession(
    llm=openai.realtime.RealtimeModel(
        voice="coral",
        model="gpt-4o-realtime-preview",
    ),
)

Gemini Live

from livekit.plugins import google

session = AgentSession(
    llm=google.realtime.RealtimeModel(
        voice="Puck",
    ),
)

xAI Grok

from livekit.plugins import xai

session = AgentSession(
    llm=xai.realtime.RealtimeModel(
        voice="aurora",
    ),
)

Advanced configuration

Use the inference module when you need additional parameters while still using LiveKit Inference:

from livekit.agents import AgentSession, inference

session = AgentSession(
    llm=inference.LLM(
        model="openai/gpt-5-mini",
        provider="openai",
        extra_kwargs={"reasoning_effort": "low"}
    ),
    stt=inference.STT(
        model="deepgram/nova-3",
        language="en",
    ),
    tts=inference.TTS(
        model="cartesia/sonic-3",
        voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
        language="en",
        extra_kwargs={"speed": 1.2, "emotion": "cheerful"}
    ),
)

VAD and turn detection

These components are configured separately from model providers:

from livekit.plugins import silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

session = AgentSession(
    vad=silero.VAD.load(),
    turn_detection=MultilingualModel(),  # Recommended
)

Turn detection options:

• MultilingualModel() - Recommended for natural conversation flow
• "vad" - VAD-only turn detection
• "stt" - STT endpointing (works with Deepgram Flux)
• "manual" - Manual control with session.commit_user_turn()

Noise cancellation

from livekit.plugins import noise_cancellation
from livekit.agents import room_io

await session.start(
    room=ctx.room,
    agent=agent,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
        ),
    ),
)

---

Using plugins (when needed)

Use plugins directly only when you need features not available in LiveKit Inference:

• Custom or fine-tuned models not available in LiveKit Inference
• Voice cloning with your own provider account
• Anthropic Claude models (not available in LiveKit Inference)
• Self-hosted models via Ollama
• Provider-specific features not exposed through inference module

Anthropic

from livekit.plugins import anthropic

session = AgentSession(
    llm=anthropic.LLM(model="claude-sonnet-4-20250514"),
)

Requires: ANTHROPIC_API_KEY

OpenAI (direct)

from livekit.plugins import openai

session = AgentSession(
    llm=openai.LLM(model="gpt-4o"),
    stt=openai.STT(),
    tts=openai.TTS(voice="alloy"),
)

Requires: OPENAI_API_KEY

Ollama (self-hosted)

from livekit.plugins import ollama

session = AgentSession(
    llm=ollama.LLM(model="llama3.2"),
)

Other plugins

Additional plugins are available for: AWS Bedrock, Azure, Baseten, Cerebras, Deepgram, ElevenLabs, Fireworks, Google Cloud, Groq, Mistral AI, and more. Each requires its own API key and account setup.

See the LiveKit Agents documentation for the full list.

Function tools reference

Function tools let your agent call external functions during conversations.

Basic function tool

from livekit.agents import Agent, function_tool, RunContext

class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful assistant.")

    @function_tool()
    async def get_weather(self, context: RunContext, location: str) -> str:
        """Get the current weather for a location.
        
        Args:
            location: The city name to get weather for
        """
        # Your implementation here
        return f"The weather in {location} is sunny and 72°F"

RunContext

Access session data and perform actions within tools:

from livekit.agents import function_tool, RunContext

@function_tool()
async def save_note(self, context: RunContext, note: str) -> str:
    """Save a note for the user."""
    # Access user data
    context.userdata["notes"] = context.userdata.get("notes", [])
    context.userdata["notes"].append(note)
    
    # Access the session
    session = context.session
    
    # Access the room
    room = context.session.room
    
    return "Note saved!"

Tool with complex parameters

from typing import Literal
from livekit.agents import function_tool, RunContext

@function_tool()
async def book_appointment(
    self,
    context: RunContext,
    date: str,
    time: str,
    service: Literal["haircut", "coloring", "styling"],
    notes: str = "",
) -> str:
    """Book an appointment.
    
    Args:
        date: The date in YYYY-MM-DD format
        time: The time in HH:MM format
        service: Type of service requested
        notes: Optional additional notes
    """
    return f"Booked {service} for {date} at {time}"

Tool returning an Agent (handoff)

Return an Agent instance to hand off control. You can also return a tuple with the agent and a message for the LLM:

from livekit.agents import function_tool, RunContext, Agent

@function_tool()
async def transfer_to_billing(self, context: RunContext) -> Agent:
    """Transfer the call to the billing department."""
    await self.session.say("I'll transfer you to our billing team.")
    return BillingAgent()

# Or return with a message for the LLM
@function_tool()
async def transfer_to_sales(self, context: RunContext) -> tuple[Agent, str]:
    """Transfer the call to the sales department."""
    return SalesAgent(), "Transferring the user to SalesAgent"

Tool with speech during execution

from livekit.agents import function_tool, RunContext

@function_tool()
async def long_running_task(self, context: RunContext, query: str) -> str:
    """Perform a long-running search."""
    # Speak while processing
    await self.session.say("Let me look that up for you...")
    
    # Do the work
    result = await search_database(query)
    
    return result

Provider tools

Use tools specific to model providers:

from livekit.plugins.google import GeminiFileSearch

class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful assistant.",
            tools=[
                GeminiFileSearch(corpus_name="my-corpus"),
            ],
        )

Standalone tool definitions

Define tools outside of an agent class:

from livekit.agents import Agent, function_tool, RunContext

@function_tool()
async def calculate_tip(context: RunContext, amount: float, percentage: float = 18.0) -> str:
    """Calculate the tip for a bill.
    
    Args:
        amount: The bill amount
        percentage: Tip percentage (default 18%)
    """
    tip = amount * (percentage / 100)
    return f"Tip: ${tip:.2f}, Total: ${amount + tip:.2f}"

class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful assistant.",
            tools=[calculate_tip],
        )

Tool interruptions

Handle interruptions during long-running tools:

import asyncio
from livekit.agents import function_tool, RunContext

@function_tool()
async def search_database(self, context: RunContext, query: str) -> str:
    """Search the database."""
    # For non-interruptible tools, call this at the start:
    # context.disallow_interruptions()
    
    wait_for_result = asyncio.ensure_future(perform_search(query))
    await context.speech_handle.wait_if_not_interrupted([wait_for_result])
    
    if context.speech_handle.interrupted:
        # Tool was interrupted, clean up
        wait_for_result.cancel()
        return None  # Return value is ignored when interrupted
    
    return wait_for_result.result()

Error handling

Use ToolError to return errors to the LLM:

from livekit.agents import function_tool, RunContext, ToolError

@function_tool()
async def lookup_weather(self, context: RunContext, location: str) -> str:
    """Look up weather for a location."""
    if location == "mars":
        raise ToolError("This location is not supported yet.")
    return f"Weather in {location}: Sunny, 72°F"

Best practices

1. Write clear docstrings - The LLM uses them to understand when to call the tool 2. Use type hints - They define the parameter schema for the LLM 3. Return strings - Results are added to the conversation context 4. Handle errors gracefully - Return error messages the LLM can understand 5. Keep tools focused - One tool should do one thing well

Workflows reference

Build complex voice AI applications with multi-agent handoffs, tasks, and pipeline customization.

Multi-agent handoffs

Switch between agents during a conversation:

from livekit.agents import Agent, function_tool

class TriageAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a triage agent. Route users to the right department."
        )

    @function_tool()
    async def transfer_to_sales(self) -> Agent:
        """Transfer to the sales department."""
        await self.session.say("I'll connect you with our sales team.")
        return SalesAgent()

    @function_tool()
    async def transfer_to_support(self) -> Agent:
        """Transfer to technical support."""
        await self.session.say("Let me connect you with support.")
        return SupportAgent()

class SalesAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a sales representative."
        )

class SupportAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a technical support specialist."
        )

Manual agent switching

# Switch agent programmatically
session.update_agent(new_agent)

Preserving context during handoffs

class BaseAgent(Agent):
    async def on_enter(self) -> None:
        # Access previous chat context
        chat_ctx = self.chat_ctx.copy()
        
        # Add context from previous agent
        if self.session.userdata.get("prev_agent"):
            prev_items = self.session.userdata["prev_agent"].chat_ctx.items
            chat_ctx.items.extend(prev_items[-6:])  # Keep last 6 messages
        
        await self.update_chat_ctx(chat_ctx)

Tasks

Tasks are focused units that perform a specific objective and return a typed result.

Defining a task

from livekit.agents import AgentTask, function_tool

class CollectEmailTask(AgentTask[str]):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="Collect and validate the user's email address.",
            chat_ctx=chat_ctx,
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="Ask the user for their email address."
        )

    @function_tool()
    async def confirm_email(self, email: str) -> None:
        """Confirm the user's email address."""
        self.complete(email)

Running a task

class MyAgent(Agent):
    async def on_enter(self) -> None:
        # Run task and get result
        email = await CollectEmailTask(chat_ctx=self.chat_ctx)
        
        # Use the result
        self.session.userdata["email"] = email
        
        await self.session.generate_reply(
            instructions=f"Thank the user and confirm their email: {email}"
        )

Task with dataclass result

from dataclasses import dataclass

@dataclass
class ContactInfo:
    name: str
    email: str
    phone: str

class CollectContactTask(AgentTask[ContactInfo]):
    def __init__(self):
        super().__init__(
            instructions="Collect the user's contact information."
        )
        self._data = {}

    @function_tool()
    async def record_name(self, name: str) -> None:
        """Record the user's name."""
        self._data["name"] = name
        self._check_complete()

    @function_tool()
    async def record_email(self, email: str) -> None:
        """Record the user's email."""
        self._data["email"] = email
        self._check_complete()

    @function_tool()
    async def record_phone(self, phone: str) -> None:
        """Record the user's phone number."""
        self._data["phone"] = phone
        self._check_complete()

    def _check_complete(self):
        if all(k in self._data for k in ["name", "email", "phone"]):
            self.complete(ContactInfo(**self._data))

TaskGroups

Execute ordered sequences of tasks with regression support.

from livekit.agents.beta.workflows import TaskGroup, GetEmailTask

# Create task group
task_group = TaskGroup()

# Add tasks in order
task_group.add(
    lambda: CollectNameTask(),
    id="collect_name",
    description="Collects the user's name"
)
task_group.add(
    lambda: GetEmailTask(),
    id="collect_email", 
    description="Collects the user's email"
)
task_group.add(
    lambda: ConfirmTask(),
    id="confirm",
    description="Confirms the collected information"
)

# Execute and get results
results = await task_group
print(results.task_results)
# {"collect_name": "John", "collect_email": GetEmailResult(...), ...}

Prebuilt tasks

from livekit.agents.beta.workflows import GetEmailTask, GetAddressTask, GetDtmfTask

# Collect email
email_result = await GetEmailTask(chat_ctx=self.chat_ctx)
print(email_result.email_address)

# Collect address
address_result = await GetAddressTask(chat_ctx=self.chat_ctx)
print(address_result.address)

# Collect DTMF input (for telephony)
dtmf_result = await GetDtmfTask(
    num_digits=10,
    chat_ctx=self.chat_ctx,
    ask_for_confirmation=True,
)
print(dtmf_result.user_input)

Pipeline nodes

Customize the voice pipeline by overriding nodes in your Agent class.

STT node

class MyAgent(Agent):
    async def stt_node(self, audio, model_settings):
        """Customize speech-to-text processing."""
        # Pre-process audio
        async for event in Agent.default.stt_node(self, audio, model_settings):
            # Post-process transcription
            yield event

LLM node

class MyAgent(Agent):
    async def llm_node(self, chat_ctx, tools, model_settings):
        """Customize LLM inference."""
        # Modify chat context before inference
        async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
            # Filter or modify output
            yield chunk

TTS node

class MyAgent(Agent):
    async def tts_node(self, text, model_settings):
        """Customize text-to-speech."""
        # Pre-process text (e.g., pronunciation fixes)
        async def modified_text():
            async for t in text:
                yield t.replace("LiveKit", "Live Kit")
        
        async for frame in Agent.default.tts_node(self, modified_text(), model_settings):
            yield frame

Transcription node

class MyAgent(Agent):
    async def transcription_node(self, text, model_settings):
        """Customize transcription output."""
        async for delta in text:
            # Remove unwanted characters
            yield delta.replace("😘", "")

Lifecycle hooks

class MyAgent(Agent):
    async def on_enter(self) -> None:
        """Called when agent becomes active."""
        await self.session.generate_reply(
            instructions="Greet the user"
        )

    async def on_exit(self) -> None:
        """Called before handoff to another agent."""
        await self.session.say("Transferring you now...")

    async def on_user_turn_completed(self, turn_ctx, new_message) -> None:
        """Called when user finishes speaking, before agent responds."""
        # Inject RAG context
        rag_content = await my_rag_lookup(new_message.text_content())
        turn_ctx.add_message(role="assistant", content=rag_content)

Best practices

1. Use tasks for structured data collection - They provide typed results and clear completion criteria 2. Preserve context during handoffs - Copy relevant chat history to the new agent 3. Keep agents focused - Each agent should have a clear responsibility 4. Use lifecycle hooks - on_enter and on_exit for proper setup and cleanup 5. Test agent flows - Use the testing framework to verify handoff behavior