Qwen-Audio-TTS Python SDK

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

Service endpoint

The SDK service endpoint is determined by the workspace/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 the target region.

Core resource (client.audio.tts)

Resource path: client.audio.tts

Function: WebSocket-based speech synthesis that offers three invocation modes: one-shot synthesis (synthesize), streaming synthesis (stream), and bidirectional streaming sessions (connect).

Invocation entry points

# One-shot synthesis: returns the complete audio as bytes
audio = client.audio.tts.synthesize(
    text,                      # text is required (positional)
    model="qwen-audio-3.0-tts-flash", # model the voice belongs to
    voice="longanhuan_v3.6",   # 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; 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()             # 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,
               voice: str, 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 for the synthesis to complete and returns the full 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 needed.

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

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

send_text() - streaming input (session method)

Submit 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 proceed concurrently (use this 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 keep 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 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 it 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, 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 of the synthesis (model, voice, format, sample_rate, volume, rate, pitch, language_hints, and enable_ssml are named parameters; all other parameters are passed through the parameters dictionary).

audio = client.audio.tts.synthesize(
    "Text to synthesize",
    model="qwen-audio-3.0-tts-flash",
    voice="longanhuan_v3.6",
    parameters={"bit_rate": 128},
)

audio = client.audio.tts.stream(
    "Text to synthesize",
    model="qwen-audio-3.0-tts-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="qwen-audio-3.0-tts-flash",
    voice="longanhuan_v3.6",
    parameters={
        "enable_aigc_tag": True,
        "aigc_propagator": "your_propagator",
        "aigc_propagate_id": "your_propagate_id"
    },
)

audio = client.audio.tts.synthesize(
    "Text to synthesize",
    model="qwen-audio-3.0-tts-flash",
    voice="longanhuan_v3.6", # Voice
    parameters={
        "hot_fix": {
            "pronunciation": [{"weather": "tian1 qi4"}],
            "replace": [{"today": "gold day"}]
        }
    },
)

Sessions 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: entering the with context opens the connection automatically → multiple send_text() calls → stop() → read all remaining 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] containing the audio binary data chunks yielded 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 produced 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. payload.output contains the event type, original text, and other details. See output field in events() events.

Description: Iterate server replies in the receiving thread. Contains the output information of synthesis events (event type, original text, sentence information); 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 at the sending or 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

• Entering the with context opens the connection automatically (initialize audio output streams or open file resources beforehand).
• After all text has been synthesized and all audio data has been produced, the events()/audio_chunks() iteration ends; performance metrics such as first_package_delay can then be read.
• Exiting the with context closes the connection automatically (release resources such as audio playback devices afterwards).

output field in events() events

The event dict yielded by the session's events() iteration 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 invocation 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 audio data by iterating over the chunks returned by stream. Best suited for short-text speech synthesis that requires low latency.
• Bidirectional streaming: A non-blocking call that sends text in multiple segments and receives the incrementally synthesized audio stream in real time by iterating over the session's audio_chunks()/events(). Best suited for long-text speech synthesis that requires low latency.

For more examples, see GitHub.

Non-streaming

The text sent in a single call must not exceed 20,000 characters. Exceeding this limit causes an error.

ImportantEach synthesize or stream call creates its own session; no instance reuse is needed.

# coding=utf-8

import os

import dashscope

# The API Keys for the Singapore and Beijing regions are different. Get API Key: https://help.aliyun.com/zh/model-studio/get-api-key
# If the environment variable is not configured, replace the following line with your Model Studio API Key: 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 = "qwen-audio-3.0-tts-flash"
# Voice
voice = "longanhuan_v3.6"

# Send the text for synthesis 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)

dashscope CLI

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 qwen-audio-3.0-tts-flash -t "Hello world" --voice longanhuan_v3.6

ImportantThe SDK Expert interactive assistant can accomplish the same development and troubleshooting in natural language. See

DashScope SDK Expert.

For the complete region table, see Base URL overview.

One-way streaming

The text sent in a single call must not exceed 20,000 characters. Exceeding this limit causes an error.

ImportantEach synthesize or stream call creates its own session; no instance reuse is needed.

# 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 the Singapore and Beijing regions are different. Get API Key: https://help.aliyun.com/zh/model-studio/get-api-key
# If the environment variable is not configured, replace the following line with your Model Studio API Key: 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 = "qwen-audio-3.0-tts-flash"
# Voice
voice = "longanhuan_v3.6"

# 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())

Bidirectional streaming

The text sent per call must not exceed 20,000 characters. The cumulative text must not exceed 200,000 characters.

• During streaming input, call send_text multiple times to submit text segments in sequence. The server automatically performs sentence segmentation on received text:

  • Complete sentences are synthesized immediately
  • Incomplete sentences are buffered until complete

  When stop is called, the server force-synthesizes all received but unprocessed text (including incomplete sentences).

• The interval between text segments must not exceed 23 seconds. Otherwise, a "request timeout after 23 seconds" exception is raised.

  If there's no text to send, call stop promptly to end the task.

  ImportantAlways call stop. Failing to do so may cause trailing text to not be converted to speech.

  The server enforces a 23-second timeout. This value can't be modified on the client side.

# 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 the Singapore and Beijing regions are different. Get API Key: https://help.aliyun.com/zh/model-studio/get-api-key
# If the environment variable is not configured, replace the following line with your Model Studio API Key: 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 = "qwen-audio-3.0-tts-flash"
# Voice
voice = "longanhuan_v3.6"

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: 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 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 for streaming synthesis
        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))