Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Elevenlabs Elevenlabs python ClientTools

From Leeroopedia
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()

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment