Appendix B: LiveKit Agents API Quick Reference

A condensed reference for the LiveKit Agents framework classes and methods used throughout this book. For the full API surface, consult the official LiveKit documentation. What follows are the parts you will use daily.

Agent

The base class for all voice agents. Defines the agent’s instructions, tools, and behavior.

from livekit.agents import Agent

class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="Your system prompt here.",
            tools=[standalone_tool],          # Optional: standalone function tools
        )

Key methods:

Properties:

AgentTask

A focused sub-conversation that produces a typed result. Extends AgentTask[T] where T is the result dataclass.

from livekit.agents import AgentTask, function_tool
from livekit.agents.voice.events import RunContext
from dataclasses import dataclass

@dataclass
class NameResult:
    first_name: str
    last_name: str

class CollectNameTask(AgentTask[NameResult]):
    def __init__(self):
        super().__init__(
            instructions="Collect the caller's first and last name."
        )

    @function_tool()
    async def record_name(
        self, ctx: RunContext, first_name: str, last_name: str
    ) -> str:
        """Record the caller's name."""
        self.complete(NameResult(first_name=first_name, last_name=last_name))
        return f"Recorded: {first_name} {last_name}"

Key methods:

AgentSession

The runtime that orchestrates the STT-LLM-TTS pipeline. Created once per call.

from livekit.agents import AgentSession

session = AgentSession(
    stt=deepgram.STT(model="nova-3"),
    llm=openai.LLM(model="gpt-4.1-nano"),
    tts=cartesia.TTS(model="sonic"),
    vad=ctx.proc.userdata["vad"],
    turn_detection=turn_detector.MultilingualModel(),
    min_endpointing_delay=0.5,   # AgentSession-level endpointing
    max_endpointing_delay=5.0,
)

await session.start(agent=MyAgent(), room=ctx.room)

Key methods:

TaskGroup

Run multiple tasks in sequence, collecting a result from each.

from livekit.agents.beta.workflows import TaskGroup

group = TaskGroup()
group.add(lambda: CollectNameTask(), id="name", description="Caller's name")
group.add(lambda: CollectPhoneTask(), id="phone", description="Callback number")
group.add(lambda: CollectServicesTask(), id="services", description="Requested services")

results = await group
# results.task_results is a dict keyed by the id you gave each task
task_results = results.task_results  # task_results["name"] -> CollectNameTask result

Key methods:

Configuration

Turn Detection

Controls when the agent decides the user has finished speaking.

from livekit.plugins import turn_detector

# The turn detector model itself takes no timing arguments.
turn_detection = turn_detector.MultilingualModel(
    unlikely_threshold=0.15,     # Optional: override the end-of-turn threshold
)

# Endpointing delays are AgentSession parameters, not model parameters:
session = AgentSession(
    turn_detection=turn_detection,
    min_endpointing_delay=0.5,   # Minimum silence before endpoint (seconds)
    max_endpointing_delay=5.0,   # Maximum wait before forcing endpoint
)

Interruption Handling

Controls whether the caller can interrupt the agent mid-speech.

session = AgentSession(
    allow_interruptions=True,    # Default: True
    min_interruption_duration=0.5,  # Minimum speech duration to trigger interrupt
)

Events

Subscribe to events on the AgentSession to react to pipeline state changes.

Handlers receive an event object, so give them a single parameter and read its fields.

@session.on("user_state_changed")
def on_user_state(ev):
    """User speech state changed. ev.old_state / ev.new_state are one of
    'speaking', 'listening', 'away'."""
    pass

@session.on("agent_state_changed")
def on_agent_state(ev):
    """Agent state changed. ev.old_state / ev.new_state are one of
    'initializing', 'idle', 'listening', 'thinking', 'speaking'."""
    pass

@session.on("user_input_transcribed")
def on_transcribed(ev):
    """STT output. ev.transcript holds the text; ev.is_final marks the
    final transcript for the turn."""
    pass

@session.on("function_tools_executed")
def on_tools_executed(ev):
    """A batch of tool calls finished. Iterate ev.zipped() to pair each
    FunctionCall with its output."""
    pass

@session.on("close")
def on_close(ev):
    """Session ended. Clean up resources."""
    pass

function_tool Decorator

Registers a method or standalone function as a tool the LLM can call.

from livekit.agents import function_tool
from livekit.agents.voice.events import RunContext

@function_tool()
async def my_tool(
    context: RunContext,
    required_arg: str,
    optional_arg: int = 10,
) -> str:
    """Tool description that the LLM reads.

    Args:
        required_arg: Description of this argument.
        optional_arg: Description with default value.
    """
    return "result string"

Parameters for @function_tool():

Rules:

• The function’s docstring becomes the tool description the LLM uses to decide when to call it.
• Type hints on parameters become the JSON schema the LLM uses for arguments.
• Always return a string. The return value is injected into the chat context as the tool result.
• Use RunContext as the first parameter to access session state and room metadata.