Overview
The ClientTools class handles registration and execution of client-side tools that can be called by the conversational AI agent. It supports both synchronous and asynchronous tools and runs them in a dedicated event loop to ensure non-blocking operation.
Constructor
ClientTools(loop: Optional[asyncio.AbstractEventLoop] = None)
loop
asyncio.AbstractEventLoop
Optional custom asyncio event loop to use for tool execution. If not provided, a new event loop will be created and run in a separate thread. Using a custom loop prevents “different event loop” runtime errors and allows for better context propagation and resource management.
Methods
start
Start the event loop in a separate thread for handling async operations. This is called automatically when creating a Conversation, but can be called manually if needed.
stop
Gracefully stop the event loop and clean up resources. This is called automatically when ending a Conversation.
register
client_tools.register(
tool_name: str,
handler: Union[Callable[[dict], Any], Callable[[dict], Awaitable[Any]]],
is_async: bool = False,
) -> None
Unique identifier for the tool. This name will be used by the agent to call the tool.
handler
Union[Callable[[dict], Any], Callable[[dict], Awaitable[Any]]]
required
Function that implements the tool’s logic. Receives a dictionary of parameters and returns the tool result.
Whether the handler is an async function. Defaults to False.
ValueError if the handler is not callableValueError if a tool with the same name is already registered
handle
await client_tools.handle(
tool_name: str,
parameters: dict
) -> Any
The name of the tool to execute.
Parameters to pass to the tool handler.
ValueError if the tool is not registered.
client_tools.execute_tool(
tool_name: str,
parameters: dict,
callback: Callable[[dict], None]
) -> None
The name of the tool to execute.
Parameters to pass to the tool handler.
callback
Callable[[dict], None]
required
Callback function to send the result to.
RuntimeError if the ClientTools event loop is not running.
Examples
from elevenlabs import ElevenLabs
from elevenlabs.conversational_ai import Conversation, DefaultAudioInterface, ClientTools
# Create client tools
client_tools = ClientTools()
# Register a synchronous tool
def get_weather(params: dict) -> str:
location = params.get("location", "unknown")
# Simulate weather lookup
return f"The weather in {location} is sunny and 72°F"
client_tools.register(
tool_name="get_weather",
handler=get_weather,
is_async=False
)
# Use with conversation
client = ElevenLabs(api_key="your-api-key")
conversation = Conversation(
client=client,
agent_id="your-agent-id",
requires_auth=True,
audio_interface=DefaultAudioInterface(),
client_tools=client_tools,
)
conversation.start_session()
# Agent can now call the get_weather tool
import asyncio
from elevenlabs import ElevenLabs
from elevenlabs.conversational_ai import Conversation, DefaultAudioInterface, ClientTools
# Create client tools
client_tools = ClientTools()
# Register an asynchronous tool
async def search_database(params: dict) -> dict:
query = params.get("query", "")
# Simulate async database search
await asyncio.sleep(1)
return {
"results": ["result1", "result2"],
"count": 2
}
client_tools.register(
tool_name="search_database",
handler=search_database,
is_async=True
)
# Use with conversation
client = ElevenLabs(api_key="your-api-key")
conversation = Conversation(
client=client,
agent_id="your-agent-id",
requires_auth=True,
audio_interface=DefaultAudioInterface(),
client_tools=client_tools,
)
conversation.start_session()
from elevenlabs.conversational_ai import ClientTools
import json
client_tools = ClientTools()
# Register multiple tools
def calculator(params: dict) -> float:
operation = params.get("operation")
a = params.get("a", 0)
b = params.get("b", 0)
if operation == "add":
return a + b
elif operation == "subtract":
return a - b
elif operation == "multiply":
return a * b
elif operation == "divide":
return a / b if b != 0 else "Error: Division by zero"
return "Unknown operation"
def get_user_info(params: dict) -> dict:
user_id = params.get("user_id")
# Simulate user lookup
return {
"id": user_id,
"name": "John Doe",
"email": "[email protected]"
}
client_tools.register("calculator", calculator, is_async=False)
client_tools.register("get_user_info", get_user_info, is_async=False)
# Now both tools are available to the agent
Using Custom Event Loop
import asyncio
from elevenlabs.conversational_ai import ClientTools
# Create a custom event loop
loop = asyncio.new_event_loop()
# Use it with ClientTools
client_tools = ClientTools(loop=loop)
# Register tools as usual
async def my_tool(params: dict) -> str:
return "Result"
client_tools.register("my_tool", my_tool, is_async=True)
Tool handlers receive parameters as a dictionary. The agent will call your tool with parameters based on the tool definition configured in your agent settings.
def my_tool(params: dict) -> Any:
# Access parameters from the dict
param1 = params.get("param1")
param2 = params.get("param2", "default_value")
# Process and return result
return {"result": "success"}
Return Values
Tools can return any JSON-serializable value:
- Strings
- Numbers
- Dictionaries
- Lists
- Booleans
- None
The return value will be sent back to the agent for processing.
Error Handling
If a tool raises an exception, it will be caught and the error message will be sent back to the agent:
def risky_tool(params: dict) -> str:
value = params.get("value")
if not value:
raise ValueError("Value parameter is required")
return f"Processed: {value}"