Alibaba Cloud Model Studio: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 and region parameters of the Client constructor (corresponding to the addresses below, which include the WorkspaceId). To switch to a different region, change the Client constructor parameters to the values of that region.

export DASHSCOPE_API_KEY="your-api-key"
# Replace {WorkspaceId} with your Beijing workspace ID.
export DASHSCOPE_HTTP_BASE_URL="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"
dashscope speech-synthesis create -m cosyvoice-v3-plus -t "Hello world" --voice longanyang

Core resource (client.audio.tts)

Resource path: client.audio.tts

Description: WebSocket-based speech synthesis, providing three calling modes: one-shot synthesis (synthesize), streaming synthesis (stream), and bidirectional streaming session (connect).

Calling entry

# One-shot synthesis: returns the complete audio bytes
audio = client.audio.tts.synthesize(
    text,                      # text is required (positional argument)
    model="cosyvoice-v3-plus", # The model that 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/play while receiving)
for chunk in client.audio.tts.stream(text, model=..., voice=...):
    ...

# Bidirectional streaming session: send text in segments; the connection is established automatically when entering the with context
with client.audio.tts.connect(model=..., voice=...) as session:
    session.send_text(text)    # Can be called multiple times to append text
    session.stop()             # Notify 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 strict real-time performance is not required. synthesize and stream create a new session for each call; there is no need to reuse instances.

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], the audio data yielded chunk by chunk.

Description: Streaming synthesis that sends the complete text at once and yields the audio chunk by chunk (Iterator[bytes]). It is best suited for short-text speech synthesis scenarios that require high real-time performance. To submit text in segments, use the bidirectional streaming session (connect).

send_text() - streaming input (session method)

Submits text in segments within 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. 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 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 to consume 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 task. After calling this method, the SDK immediately ends the current task. After cancellation, the current session can no longer send text; create a new session to continue synthesis.

Session metrics and responses (session attributes)

session.first_package_delay
session.usage
session.last_response

• first_package_delay: float, the delay 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 (with header/payload keys). The request_id of the most recent request can be obtained from the header of session.last_response, which is 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 (model, voice, format, sample_rate, volume, rate, pitch, language_hints, and enable_ssml are named parameters; the remaining 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 voice 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.5-plus",
    voice="cosyvoice-v3.5-plus-vd-announcer-xxxxxx", # 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 callback interfaces: 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: enter the with context to establish the connection automatically → multiple send_text() calls → stop() → read all results → exit the context to close the connection automatically.

audio_chunks() - iterate audio data

Method signature:

def audio_chunks(self) -> Iterator[bytes]

Return value: Iterator[bytes], the audio binary data yielded chunk by chunk, in the format specified by the format parameter.

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

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 the event type, original text, and other information. See output field in events() messages.

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

Error handling

When an error occurs during synthesis, the WebSocket layer raises a DashScopeError exception, and the connection is closed automatically after the exception is raised. Use try/except at the sending/receiving side 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 is synthesized and all audio data has been yielded, the events()/audio_chunks() iteration ends; after that, you can read performance metrics such as first_package_delay.
• The connection is closed automatically when exiting the with context (you can release audio playback devices and other resources afterwards).

output field in events() messages

The event dict yielded by the session events() iterator contains a payload.output field with synthesis event information. Use this field to track synthesis progress and retrieve per-sentence details. The output field structure 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 complete audio directly. Best suited for short-text speech synthesis.
• Unidirectional streaming: A non-blocking call that sends the complete text at once and receives the data by iterating the audio chunks yielded by stream. Best suited for short-text speech synthesis that requires high real-time performance.
• Bidirectional streaming: A non-blocking call that sends text in multiple segments and receives the incrementally synthesized audio stream in real time by iterating the session's audio_chunks()/events(). Best suited for long-text speech synthesis that requires high real-time performance.

# coding=utf-8

import os

import dashscope

# The API Key of the Singapore region is different from that of the Beijing region. Get an API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# If the environment variable is not configured, set your API key directly: api_key = "sk-xxx"
# The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The region value varies by region.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

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

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

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

# 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 Key of the Singapore region is different from that of the Beijing region. Get an API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# If the environment variable is not configured, set your API key directly: api_key = "sk-xxx"
# The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The region value varies by region.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

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

# Send the complete text at once, and receive the 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 Key of the Singapore region is different from that of the Beijing region. Get an API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# If the environment variable is not configured, set your API key directly: api_key = "sk-xxx"
# The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The region value varies by region.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# 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 session: the connection is established automatically when entering the with context
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 over the 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)
        # Send the text to synthesize in a streaming manner
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # End the streaming 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 WebSocket connection is established when the first text is sent, so the first package delay includes the connection establishment time
    print('[Metric] first package delay: {} ms'.format(session.first_package_delay))