Implementation:Elevenlabs Elevenlabs python Conversation

Knowledge Sources	ElevenLabs Python ElevenLabs ConvAI Docs
Domains	Conversational_AI, WebSocket, Real_Time_Systems
Last Updated	2026-02-15 00:00 GMT

Overview

Concrete tool for managing real-time voice AI conversation sessions over WebSocket provided by the elevenlabs-python SDK.

Description

The Conversation class orchestrates a complete bidirectional voice conversation with an ElevenLabs AI agent. It extends BaseConversation (which handles WebSocket URL construction, initiation messages, and core message routing) and adds:

• Background thread for the WebSocket message loop
• Audio interface integration for microphone/speaker I/O
• Callback dispatch for conversation events
• Thread-safe session start/stop/wait lifecycle

The class handles 9 WebSocket message types: conversation_initiation_metadata, audio, agent_response, agent_chat_response_part, agent_response_correction, user_transcript, interruption, ping, and client_tool_call.

Usage

Create a Conversation with the SDK client, agent_id, audio interface, and optional callbacks/tools. Call start_session() to begin the conversation in a background thread. Call end_session() to terminate, and wait_for_session_end() to block until cleanup completes.

Code Reference

Source Location

• Repository: elevenlabs-python
• File: src/elevenlabs/conversational_ai/conversation.py
• Lines: L571-777 (Conversation class), L348-498 (BaseConversation shared logic)

Signature

class Conversation(BaseConversation):
    def __init__(
        self,
        client: BaseElevenLabs,
        agent_id: str,
        user_id: Optional[str] = None,
        *,
        requires_auth: bool,
        audio_interface: AudioInterface,
        config: Optional[ConversationInitiationData] = None,
        client_tools: Optional[ClientTools] = None,
        callback_agent_response: Optional[Callable[[str], None]] = None,
        callback_agent_response_correction: Optional[Callable[[str, str], None]] = None,
        callback_agent_chat_response_part: Optional[Callable[[str, AgentChatResponsePartType], None]] = None,
        callback_user_transcript: Optional[Callable[[str], None]] = None,
        callback_latency_measurement: Optional[Callable[[int], None]] = None,
        callback_audio_alignment: Optional[Callable[[AudioEventAlignment], None]] = None,
        callback_end_session: Optional[Callable] = None,
        on_prem_config: Optional[OnPremInitiationData] = None,
    ):
        """BETA: This API is subject to change.

        Args:
            client: ElevenLabs SDK client instance.
            agent_id: Agent ID from the ElevenLabs dashboard.
            user_id: Optional user identifier.
            requires_auth: Whether to use signed URL authentication.
            audio_interface: Audio I/O handler (microphone + speaker).
            config: Optional session config overrides.
            client_tools: Registered client-side tools.
            callback_agent_response: Called with agent's text response.
            callback_agent_response_correction: Called with (original, corrected) response.
            callback_agent_chat_response_part: Called with (text, part_type) for streaming.
            callback_user_transcript: Called with user's transcribed speech.
            callback_latency_measurement: Called with latency in milliseconds.
            callback_audio_alignment: Called with character-level timing data.
            callback_end_session: Called when session ends.
            on_prem_config: On-premises deployment configuration.
        """

    def start_session(self) -> None:
        """Start conversation in background thread."""

    def end_session(self) -> None:
        """End conversation and clean up resources."""

    def wait_for_session_end(self) -> Optional[str]:
        """Block until session ends. Returns conversation_id."""

    def send_user_message(self, text: str) -> None:
        """Send a text message from the user to the agent."""

    def send_contextual_update(self, text: str) -> None:
        """Send non-interrupting context to update conversation state."""

    def register_user_activity(self) -> None:
        """Ping to prevent session timeout."""

Import

from elevenlabs.conversational_ai.conversation import Conversation

I/O Contract

Inputs

Name	Type	Required	Description
client	BaseElevenLabs	Yes	Initialized ElevenLabs SDK client
agent_id	str	Yes	Agent ID from the ElevenLabs dashboard
requires_auth	bool	Yes	Use signed URL authentication
audio_interface	AudioInterface	Yes	Audio I/O handler
config	Optional[ConversationInitiationData]	No	Session config: extra_body, config_override, dynamic_variables
client_tools	Optional[ClientTools]	No	Registered client-side tool handlers
callback_agent_response	Optional[Callable[[str], None]]	No	Agent text response callback
callback_user_transcript	Optional[Callable[[str], None]]	No	User transcript callback
callback_latency_measurement	Optional[Callable[[int], None]]	No	Latency measurement callback (ms)

Outputs

Name	Type	Description
wait_for_session_end()	Optional[str]	Returns conversation_id after session completes
Callbacks	various	Agent responses, user transcripts, latency, alignment, corrections fired during session
Audio output	speaker playback	Agent audio played through AudioInterface

Usage Examples

Minimal Conversation

from elevenlabs import ElevenLabs
from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface
from elevenlabs.conversational_ai.conversation import Conversation

client = ElevenLabs()

conversation = Conversation(
    client=client,
    agent_id="your-agent-id",
    requires_auth=False,
    audio_interface=DefaultAudioInterface(),
)

conversation.start_session()
input("Press Enter to end the conversation...")
conversation.end_session()
conversation_id = conversation.wait_for_session_end()
print(f"Conversation ID: {conversation_id}")

With Callbacks and Tools

from elevenlabs import ElevenLabs
from elevenlabs.conversational_ai.default_audio_interface import DefaultAudioInterface
from elevenlabs.conversational_ai.conversation import (
    Conversation,
    ClientTools,
    ConversationInitiationData,
)

client = ElevenLabs()

# Set up tools
tools = ClientTools()
tools.register("get_time", lambda p: "2:30 PM UTC")

# Set up config overrides
config = ConversationInitiationData(
    dynamic_variables={"user_name": "Alice"},
)

conversation = Conversation(
    client=client,
    agent_id="your-agent-id",
    requires_auth=True,
    audio_interface=DefaultAudioInterface(),
    config=config,
    client_tools=tools,
    callback_agent_response=lambda text: print(f"Agent: {text}"),
    callback_user_transcript=lambda text: print(f"User: {text}"),
    callback_latency_measurement=lambda ms: print(f"Latency: {ms}ms"),
    callback_end_session=lambda: print("Session ended"),
)

conversation.start_session()
conversation.wait_for_session_end()

Related Pages

Implements Principle

• Principle:Elevenlabs_Elevenlabs_python_Conversational_Session

Requires Environment

• Environment:Elevenlabs_Elevenlabs_python_Python_Websockets
• Environment:Elevenlabs_Elevenlabs_python_PyAudio

Uses Heuristic

• Heuristic:Elevenlabs_Elevenlabs_python_Audio_Buffer_Sizes