TTS

A TTS Model converts text into synthesized speech audio, supporting both standard and realtime (streaming-input) synthesis modes. AgentScope currently ships the following TTS model classes:

Provider	Model Class	Highlights
OpenAI	`OpenAITTSModel`	tts-1, tts-1-hd, gpt-4o-mini-tts; multiple voices; configurable output format (mp3, wav, opus, …)
Gemini	`GeminiTTSModel`	gemini-2.5-flash-preview-tts, gemini-2.5-pro-preview-tts; 30 prebuilt voices; streaming output
DashScope	`DashScopeTTSModel`	Qwen3-TTS, multiple voices, streaming output
DashScope (Realtime)	`DashScopeRealtimeTTSModel`	Qwen3-TTS WebSocket streaming input, ideal for LLM output piping
DashScope (CosyVoice)	`DashScopeCosyVoiceTTSModel`	CosyVoice-v3, supports both standard and realtime (streaming-input) modes; cosyvoice-v3-flash/plus

Create TTS Model

Every TTS model takes a credential, a model name, and an optional provider-specific Parameters object. The tabs below show the standard and realtime setups:

import os
from agentscope.tts import OpenAITTSModel
from agentscope.credential import OpenAICredential

tts = OpenAITTSModel(
    credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
    model="tts-1",
    parameters=OpenAITTSModel.Parameters(voice="alloy", response_format="mp3"),
    stream=True,
)

import os
from agentscope.tts import GeminiTTSModel
from agentscope.credential import GeminiCredential

tts = GeminiTTSModel(
    credential=GeminiCredential(api_key=os.environ["GEMINI_API_KEY"]),
    model="gemini-2.5-flash-preview-tts",
    parameters=GeminiTTSModel.Parameters(voice="Kore"),
    stream=True,
)

import os
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential

tts = DashScopeTTSModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="qwen3-tts-flash",
    parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
    stream=True,
)

import os
from agentscope.tts import DashScopeRealtimeTTSModel
from agentscope.credential import DashScopeCredential

tts = DashScopeRealtimeTTSModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="qwen3-tts-flash-realtime",
    parameters=DashScopeRealtimeTTSModel.Parameters(voice="Serena"),
    stream=True,
)

import os
from agentscope.tts import DashScopeCosyVoiceTTSModel
from agentscope.credential import DashScopeCredential

tts = DashScopeCosyVoiceTTSModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="cosyvoice-v3-flash",
    parameters=DashScopeCosyVoiceTTSModel.Parameters(voice="longanhuan"),
    stream=True,
)

import os
from agentscope.tts import DashScopeCosyVoiceTTSModel
from agentscope.credential import DashScopeCredential

tts = DashScopeCosyVoiceTTSModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="cosyvoice-v3-flash",
    parameters=DashScopeCosyVoiceTTSModel.Parameters(voice="longanhuan", realtime=True),
    stream=True,
)

Common constructor arguments shared by every TTS model:

Argument	Type	Description
`credential`	`CredentialBase`	Provider-specific credential
`model`	`str`	Model identifier (e.g. `"qwen3-tts-flash"`)
`parameters`	`Parameters \| None`	Provider-specific parameters such as `voice`
`stream`	`bool`	Whether to stream audio output

Additional arguments for DashScopeRealtimeTTSModel and DashScopeCosyVoiceTTSModel (realtime mode):

Argument	Type	Default	Description
`cold_start_length`	`int \| None`	`None`	Minimum character count before first text chunk is sent to the API
`cold_start_words`	`int \| None`	`None`	Minimum word count before first text chunk is sent
`max_retries`	`int`	`3`	Maximum retry attempts on WebSocket failure
`retry_delay`	`float`	`5.0`	Initial retry delay in seconds (exponential backoff)

Call TTS Model

Invoke the model by calling synthesize() with the text to speak:

async def synthesize(
    self,
    text: str | None = None,
    **kwargs: Any,
) -> TTSResponse | AsyncGenerator[TTSResponse, None]:

The return type follows the model’s stream setting:

stream=False: returns a single TTSResponse with the complete audio.
stream=True: returns an AsyncGenerator[TTSResponse, None]. Each chunk carries an incremental audio delta; the final chunk has is_last=True.

Each TTSResponse carries:

Field	Type	Description
`content`	`DataBlock \| None`	Audio data. Format indicated by `content.source.media_type` (e.g. `"audio/wav"`, `"audio/pcm;rate=24000"`)
`is_last`	`bool`	`True` on the final streaming chunk
`usage`	`TTSUsage \| None`	Token counts (`input_tokens`, `output_tokens`) and elapsed `time` in seconds
`id`	`str`	Auto-generated unique identifier
`metadata`	`dict \| None`	Optional provider-specific metadata

import asyncio
import os
from agentscope.tts import OpenAITTSModel
from agentscope.credential import OpenAICredential

async def main():
    tts = OpenAITTSModel(
        credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
        model="tts-1",
        parameters=OpenAITTSModel.Parameters(voice="alloy"),
        stream=True,
    )

    async for chunk in await tts.synthesize("Hello, world!"):
        if chunk.content:
            # chunk.content is a DataBlock with base64-encoded audio/mpeg
            print(f"Audio chunk: {len(chunk.content.source.data)} bytes")

asyncio.run(main())

import asyncio
import os
from agentscope.tts import GeminiTTSModel
from agentscope.credential import GeminiCredential

async def main():
    tts = GeminiTTSModel(
        credential=GeminiCredential(api_key=os.environ["GEMINI_API_KEY"]),
        model="gemini-2.5-flash-preview-tts",
        parameters=GeminiTTSModel.Parameters(voice="Kore"),
        stream=True,
    )

    async for chunk in await tts.synthesize("Hello, world!"):
        if chunk.content:
            # chunk.content is a DataBlock with base64-encoded audio/wav
            print(f"Audio chunk: {len(chunk.content.source.data)} bytes")

asyncio.run(main())

import asyncio
import os
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential

async def main():
    tts = DashScopeTTSModel(
        credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
        model="qwen3-tts-flash",
        parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
        stream=True,
    )

    # Streaming synthesis
    async for chunk in await tts.synthesize("Hello, world!"):
        if chunk.content:
            # chunk.content is a DataBlock with base64-encoded audio/wav
            print(f"Audio chunk: {len(chunk.content.source.data)} bytes")

asyncio.run(main())

Realtime TTS (Streaming Input)

For realtime models (DashScopeRealtimeTTSModel and DashScopeCosyVoiceTTSModel with realtime=True), text can be pushed incrementally as it arrives from a streaming LLM. Both share the same push() / synthesize() interface. The lifecycle is managed via async with or manual connect() / close():

DashScopeRealtimeTTSModel (Qwen3) produces audio at token-level granularity, so each push() call typically returns audio data. In contrast, DashScopeCosyVoiceTTSModel with realtime=True relies on the CosyVoice server which automatically segments text into sentences before synthesizing. Audio is only returned after a complete sentence boundary is detected, so push() may return empty responses for partial sentences. Calling synthesize() forces synthesis of all remaining text including incomplete sentences.

import asyncio
import os
from agentscope.tts import DashScopeRealtimeTTSModel
from agentscope.credential import DashScopeCredential

async def main():
    tts = DashScopeRealtimeTTSModel(
        credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
        model="qwen3-tts-flash-realtime",
        parameters=DashScopeRealtimeTTSModel.Parameters(voice="Cherry"),
        stream=True,
    )

    async with tts:
        # Push text incrementally as it arrives from a streaming LLM.
        # Each push() returns a TTSResponse with audio accumulated so far
        # (or content=None if not yet available).
        resp1 = await tts.push("Hello, ")
        if resp1.content:
            print("Audio available after first push")

        resp2 = await tts.push("how are you today?")
        if resp2.content:
            print("Audio available after second push")

        # Finalize: flush remaining buffered text and collect final audio.
        # text= is optional: pass extra text to append before finalizing,
        # or omit to finalize previously pushed text only.
        response = await tts.synthesize()

asyncio.run(main())

The realtime lifecycle methods:

Method	Description
`connect()`	Open WebSocket connection
`push(text)`	Append text incrementally (non-blocking), returns audio accumulated so far
`synthesize()`	Finalize and return remaining audio
`close()`	Tear down connection

Integrate with Agent

In the agent layer, TTS is integrated via TTSMiddleware, which intercepts the agent’s text output and synthesizes speech automatically:

from agentscope.agent import Agent
from agentscope.middleware import TTSMiddleware
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential

agent = Agent(
    name="assistant",
    model=chat_model,
    middlewares=[
        TTSMiddleware(
            DashScopeTTSModel(
                credential=DashScopeCredential(api_key="..."),
                model="qwen3-tts-flash",
                parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
                stream=True,
            ),
        ),
    ],
)

# The agent's reply stream now includes audio events
async for event in agent.reply_stream(user_msg):
    # TextBlockDeltaEvent: text content
    # DataBlockDeltaEvent: audio content (WAV)
    ...

The middleware automatically selects the optimal synthesis strategy:

TTS Mode	Middleware Behavior
Non-realtime	Waits for full text, then synthesizes all at once
Realtime	Pushes text deltas as they arrive, streams audio back concurrently

TTS Model Card

TTSModelCard describes a TTS model’s capabilities (available voices, streaming support, and parameter ranges) and follows the same schema and override mechanics as the general ModelCard. Each card is defined by a YAML file alongside the model implementation. The tabs below show one card per provider:

name: tts-1
label: TTS-1
status: active
input_types:
  - text/plain
output_types:
  - audio/mpeg
  - audio/opus
  - audio/aac
  - audio/flac
  - audio/wav
  - audio/pcm
voices:
  - alloy
  - ash
  - ballad
  - coral
  - echo
  - fable
  - onyx
  - nova
  - sage
  - shimmer
  - verse
parameter_overrides:
  instructions:
    hidden: true

name: gemini-2.5-flash-preview-tts
label: Gemini 2.5 Flash Preview TTS
status: active
input_types:
  - text/plain
output_types:
  - audio/wav
voices:
  - Zephyr
  - Puck
  - Charon
  - Kore
  - Fenrir
  - Leda
  - Orus
  - Aoede
parameter_overrides: {}

name: qwen3-tts-flash
label: Qwen3-TTS-Flash
status: active
input_types:
  - text/plain
output_types:
  - audio/wav
voices:
  - Cherry
  - Serena
  - Ethan
  - Chelsie
parameter_overrides: {}

name: cosyvoice-v3-flash
label: CosyVoice-v3-Flash
status: active
input_types:
  - text/plain
output_types:
  - audio/wav
voices:
  - longanhuan
  - longanyang
  - longhuhu_v3
  - longyingmu_v3
  - longxiaochun_v3
  - longxiaoxia_v3
  - longlaotie_v3
  - longshuo_v3
  - longshu_v3
parameter_overrides: {}

The voices list is automatically injected into the parameter_schema as an enum constraint on the voice field, so the frontend renders a dropdown selector.

Field	Type	Description
`name`	`str`	Model identifier (e.g. `"qwen3-tts-flash"`)
`label`	`str`	Display name (e.g. `"Qwen3-TTS-Flash"`)
`status`	`str`	`"active"`, `"deprecated"`, or `"sunset"`
`realtime`	`bool`	Whether model supports streaming input
`input_types`	`list[str]`	Accepted input MIME types (always `["text/plain"]`)
`output_types`	`list[str]`	Output MIME types (typically `["audio/wav"]`)
`parameter_schema`	`dict`	Merged JSON Schema for the parameter form: base schema from `Parameters` class, enriched with `voices` enum from YAML
`parameters_overrides`	`dict`	Per-model overrides (same syntax as chat model cards)

Retrieve TTS model cards via the credential:

from agentscope.credential import DashScopeCredential, GeminiCredential

cards = DashScopeCredential.list_tts_models()
for card in cards:
    print(f"{card.name} (realtime={card.realtime}): {card.label}")

# Gemini TTS model cards
gemini_cards = GeminiCredential.list_tts_models()

Or directly on the model class:

from agentscope.tts import OpenAITTSModel, GeminiTTSModel, DashScopeTTSModel, DashScopeCosyVoiceTTSModel

# OpenAI TTS models
openai_cards = OpenAITTSModel.list_models()

# Gemini TTS models
gemini_cards = GeminiTTSModel.list_models()

# Qwen3 TTS models
cards = DashScopeTTSModel.list_models()

# CosyVoice models
cosyvoice_cards = DashScopeCosyVoiceTTSModel.list_models()

Custom TTS Provider

To add a new TTS provider, implement a TTSModelBase subclass and register it on the credential:

from typing import Literal, Type, TYPE_CHECKING, AsyncGenerator, Any
from pydantic import BaseModel, Field
from agentscope.tts import TTSModelBase, TTSResponse
from agentscope.credential import CredentialBase

if TYPE_CHECKING:
    from agentscope.tts import TTSModelBase as TTSBase

class MyTTSModel(TTSModelBase):
    class Parameters(BaseModel):
        voice: str = Field(default="default", title="Voice")

    type: Literal["my_tts"] = "my_tts"

    async def synthesize(
        self, text: str | None = None, **kwargs: Any
    ) -> TTSResponse | AsyncGenerator[TTSResponse, None]:
        # Call your provider's API here
        ...

# Register on your credential
class MyCredential(CredentialBase):
    @classmethod
    def get_tts_model_classes(cls) -> list[Type["TTSBase"]]:
        return [MyTTSModel]

Get Started

Building Blocks

Agent as Service

Others

Create TTS Model

Call TTS Model

Realtime TTS (Streaming Input)

Integrate with Agent

TTS Model Card

Custom TTS Provider

​Create TTS Model

​Call TTS Model

​Realtime TTS (Streaming Input)

​Integrate with Agent

​TTS Model Card

​Custom TTS Provider

Create TTS Model

Call TTS Model

Realtime TTS (Streaming Input)

Integrate with Agent

TTS Model Card

Custom TTS Provider