ElevenLabs Python Library

The official Python SDK for ElevenLabs. ElevenLabs brings the most compelling, rich and lifelike voices to creators and developers in just a few lines of code.

📖 API & Docs

Check out the HTTP API documentation.

Install

pip install elevenlabs

Usage

Main Models

1. Eleven v3 (eleven_v3)

   • Dramatic delivery and performances
   • 70+ languages supported
   • Supported for natural multi-speaker dialogue

2. Eleven Multilingual v2 (eleven_multilingual_v2)

   • Excels in stability, language diversity, and accent accuracy
   • Supports 29 languages
   • Recommended for most use cases

3. Eleven Flash v2.5 (eleven_flash_v2_5)

   • Ultra-low latency
   • Supports 32 languages
   • Faster model, 50% lower price per character

4. Eleven Turbo v2.5 (eleven_turbo_v2_5)

   • Good balance of quality and latency
   • Ideal for developer use cases where speed is crucial
   • Supports 32 languages

For more detailed information about these models and others, visit the ElevenLabs Models documentation.

from dotenv import load_dotenv
from elevenlabs.client import ElevenLabs
from elevenlabs.play import play

load_dotenv()

elevenlabs = ElevenLabs()

audio = elevenlabs.text_to_speech.convert(
    text="The first move is what sets everything in motion.",
    voice_id="JBFqnCBsd6RMkjVDRZzb",
    model_id="eleven_v3",
    output_format="mp3_44100_128",
)

play(audio)

Voices

List all your available voices with search().

from elevenlabs.client import ElevenLabs

elevenlabs = ElevenLabs(
  api_key="YOUR_API_KEY",
)

response = elevenlabs.voices.search()
print(response.voices)

For information about the structure of the voices output, please refer to the official ElevenLabs API documentation for Get Voices.

Build a voice object with custom settings to personalize the voice style, or call elevenlabs.voices.settings.get("your-voice-id") to get the default settings for the voice.

Clone Voice

Clone your voice in an instant. Note that voice cloning requires an API key, see below.

from elevenlabs.client import ElevenLabs
from elevenlabs.play import play

elevenlabs = ElevenLabs(
  api_key="YOUR_API_KEY",
)

voice = elevenlabs.voices.ivc.create(
    name="Alex",
    description="An old American male voice with a slight hoarseness in his throat. Perfect for news", # Optional
    files=["./sample_0.mp3", "./sample_1.mp3", "./sample_2.mp3"],
)

Streaming

Stream audio in real-time, as it's being generated.

from elevenlabs import stream
from elevenlabs.client import ElevenLabs

elevenlabs = ElevenLabs(
  api_key="YOUR_API_KEY",
)

audio_stream = elevenlabs.text_to_speech.stream(
    text="This is a test",
    voice_id="JBFqnCBsd6RMkjVDRZzb",
    model_id="eleven_multilingual_v2"
)

# option 1: play the streamed audio locally
stream(audio_stream)

# option 2: process the audio bytes manually
for chunk in audio_stream:
    if isinstance(chunk, bytes):
        print(chunk)

Async Client

Use AsyncElevenLabs if you want to make API calls asynchronously.

import asyncio

from elevenlabs.client import AsyncElevenLabs

elevenlabs = AsyncElevenLabs(
  api_key="MY_API_KEY"
)

async def print_models() -> None:
    models = await elevenlabs.models.list()
    print(models)

asyncio.run(print_models())

ElevenAgents

Build interactive AI agents with real-time audio capabilities using ElevenAgents.

Basic Usage

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 audio interface for real-time audio input/output
audio_interface = DefaultAudioInterface()

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

# Start the conversation
conversation.start_session()

# The conversation runs in background until you call:
conversation.end_session()

Disconnection Details

Pass callback_disconnect to distinguish user shutdown, normal server closure, and errors. It replaces callback_end_session, which is deprecated:

from elevenlabs.conversational_ai.conversation import Conversation, DisconnectionDetails


def on_disconnect(details: DisconnectionDetails):
    print(details.reason, details.close_code, details.close_reason, details.message)


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

reason is "user" for end_session(), "agent" for a normal server close, or "error" for an abnormal close or other exception. WebSocket close details are included when available; message describes errors. The callback runs once, after callback_end_session. For AsyncConversation, provide an async callback.

Custom Event Loop Support

For advanced use cases involving context propagation, resource reuse, or specific event loop management, ClientTools supports custom asyncio event loops:

import asyncio
from elevenlabs.conversational_ai.conversation import ClientTools

elevenlabs = ElevenLabs(
  api_key="YOUR_API_KEY",
)

async def main():
    # Get the current event loop
    custom_loop = asyncio.get_running_loop()

    # Create ClientTools with custom loop to prevent "different event loop" errors
    client_tools = ClientTools(loop=custom_loop)

    # Register your tools
    async def get_weather(params):
        location = params.get("location", "Unknown")
        # Your async logic here
        return f"Weather in {location}: Sunny, 72°F"

    client_tools.register("get_weather", get_weather, 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
    )

asyncio.run(main())

Benefits of Custom Event Loop:

• Context Propagation: Maintain request-scoped state across async operations
• Resource Reuse: Share existing async resources like HTTP sessions or database pools
• Loop Management: Prevent "Task got Future attached to a different event loop" errors
• Performance: Better control over async task scheduling and execution

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

Tool Registration

Register custom tools that the AI agent can call during conversations:

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")
    # Your async HTTP request logic
    return {"data": "fetched"}

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

Agent workflows as code

Build an agent workflow with chained methods instead of a hand-written graph of nodes and edges. The Workflow object stands for the start node. Each verb adds a node and returns it, so flows read top to bottom:

when(condition) opens a conditional branch. A string is an LLM condition; an Expr is an expression over dynamic variables. go_to(node) moves the conversation to a node that already exists, for merges and loops. The API allows one edge between any two nodes, so going back to a node that already leads here is sent as that edge's backward condition; any other second edge between the same two nodes fails build().

up() returns the node the current one was chained from, so you can write a whole workflow as one chain:

(
    wf.converse("triage", "Greet the caller and find out why they're calling")
    .when("Caller wants a human").transfer("human", Destination.phone("+15551234567")).up()
    .when("Caller is done").end()
)

from elevenlabs import ElevenLabs
from elevenlabs.agent_workflows import Destination, Expr, Workflow

wf = Workflow()

triage = wf.converse("triage", "Greet the caller and find out why they're calling")
identify = triage.when("Caller wants to check an order").converse("identify", "Ask for the order number")

lookup = identify.when("Caller gave an order number").tool("lookup_order", ["tool_get_order"])
lookup.on_result(
    success=lambda ok: ok.converse("resolve", "Explain the order status").when("Caller is satisfied").end(),
    failure=lambda failed: failed.go_to(identify),
)

triage.when(Expr.variable("tier").eq("gold")).transfer("vip", Destination.agent("agent_vip"))
triage.when("Caller wants a human").transfer("human", Destination.phone("+15551234567"))

client = ElevenLabs()
client.conversational_ai.agents.update("agent_id", workflow=wf.build())

build() checks the whole graph and raises one WorkflowValidationError whose issues list every problem it finds. It also generates edge ids and edge order, and lays out nodes that don't have a position, so they don't all stack on top of each other in the editor.

Editing an existing agent

agent = client.conversational_ai.agents.get("agent_id")
wf = Workflow.from_agent(agent)  # keeps node ids, editor positions and edge order

wf.node("resolve").update(prompt="Explain the status, then offer a discount code")
verify = wf.insert_between("identify", "lookup_order", lambda start: start.converse("verify", "Verify the caller's identity"))
if verify is not None:  # None if there's no such edge
    verify.when("Caller can't verify").end(id="unverified")
wf.remove("vip")

client.conversational_ai.agents.update("agent_id", workflow=wf.build())

Storing workflows in git

to_json_string() writes the workflow in API format: snake_case keys, sorted, the same shape the REST API and the ElevenLabs CLI use, and byte for byte what the other ElevenLabs SDKs write. Workflow.from_json() reads it back. It also accepts a full agent config file that has a workflow key.

from pathlib import Path

Path("workflows/support.json").write_text(wf.to_json_string())
stored = Workflow.from_json(Path("workflows/support.json").read_text())

A workflow loaded from the API or a file can contain node types that are newer than your SDK version. They round-trip through from_json() and to_json_string() unchanged, and build() warns about each one (pass on_warning= to see them). build() can't type those nodes, so Pydantic warns when the request is serialized. Upgrade the SDK, or send the wire format directly:

client.conversational_ai.agents.update(
    "agent_id", request_options={"additional_body_parameters": {"workflow": wf.to_json()}}
)

Speech Engine

Speech Engine lets you build server-side voice agents that receive real-time transcripts from the ElevenLabs API and stream LLM responses back for text-to-speech synthesis. Your server acts as a WebSocket endpoint — ElevenLabs connects to it, sends user transcripts, and your code decides how to respond.

Speech Engine is async-only and available on AsyncElevenLabs.

Quick Start

import asyncio
from openai import AsyncOpenAI
from elevenlabs import AsyncElevenLabs

openai_client = AsyncOpenAI()
elevenlabs = AsyncElevenLabs()

async def main():
    engine = await elevenlabs.speech_engine.get("seng_123")

    async def on_transcript(transcript, session):
        stream = await openai_client.responses.create(
            model="gpt-4o",
            input=[
                {"role": "assistant" if m.role == "agent" else m.role, "content": m.content}
                for m in transcript
            ],
            stream=True,
        )
        await session.send_response(stream)

    async def on_init(conversation_id, session):
        print(f"Session started: {conversation_id}")

    async def on_close(session):
        print(f"Session ended: {session.conversation_id}")

    async def on_error(err, session):
        print(f"Error: {err}")

    await engine.serve(
        port=3001,
        debug=True,
        on_init=on_init,
        on_transcript=on_transcript,
        on_close=on_close,
        on_error=on_error,
    )

asyncio.run(main())

How It Works

When engine.serve() starts, it opens a WebSocket server on the specified port. For each incoming connection from the ElevenLabs API:

1. An init message arrives with a conversation_id
2. As the user speaks, user_transcript messages arrive with the full conversation history
3. Your on_transcript handler generates a response (using any LLM) and calls session.send_response()
4. If the user interrupts (speaks again mid-response), the previous handler is automatically cancelled

Sending Responses

send_response() accepts strings or async iterators. LLM stream formats from OpenAI, Anthropic, and Google Gemini are auto-detected:

# Plain string
await session.send_response("Hello world")

# OpenAI stream (auto-parsed)
stream = await openai_client.responses.create(model="gpt-4o", ..., stream=True)
await session.send_response(stream)

# Anthropic stream (auto-parsed)
stream = anthropic_client.messages.stream(model="claude-sonnet-4-20250514", ...)
await session.send_response(stream)

# Any async iterator of strings
async def my_generator():
    yield "Hello "
    yield "world"
await session.send_response(my_generator())

Interruption Handling

When a new transcript arrives while a previous response is still streaming, the previous handler's asyncio.Task is cancelled automatically. Any await in your handler (including LLM SDK calls) will raise asyncio.CancelledError, which cleanly aborts the in-flight request. No manual signal handling needed.

Custom Server Integration (FastAPI, Starlette)

For integrating with an existing web server, use create_session() instead of serve():

from fastapi import FastAPI, WebSocket

app = FastAPI()
engine = ...  # SpeechEngineResource from await client.speech_engine.get(...)

@app.websocket("/api/speech-engine/ws")
async def speech_engine_ws(ws: WebSocket):
    await ws.accept()
    session = engine.create_session(ws, debug=True)
    session.on("user_transcript", handle_transcript)
    await session.run()

When using session.on() directly, handlers receive just the event data (no session argument, since you already have the reference):

Standalone Server

For full control over the server lifecycle, use SpeechEngineServer directly:

from elevenlabs.speech_engine import SpeechEngineServer

server = SpeechEngineServer(
    port=3001,
    debug=True,
    on_transcript=handle_transcript,
)

# In one task:
await server.serve()

# In another task (e.g. signal handler):
await server.stop()

Authentication

Both engine.serve() and SpeechEngineServer automatically verify the X-Elevenlabs-Speech-Engine-Authorization header on every incoming connection, rejecting any requests that were not signed by ElevenLabs. The API key is read from the AsyncElevenLabs client (or api_key= on SpeechEngineServer), falling back to the ELEVENLABS_API_KEY environment variable.

Disabling authentication

If your server sits behind an infrastructure layer that already restricts incoming traffic to ElevenLabs (typically an IP allowlist scoped to ElevenLabs' egress ranges), you can skip JWT verification by passing disable_auth=True:

# Via engine.serve() — no api_key required when disable_auth is True
await engine.serve(port=3001, disable_auth=True, on_transcript=on_transcript)

# Or directly on SpeechEngineServer
server = SpeechEngineServer(port=3001, disable_auth=True, on_transcript=on_transcript)
await server.serve()

When auth is disabled the server accepts any client that can reach it and emits a UserWarning on startup. Only use this if you have an IP allowlist or equivalent network-level restriction in front of the server — without one, anyone on the internet can open a session and consume your compute and downstream LLM quota.

Languages Supported

Explore all models & languages.

Contributing

While we value open-source contributions to this SDK, this library is generated programmatically. Additions made directly to this library would have to be moved over to our generation code, otherwise they would be overwritten upon the next generated release. Feel free to open a PR as a proof of concept, but know that we will not be able to merge it as-is. We suggest opening an issue first to discuss with us!

On the other hand, contributions to the README are always very welcome!