Implementation:Elevenlabs Elevenlabs python ClientTools
| Knowledge Sources | |
|---|---|
| Domains | Conversational_AI, Tool_Calling, Agent_Architecture |
| Last Updated | 2026-02-15 00:00 GMT |
Overview
Concrete tool for registering and executing client-side tool handlers in conversational AI sessions provided by the elevenlabs-python SDK.
Description
The ClientTools class manages registration and execution of client-side tools that can be called by the AI agent during a conversation. It maintains a thread-safe registry of named handlers (sync or async), runs an internal asyncio event loop in a dedicated thread, and handles execution dispatch with proper error handling.
When a tool call arrives from the agent, execute_tool schedules the handler coroutine on the event loop and delivers the result (or error) via a callback function that sends it back over WebSocket.
Usage
Create a ClientTools instance, register handlers using register(), and pass the instance to the Conversation constructor. The ClientTools lifecycle (start/stop) is managed automatically by the Conversation.
Code Reference
Source Location
- Repository: elevenlabs-python
- File: src/elevenlabs/conversational_ai/conversation.py
- Lines: L165-298
Signature
class ClientTools:
def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
"""
Args:
loop: Optional custom asyncio event loop. If not provided,
a new loop is created in a separate daemon thread.
"""
def register(
self,
tool_name: str,
handler: Union[Callable[[dict], Any], Callable[[dict], Awaitable[Any]]],
is_async: bool = False,
) -> None:
"""Register a tool handler.
Args:
tool_name: Unique tool identifier (must match agent config).
handler: Function implementing tool logic (receives dict of parameters).
is_async: Whether handler is an async function.
Raises:
ValueError: If handler is not callable or tool_name already registered.
"""
def start(self) -> None:
"""Start the event loop thread. Called automatically by Conversation."""
def stop(self) -> None:
"""Stop the event loop and clean up. Called automatically by Conversation."""
def execute_tool(
self, tool_name: str, parameters: dict, callback: Callable[[dict], None]
) -> None:
"""Execute a tool and send result via callback (non-blocking).
Args:
tool_name: Name of the registered tool.
parameters: Tool parameters from agent (includes tool_call_id).
callback: Function to call with the result dict.
"""
Import
from elevenlabs.conversational_ai.conversation import ClientTools
I/O Contract
Inputs (register)
| Name | Type | Required | Description |
|---|---|---|---|
| tool_name | str | Yes | Unique tool identifier matching agent dashboard config |
| handler | Callable[[dict], Any] | Yes | Sync or async function implementing tool logic |
| is_async | bool | No | Whether handler is async (default False) |
Inputs (execute_tool)
| Name | Type | Required | Description |
|---|---|---|---|
| tool_name | str | Yes | Name of registered tool to execute |
| parameters | dict | Yes | Tool parameters from agent (includes tool_call_id) |
| callback | Callable[[dict], None] | Yes | Callback receiving result dict with type, tool_call_id, result, is_error |
Outputs
| Name | Type | Description |
|---|---|---|
| callback receives | dict | {"type": "client_tool_result", "tool_call_id": str, "result": Any, "is_error": bool} |
Usage Examples
Register Sync Tool
from elevenlabs.conversational_ai.conversation import ClientTools
client_tools = ClientTools()
def get_weather(params):
city = params.get("city", "Unknown")
return f"The weather in {city} is sunny, 22°C."
client_tools.register("get_weather", get_weather)
Register Async Tool
import aiohttp
from elevenlabs.conversational_ai.conversation import ClientTools
client_tools = ClientTools()
async def search_database(params):
query = params.get("query", "")
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.example.com/search?q={query}") as resp:
data = await resp.json()
return str(data["results"][:3])
client_tools.register("search_database", search_database, is_async=True)
Full Conversation with Tools
from elevenlabs import ElevenLabs
from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface
from elevenlabs.conversational_ai.conversation import Conversation, ClientTools
client = ElevenLabs()
# Register tools
client_tools = ClientTools()
client_tools.register("get_weather", lambda p: f"Sunny, 22°C in {p.get('city')}")
client_tools.register("get_time", lambda p: "14:30 UTC")
# Create conversation with tools
conversation = Conversation(
client=client,
agent_id="your-agent-id",
requires_auth=False,
audio_interface=DefaultAudioInterface(),
client_tools=client_tools,
callback_agent_response=lambda text: print(f"Agent: {text}"),
)
conversation.start_session()
conversation.wait_for_session_end()