Comprehensive guide for building functional tools for LiveKit voice agents using the @function_tool decorator.
Build robust, production-ready tools for LiveKit voice agents that enable LLMs to interact with external services, manage state, coordinate multi-agent workflows, and handle real-time voice interactions.
from livekit.agents import Agent
from livekit.agents.llm import function_tool
from livekit.agents.voice import RunContext
class MyAgent(Agent):
def __init__(self):
super().__init__(
instructions="You are a helpful assistant...",
llm="openai/gpt-4o-mini",
)
@function_tool
async def get_weather(self, location: str) -> str:
"""Get current weather for a location.
Args:
location: City name or address to get weather for
"""
# Your implementation here
return f"Weather in {location}: Sunny, 72°F"
The @function_tool decorator automatically registers methods as callable tools for the LLM. The docstring is critical—it tells the LLM when and how to use the tool.
Critical for reliability: A good tool definition is key to reliable tool use from your LLM. Be specific about:
Use clear, action-oriented names that help the LLM discover the right tool:
@function_tool
async def search_flights(self, origin: str, destination: str, date: str) -> str:
"""Search for available flights between two cities on a specific date.
Use this when the user asks about flight availability, prices, or schedules.
Do NOT use this for booking—only for searching.
Args:
origin: Departure city or airport code
destination: Arrival city or airport code
date: Travel date in YYYY-MM-DD format
"""
The RunContext parameter provides access to session state, speech control, and user data:
@function_tool
async def save_preference(
self,
preference_name: str,
value: str,
context: RunContext
) -> str:
"""Save a user preference for later use.
Args:
preference_name: Name of the preference (e.g., "favorite_color")
value: The preference value
context: Runtime context (automatically provided)
"""
# Access shared state across tools
context.userdata[preference_name] = value
return f"Saved {preference_name} as {value}"
See Context & State Management for complete RunContext patterns.
Use type hints and annotations for rich parameter documentation:
from typing import Annotated, Literal
from pydantic import Field
from enum import Enum
class Priority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
@function_tool
async def create_task(
self,
title: str,
priority: Priority,
due_date: Annotated[str | None, Field(description="Due date in YYYY-MM-DD format")] = None
) -> str:
"""Create a new task with specified priority.
Args:
title: Brief description of the task
priority: Task priority level
due_date: Optional deadline for the task
"""
See Parameter Patterns for all documentation approaches.
Use for straightforward operations that execute quickly:
Use for external service calls:
See examples/api-integration-tool.py
Use for operations that might take time and should handle interruptions:
See Long-Running Functions and examples/long-running-tool.py
Use for maintaining context across interactions:
See Context & State Management and examples/stateful-tool.py
Use for agent handoffs and specialized routing:
See Multi-Agent Patterns and examples/agent-handoff-tool.py
Tools can be created at runtime for maximum flexibility:
from livekit.agents.llm import function_tool
# Option 1: Pass tools at agent creation
agent = MyAgent(
instructions="...",
tools=[
function_tool(
get_user_data,
name="get_user_data",
description="Fetch user information from database"
)
]
)
# Option 2: Update tools after creation
await agent.update_tools(
agent.tools + [
function_tool(
new_capability,
name="new_capability",
description="Dynamically added tool"
)
]
)
See Dynamic Tool Creation for complete patterns.
Testing is essential for reliable agents. LiveKit provides helpers that work with pytest:
import pytest
from livekit.agents.testing import VoiceAgentTestSession
@pytest.mark.asyncio
async def test_weather_tool():
async with VoiceAgentTestSession(agent=MyAgent()) as session:
response = await session.send_text("What's the weather in Tokyo?")
assert "Tokyo" in response.text
assert session.tool_calls[-1].name == "get_weather"
See Testing Guide for comprehensive testing strategies.
Always handle errors gracefully and return meaningful messages:
@function_tool
async def fetch_data(self, user_id: str) -> str:
"""Fetch user data from the database."""
try:
data = await database.get_user(user_id)
return f"User data: {data}"
except UserNotFoundError:
return f"No user found with ID {user_id}. Please check the ID and try again."
except DatabaseError as e:
return "I'm having trouble accessing the database right now. Please try again in a moment."
Design tools that respect user interruptions for better UX:
@function_tool
async def search_database(self, query: str, context: RunContext) -> str | None:
"""Search the database for matching records."""
# Allow user to interrupt long searches
search_task = asyncio.ensure_future(perform_search(query))
await context.speech_handle.wait_if_not_interrupted([search_task])
if context.speech_handle.interrupted:
search_task.cancel()
return None # Skip the tool reply
return search_task.result()
For complex agents with many tools:
order_create, order_update, order_cancel)See Best Practices for production patterns.
Load these as needed for specific patterns:
Complete, runnable examples demonstrating each pattern:
Before running examples, create a .env file with required API keys:
# LiveKit Configuration
LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret
# LLM Provider (choose one)
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
Install dependencies:
pip install livekit-agents livekit-plugins-openai livekit-plugins-silero python-dotenv aiohttp
| Pattern | When to Use | Key Features | Example |
|---|---|---|---|
| Basic Tools | Simple, fast operations | Direct return values, no state | basic-tool.py |
| API Integration | External service calls | Async HTTP, error handling | api-integration-tool.py |
| Long-Running | Time-consuming ops | Interruption support, cancellation | long-running-tool.py |
| Stateful | Multi-step workflows | RunContext.userdata, persistence | stateful-tool.py |
| Multi-Agent | Specialized routing | Agent handoffs, context transfer | agent-handoff-tool.py |
| Dynamic | Runtime tool creation | Permission-based, conditional | See dynamic-tools.md |
# Basic tool
@function_tool
async def tool_name(self, param: str) -> str:
"""Tool description with usage guidance."""
return result
# With state
@function_tool
async def stateful_tool(self, param: str, context: RunContext) -> str:
context.userdata["key"] = value
return result
# With interruption handling
@function_tool
async def long_tool(self, param: str, context: RunContext) -> str | None:
task = asyncio.ensure_future(operation())
await context.speech_handle.wait_if_not_interrupted([task])
if context.speech_handle.interrupted:
task.cancel()
return None
return task.result()
# Agent handoff
@function_tool
async def transfer_agent(self, context: RunContext):
new_agent = SpecialistAgent()
return new_agent, "Transferring to specialist"
Tool not being called: Improve the description to be more specific about when to use it.
Type errors: Ensure all parameters have proper type hints.
Interruption issues: Use RunContext.speech_handle.wait_if_not_interrupted() for long operations.
State not persisting: Use context.userdata to share state across tool calls.
Agent transitions fail: Return a tuple (new_agent, message) from the tool.
Import errors: Verify all required packages are installed (pip install livekit-agents livekit-plugins-openai).
For detailed guidance, consult the reference documentation above.