CosyVoice Python SDK

Use the DashScope Python SDK to integrate CosyVoice real-time speech synthesis into your application through non-streaming, one-way streaming, or bidirectional streaming modes.

Service endpoint

The SDK endpoint is determined by the workspace/region parameters of the Client constructor (corresponding to the addresses below, which include WorkspaceId). To switch to a different region, change the Client constructor parameters to the values for that region.

Core resource (client.audio.tts)

Resource path: client.audio.tts

Capabilities: WebSocket-based speech synthesis with three calling modes: one-shot synthesis (synthesize), streaming synthesis (stream), and bidirectional streaming sessions (connect).

Calling modes

# One-shot synthesis: returns the complete audio as bytes
audio = client.audio.tts.synthesize(
    text,                      # text is required (positional parameter)
    model="cosyvoice-v3-plus", # The model the voice belongs to
    voice="longanyang",        # The voice
    format="mp3",              # Audio encoding format. Default: "mp3"
    sample_rate=22050,         # Sample rate. Default: 22050
)

# Streaming synthesis: yields audio bytes chunk by chunk (write or play as you receive)
for chunk in client.audio.tts.stream(text, model=..., voice=...):
    ...

# Bidirectional streaming session: submit text in segments. Entering the with context establishes the connection automatically
with client.audio.tts.connect(model=..., voice=...) as session:
    session.send_text(text)    # Can be called multiple times to append text
    session.stop()             # Notifies the server that the input is complete
    for chunk in session.audio_chunks():
        ...

synthesize() - non-streaming

Method signature:

def synthesize(self, text: str, *, model: str = "cosyvoice-v1",
               voice: str = "longxiaochun", format: str = "mp3",
               sample_rate: int = 22050, volume: int = None,
               rate: float = None, pitch: float = None,
               language_hints: list = None, enable_ssml: bool = None,
               parameters: Dict = None, ...) -> bytes

Parameters:

Return value: bytes containing the complete audio data.

Description: This blocking call waits and returns the complete audio data at once. It is best suited for short text where real-time streaming is not required. Each synthesize or stream call creates its own session; no instance reuse is required.

stream() - streaming

Method signature:

def stream(self, text: str, *, model: str, voice: str,
           format: str = "mp3", sample_rate: int = 22050, ...) -> Iterator[bytes]

Parameters:

Return value: Iterator[bytes] that yields audio data chunk by chunk.

Description: Streaming synthesis that sends the complete text at once and yields the audio chunk by chunk (Iterator[bytes]). Best suited for short-text scenarios that require low latency. To submit text in segments, use a bidirectional streaming session (connect).

send_text() - streaming input (session method)

Submits text in segments during a bidirectional streaming session (client.audio.tts.connect(...)).

Method signature:

def send_text(self, text: str) -> None

Parameters: text is the current text segment to synthesize. You can call send_text multiple times to append text. Maximum per call: 20,000 characters. Cumulative maximum: 200,000 characters. Sending text and receiving audio must run concurrently (use send_text together with audio_chunks()/events() and stop).

stop() - end streaming input (session method)

Method signature:

def stop(self) -> None

Description: Notifies the server that all text has been sent. This call does not block the current thread; the receiving thread must continue consuming audio_chunks()/events() until the remaining text is synthesized and all audio data is returned. Failing to call this method may result in trailing text not being converted to speech.

cancel() - cancel streaming synthesis (session method)

Method signature:

def cancel(self) -> None

Description: Cancels the current streaming speech synthesis turn. After calling this method, the SDK immediately ends the current task. After cancellation, the session can no longer send text. Create a new session to continue synthesis.

Session metrics and response (session properties)

session.first_package_delay
session.usage
session.last_response

• first_package_delay: float, the latency in milliseconds from sending text to receiving the first audio chunk. Read this after synthesis completes.
• usage: dict, the usage information of the most recent synthesis task.
• last_response: dict, the response message of the most recent synthesis task (containing header/payload keys). The request ID of the most recent request can be retrieved from the header of session.last_response, useful for troubleshooting and log correlation.

Request parameters

The following parameters are set through the synthesize/stream/connect methods of client.audio.tts to control the model, voice, format, and audio characteristics of the synthesized audio (model, voice, format, sample_rate, volume, rate, pitch, language_hints, and enable_ssml are named parameters; the other parameters are passed through the parameters dictionary).

audio = client.audio.tts.synthesize(
    "Text to synthesize",
    model="cosyvoice-v3-plus",
    voice="longanyang",
    parameters={"bit_rate": 128},
)

audio = client.audio.tts.stream(
    "Text to synthesize",
    model="cosyvoice-v3-flash",
    voice="your_voice",  # A system or cloned voice that supports word-level timestamps
    parameters={"word_timestamp_enabled": True},
)

audio = client.audio.tts.synthesize(
    "Text to synthesize",
    model="cosyvoice-v3-plus",
    voice="longanyang",
    parameters={
        "enable_aigc_tag": True,
        "aigc_propagator": "your_propagator",
        "aigc_propagate_id": "your_propagate_id"
    },
)

audio = client.audio.tts.synthesize(
    "Text to synthesize",
    model="cosyvoice-v3-plus",
    voice="longanyang", # Voice
    parameters={
        "hot_fix": {
            "pronunciation": [{"weather": "tian1 qi4"}],
            "replace": [{"today": "gold day"}]
        }
    },
)

audio = client.audio.tts.synthesize(
    "Text to synthesize",
    model="cosyvoice-v3-flash",
    voice="your_voice",  # A cloned voice of cosyvoice-v3-flash
    parameters={"enable_markdown_filter": True},
)

Session and event iteration

Streaming synthesis no longer uses callbacks: audio and server events are received through the session's iteration methods. stream()/synthesize() require no session management. The lifecycle of a bidirectional streaming session (client.audio.tts.connect(...)) is: entering the with context establishes the connection automatically → multiple send_text() calls → stop() → consume all results → exiting the context closes the connection automatically.

audio_chunks() - iterate audio data

Method signature:

def audio_chunks(self) -> Iterator[bytes]

Return value: Iterator[bytes] that yields binary audio data chunks one by one, in the format specified by the format parameter.

Description: Iterate in the receiving thread to receive audio data chunk by chunk; multiple chunks are yielded during synthesis. Write the data to a file or feed it to a playback device. Only one events() or audio_chunks() reader is allowed per session.

events() - iterate server events

Method signature:

def events(self) -> Generator[Union[Dict, bytes], None, None]

Return value: Yields both server response events (dict containing header (request information) and payload (output information)) and audio bytes. The payload.output field contains event type, original text, and other details. See output field in events() messages.

Description: Iterate in the receiving thread over server responses, which contain the output information of synthesis events (event type, original text, sentence information, etc.). You can access payload.output directly for details.

Error handling

When an error occurs during synthesis, the WebSocket layer raises a DashScopeError exception, and the connection closes automatically after the exception is raised. Use try/except where you send or receive to catch and log errors for troubleshooting:

from dashscope.v2 import DashScopeError

try:
    ...
except DashScopeError as e:
    print(e.code, e.message, e.request_id)

Session lifecycle

• The connection is established automatically when entering the with context (you can initialize audio output streams or open file resources beforehand).
• After all text has been synthesized and all audio data has been yielded, the events()/audio_chunks() iteration ends. Performance metrics such as first_package_delay can then be read.
• The connection closes automatically when exiting the with context (you can release resources such as audio playback devices afterwards).

output field in events() messages

The event dict yielded by the session's events() iteration contains a payload.output field with synthesis event output information. Use this field to track synthesis progress and retrieve per-sentence details. The structure of the output field is as follows:

Message example:

{
  "header": {
    "task_id": "xxx",
    "event": "result-generated",
    "attributes": {}
  },
  "payload": {
    "output": {
      "type": "sentence-begin",
      "original_text": "How is the weather today?",
      "sentence": {
        "index": 0,
        "words": []
      }
    }
  }
}

Parsing example:

def handle_event(event):
    output = (event.get('payload') or {}).get('output') or {}
    event_type = output.get('type', '')
    original_text = output.get('original_text', '')
    if event_type:
        print(f'Event type: {event_type}, Original text: {original_text}')

Code examples

The SDK provides the key interfaces for speech synthesis and supports the following calling modes:

• Non-streaming: A blocking call that sends the complete text at once and returns the full audio directly. Best suited for short-text speech synthesis.
• Unidirectional streaming: A non-blocking call that sends the complete text at once and receives data by iterating over the audio chunks returned by stream. Best suited for short-text scenarios that require low latency.
• Bidirectional streaming: A non-blocking call that sends text in multiple segments and receives incrementally synthesized audio in real time by iterating the session's audio_chunks()/events(). Best suited for long-text scenarios that require low latency.

For more examples, see GitHub.

# coding=utf-8

import os

import dashscope

# The API Keys for Singapore and Beijing regions are different. Get API Key: https://help.aliyun.com/model-studio/get-api-key
# If the environment variable is not configured, use your Model Studio API Key instead: api_key="sk-xxx"
# The following configuration is for the China (Beijing) region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# Model
model = "cosyvoice-v3-plus"
# Voice
voice = "longanyang"

# Send text for synthesis and get binary audio
audio = client.audio.tts.synthesize("How is the weather today?", model=model, voice=voice)

# Save audio to local file
with open('output.mp3', 'wb') as f:
    f.write(audio)

export DASHSCOPE_API_KEY="your-api-key"
# Replace {WorkspaceId} with your Workspace ID, and cn-beijing with the corresponding region (Singapore: ap-southeast-1, US East: us-east-1)
export DASHSCOPE_HTTP_BASE_URL="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"
dashscope speech-synthesis create -m cosyvoice-v2 -t "Hello world" --voice longxiaochun_v2

# coding=utf-8

import os

from datetime import datetime

import dashscope

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# The API Keys for Singapore and Beijing regions are different. Get API Key: https://help.aliyun.com/model-studio/get-api-key
# If the environment variable is not configured, use your Model Studio API Key instead: api_key="sk-xxx"
# The following configuration is for the China (Beijing) region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# Model
model = "cosyvoice-v3-plus"
# Voice
voice = "longanyang"

# Send the complete text at once and receive binary audio chunk by chunk
with open("output.mp3", "wb") as f:
    for chunk in client.audio.tts.stream("How is the weather today?", model=model, voice=voice):
        print(get_timestamp() + " Binary audio length: " + str(len(chunk)))
        f.write(chunk)
print("Speech synthesis completed, all results received: " + get_timestamp())

# coding=utf-8
#
# pyaudio installation instructions:
# For macOS, run the following commands:
#   brew install portaudio
#   pip install pyaudio
# For Debian/Ubuntu, run the following commands:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   or
#   pip install pyaudio
# For CentOS, run the following commands:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# For Microsoft Windows, run the following command:
#   python -m pip install pyaudio

import os
import time
import pyaudio
from concurrent.futures import ThreadPoolExecutor

import dashscope

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# The API Keys for Singapore and Beijing regions are different. Get API Key: https://help.aliyun.com/model-studio/get-api-key
# If the environment variable is not configured, use your Model Studio API Key instead: api_key="sk-xxx"
# The following configuration is for the China (Beijing) region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# Model
model = "cosyvoice-v3-plus"
# Voice
voice = "longanyang"

test_text = [
    "Streaming text-to-speech SDK,",
    "converts input text",
    "into binary audio data.",
    "Compared with non-streaming speech synthesis,",
    "streaming synthesis offers better real-time performance.",
    "Users hear near-synchronous audio output while typing,",
    "greatly improving interaction experience",
    "and reducing wait time.",
    "Ideal for large language model (LLM) integration,",
    "where text is streamed for speech synthesis.",
]

# Create a bidirectional streaming synthesis session: entering the with context establishes the connection automatically
with client.audio.tts.connect(
        model=model,
        voice=voice,
        format="pcm",
        sample_rate=22050,
) as session:
    # Initialize the player
    player = pyaudio.PyAudio()
    stream = player.open(format=pyaudio.paInt16, channels=1, rate=22050, output=True)

    # Receiving thread: iterate audio data and play it in real time
    def receive_audio():
        for chunk in session.audio_chunks():
            print(get_timestamp() + " Binary audio length: " + str(len(chunk)))
            stream.write(chunk)

    with ThreadPoolExecutor(max_workers=1) as pool:
        receiver = pool.submit(receive_audio)
        # Stream the text to synthesize
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # End streaming speech synthesis
        session.stop()
        receiver.result()

    print("Speech synthesis completed, all results received: " + get_timestamp())
    # Stop the player
    stream.stop_stream()
    stream.close()
    player.terminate()

    # The first text submission requires establishing a WebSocket connection, so the first packet latency includes the connection setup time
    print('[Metric] first packet latency: {} ms'.format(session.first_package_delay))