Tools

Register custom tools with ClientTools for sync and async functions

Overview

ClientTools allows you to register custom functions that your AI agent can call during conversations. Tools can be synchronous or asynchronous and run in a dedicated event loop to ensure non-blocking operation.

Basic Tool Registration

Register a simple synchronous tool:

from elevenlabs.conversational_ai.conversation import ClientTools

client_tools = ClientTools()

def calculate_sum(params):
    numbers = params.get("numbers", [])
    return sum(numbers)

client_tools.register("calculate_sum", calculate_sum, is_async=False)

Async Tools

Register asynchronous tools for I/O operations:

import asyncio
import aiohttp

async def fetch_weather(params):
    city = params.get("city", "San Francisco")
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.weather.com/{city}") as resp:
            data = await resp.json()
            return f"Weather in {city}: {data['temp']}°F"

client_tools.register("fetch_weather", fetch_weather, is_async=True)

Using Tools in Conversations

Pass ClientTools to your conversation:

from elevenlabs.client import ElevenLabs
from elevenlabs.conversational_ai.conversation import Conversation, ClientTools
from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface

elevenlabs = ElevenLabs(api_key="YOUR_API_KEY")

# Create and register tools
client_tools = ClientTools()

def get_account_balance(params):
    user_id = params.get("user_id")
    # Look up account balance
    return f"Account balance for {user_id}: $1,234.56"

client_tools.register("get_account_balance", get_account_balance, is_async=False)

# Use tools in conversation
audio_interface = DefaultAudioInterface()

conversation = Conversation(
    client=elevenlabs,
    agent_id="your-agent-id",
    requires_auth=True,
    audio_interface=audio_interface,
    client_tools=client_tools,
)

conversation.start_session()

When the agent calls get_account_balance, the tool executes and returns the result to the agent.

Tool Parameters

Tools receive a dictionary of parameters:

def book_appointment(params):
    """
    params will contain:
    - tool_call_id: Unique identifier for this tool call
    - Plus any parameters sent by the agent
    """
    date = params.get("date")
    time = params.get("time")
    service = params.get("service")
    
    # Validate parameters
    if not date or not time:
        raise ValueError("Date and time are required")
    
    # Book the appointment
    confirmation = f"Booked {service} on {date} at {time}"
    return confirmation

client_tools.register("book_appointment", book_appointment, is_async=False)

Parameter Structure

tool_call_id · str

Unique identifier for this tool invocation (automatically added)

...custom_params · any

Any additional parameters sent by the agent when calling the tool

Error Handling

Handle errors gracefully in your tools:

def process_payment(params):
    try:
        amount = params.get("amount")
        card_number = params.get("card_number")
        
        if amount <= 0:
            raise ValueError("Amount must be positive")
        
        # Process payment logic
        return f"Payment of ${amount} processed successfully"
        
    except ValueError as e:
        # Error will be sent to the agent
        raise ValueError(f"Invalid payment: {str(e)}")
    except Exception as e:
        # Log unexpected errors
        logging.error(f"Payment processing error: {e}")
        raise Exception("Payment processing failed. Please try again.")

client_tools.register("process_payment", process_payment, is_async=False)

When a tool raises an exception, the error message is sent to the agent with is_error=True.

Custom Event Loops

For advanced use cases involving context propagation or resource reuse, provide a custom asyncio event loop:

import asyncio
from elevenlabs.conversational_ai.conversation import ClientTools

async def main():
    # Get the current event loop
    custom_loop = asyncio.get_running_loop()
    
    # Create ClientTools with custom loop
    client_tools = ClientTools(loop=custom_loop)
    
    # Register async tool
    async def get_user_data(params):
        user_id = params.get("user_id")
        # Use shared async resources (DB connections, HTTP sessions, etc.)
        return {"user_id": user_id, "name": "John Doe"}
    
    client_tools.register("get_user_data", get_user_data, is_async=True)
    
    # Use with conversation
    conversation = Conversation(
        client=elevenlabs,
        agent_id="your-agent-id",
        requires_auth=True,
        audio_interface=audio_interface,
        client_tools=client_tools,
    )
    
    await conversation.start_session()
    # ... rest of conversation logic

asyncio.run(main())

Custom Loop Parameters

loop · asyncio.AbstractEventLoop

Custom event loop to use for tool execution. If not provided, a new loop is created in a separate thread.

Benefits of Custom Event Loops

Context Propagation

Maintain request-scoped state across async operations

Resource Reuse

Share async resources like HTTP sessions or database pools

Loop Management

Prevent “different event loop” runtime errors

Performance

Better control over async task scheduling and execution

When using a custom loop, you’re responsible for its lifecycle. Don’t close the loop while ClientTools are still using it.

Complex Tool Example

Here’s a complete example with database access and error handling:

import asyncio
import aiosqlite
from elevenlabs.conversational_ai.conversation import ClientTools

class DatabaseTools:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.client_tools = ClientTools()
        self._register_tools()
    
    def _register_tools(self):
        self.client_tools.register(
            "get_order_status",
            self.get_order_status,
            is_async=True
        )
        self.client_tools.register(
            "update_order_status",
            self.update_order_status,
            is_async=True
        )
    
    async def get_order_status(self, params):
        order_id = params.get("order_id")
        
        if not order_id:
            raise ValueError("order_id is required")
        
        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                "SELECT status, estimated_delivery FROM orders WHERE id = ?",
                (order_id,)
            )
            row = await cursor.fetchone()
            
            if not row:
                return f"Order {order_id} not found"
            
            status, delivery = row
            return f"Order {order_id} is {status}. Estimated delivery: {delivery}"
    
    async def update_order_status(self, params):
        order_id = params.get("order_id")
        new_status = params.get("status")
        
        if not order_id or not new_status:
            raise ValueError("order_id and status are required")
        
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute(
                "UPDATE orders SET status = ? WHERE id = ?",
                (new_status, order_id)
            )
            await db.commit()
            return f"Order {order_id} updated to {new_status}"

# Usage
db_tools = DatabaseTools("orders.db")

conversation = Conversation(
    client=elevenlabs,
    agent_id="your-agent-id",
    requires_auth=True,
    audio_interface=audio_interface,
    client_tools=db_tools.client_tools,
)

Tool Registration Reference

register()

tool_name · str · required

Unique identifier for the tool. Must match the tool name configured in the agent.

handler · Callable · required

Function that implements the tool logic. Can be sync or async.

is_async · bool · required

Whether the handler is an async function

Lifecycle Methods

start · method

Starts the event loop for tool execution. Called automatically when the conversation starts.

client_tools.start()

stop · method

Stops the event loop and cleans up resources. Called automatically when the conversation ends.

client_tools.stop()

Best Practices

Use async for I/O operations

Always use async tools for network requests, database queries, or file I/O to avoid blocking the main conversation thread.

async def fetch_data(params):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

Validate parameters

Always validate tool parameters before processing to provide clear error messages.

def my_tool(params):
    required = ["param1", "param2"]
    for param in required:
        if param not in params:
            raise ValueError(f"{param} is required")

Return user-friendly messages

Tool return values are sent to the agent, so make them clear and actionable.

# Good
return "Appointment booked for March 15 at 2:00 PM"

# Less helpful
return {"success": True, "id": 12345}

Handle errors gracefully

Catch and handle errors to provide meaningful feedback to users.

try:
    result = process_request(params)
    return result
except ValueError as e:
    raise ValueError(f"Invalid input: {e}")
except Exception as e:
    logging.error(f"Unexpected error: {e}")
    raise Exception("Something went wrong. Please try again.")

Keep tools focused

Each tool should do one thing well. Break complex operations into multiple tools.

# Good: Separate tools
client_tools.register("get_order", get_order, is_async=True)
client_tools.register("update_order", update_order, is_async=True)
client_tools.register("cancel_order", cancel_order, is_async=True)

# Less ideal: One tool doing everything
client_tools.register("manage_order", manage_order, is_async=True)

Complete Example

import asyncio
import aiohttp
from elevenlabs.client import ElevenLabs
from elevenlabs.conversational_ai.conversation import (
    Conversation,
    ClientTools,
)
from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface

elevenlabs = ElevenLabs(api_key="YOUR_API_KEY")

# Create tools
client_tools = ClientTools()

# Sync tool
def calculate_sum(params):
    numbers = params.get("numbers", [])
    return sum(numbers)

# Async tool
async def fetch_data(params):
    url = params.get("url")
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            data = await resp.json()
            return data

# Register tools
client_tools.register("calculate_sum", calculate_sum, is_async=False)
client_tools.register("fetch_data", fetch_data, is_async=True)

# Create conversation with tools
audio_interface = DefaultAudioInterface()

conversation = Conversation(
    client=elevenlabs,
    agent_id="your-agent-id",
    requires_auth=True,
    audio_interface=audio_interface,
    client_tools=client_tools,
)

conversation.start_session()
input("Press Enter to end...")
conversation.end_session()