Alibaba Cloud Model Studio:CosyVoice WebSocket API for speech synthesis

Go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

const (
	wsURL      = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/" // WebSocket server endpoint.
	outputFile = "output.mp3"                                        // Output file path.
)

func main() {
	// If you have not configured the API key as an environment variable, you can replace the next line with: apiKey := "your_api_key". We do not recommend hard coding the API key in your code in a production environment to reduce the risk of API key leaks.
	apiKey := os.Getenv("DASHSCOPE_API_KEY")
	// Check and clear the output file.
	if err := clearOutputFile(outputFile); err != nil {
		fmt.Println("Failed to clear output file: ", err)
		return
	}

	// Connect to the WebSocket service.
	conn, err := connectWebSocket(apiKey)
	if err != nil {
		fmt.Println("Failed to connect to WebSocket: ", err)
		return
	}
	defer closeConnection(conn)

	// Start a goroutine to receive results.
	done, taskStarted := startResultReceiver(conn)

	// Send the run-task instruction.
	taskID, err := sendRunTaskCmd(conn)
	if err != nil {
		fmt.Println("Failed to send run-task instruction: ", err)
		return
	}

	// Wait for the task-started event.
	for !*taskStarted {
		time.Sleep(100 * time.Millisecond)
	}

	// Send the text to be synthesized.
	if err := sendContinueTaskCmd(conn, taskID); err != nil {
		fmt.Println("Failed to send text for synthesis: ", err)
		return
	}

	// Send the finish-task instruction.
	if err := sendFinishTaskCmd(conn, taskID); err != nil {
		fmt.Println("Failed to send finish-task instruction: ", err)
		return
	}

	// Wait for the goroutine that receives results to complete.
	<-done
}

var dialer = websocket.DefaultDialer

// Define structs to represent JSON data.
type Header struct {
	Action       string                 `json:"action"`
	TaskID       string                 `json:"task_id"`
	Streaming    string                 `json:"streaming"`
	Event        string                 `json:"event"`
	ErrorCode    string                 `json:"error_code,omitempty"`
	ErrorMessage string                 `json:"error_message,omitempty"`
	Attributes   map[string]interface{} `json:"attributes"`
}

type Payload struct {
	TaskGroup  string     `json:"task_group"`
	Task       string     `json:"task"`
	Function   string     `json:"function"`
	Model      string     `json:"model"`
	Parameters Params     `json:"parameters"`
	Resources  []Resource `json:"resources"`
	Input      Input      `json:"input"`
}

type Params struct {
	TextType   string `json:"text_type"`
	Voice      string `json:"voice"`
	Format     string `json:"format"`
	SampleRate int    `json:"sample_rate"`
	Volume     int    `json:"volume"`
	Rate       int    `json:"rate"`
	Pitch      int    `json:"pitch"`
}

type Resource struct {
	ResourceID   string `json:"resource_id"`
	ResourceType string `json:"resource_type"`
}

type Input struct {
	Text string `json:"text"`
}

type Event struct {
	Header  Header  `json:"header"`
	Payload Payload `json:"payload"`
}

// Connect to the WebSocket service.
func connectWebSocket(apiKey string) (*websocket.Conn, error) {
	header := make(http.Header)
	header.Add("X-DashScope-DataInspection", "enable")
	header.Add("Authorization", fmt.Sprintf("bearer %s", apiKey))
	conn, _, err := dialer.Dial(wsURL, header)
	if err != nil {
		fmt.Println("Failed to connect to WebSocket: ", err)
		return nil, err
	}
	return conn, nil
}

// Send the run-task instruction.
func sendRunTaskCmd(conn *websocket.Conn) (string, error) {
	runTaskCmd, taskID, err := generateRunTaskCmd()
	if err != nil {
		return "", err
	}
	err = conn.WriteMessage(websocket.TextMessage, []byte(runTaskCmd))
	return taskID, err
}

// Generate the run-task instruction.
func generateRunTaskCmd() (string, string, error) {
	taskID := uuid.New().String()
	runTaskCmd := Event{
		Header: Header{
			Action:    "run-task",
			TaskID:    taskID,
			Streaming: "duplex",
		},
		Payload: Payload{
			TaskGroup: "audio",
			Task:      "tts",
			Function:  "SpeechSynthesizer",
			Model:     "cosyvoice-v3-flash",
			Parameters: Params{
				TextType:   "PlainText",
				Voice:      "longanyang",
				Format:     "mp3",
				SampleRate: 22050,
				Volume:     50,
				Rate:       1,
				Pitch:      1,
			},
			Input: Input{},
		},
	}
	runTaskCmdJSON, err := json.Marshal(runTaskCmd)
	return string(runTaskCmdJSON), taskID, err
}

// Send the text to be synthesized.
func sendContinueTaskCmd(conn *websocket.Conn, taskID string) error {
	texts := []string{"A quiet night thought,", "I see the moonlight before my bed.", "I lift my head and watch the moon.", "I lower my head and think of home."}

	for _, text := range texts {
		runTaskCmd, err := generateContinueTaskCmd(text, taskID)
		if err != nil {
			return err
		}

		err = conn.WriteMessage(websocket.TextMessage, []byte(runTaskCmd))
		if err != nil {
			return err
		}
	}

	return nil
}

// Generate the continue-task instruction.
func generateContinueTaskCmd(text string, taskID string) (string, error) {
	runTaskCmd := Event{
		Header: Header{
			Action:    "continue-task",
			TaskID:    taskID,
			Streaming: "duplex",
		},
		Payload: Payload{
			Input: Input{
				Text: text,
			},
		},
	}
	runTaskCmdJSON, err := json.Marshal(runTaskCmd)
	return string(runTaskCmdJSON), err
}

// Start a goroutine to receive results.
func startResultReceiver(conn *websocket.Conn) (chan struct{}, *bool) {
	done := make(chan struct{})
	taskStarted := new(bool)
	*taskStarted = false

	go func() {
		defer close(done)
		for {
			msgType, message, err := conn.ReadMessage()
			if err != nil {
				fmt.Println("Failed to parse server message: ", err)
				return
			}

			if msgType == websocket.BinaryMessage {
				// Process the binary audio stream.
				if err := writeBinaryDataToFile(message, outputFile); err != nil {
					fmt.Println("Failed to write binary data: ", err)
					return
				}
			} else {
				// Process the text message.
				var event Event
				err = json.Unmarshal(message, &event)
				if err != nil {
					fmt.Println("Failed to parse event: ", err)
					continue
				}
				if handleEvent(conn, event, taskStarted) {
					return
				}
			}
		}
	}()

	return done, taskStarted
}

// Handle events.
func handleEvent(conn *websocket.Conn, event Event, taskStarted *bool) bool {
	switch event.Header.Event {
	case "task-started":
		fmt.Println("Received task-started event")
		*taskStarted = true
	case "result-generated":
		// Ignore the result-generated event.
		return false
	case "task-finished":
		fmt.Println("Task finished")
		return true
	case "task-failed":
		handleTaskFailed(event, conn)
		return true
	default:
		fmt.Printf("Unexpected event: %v\n", event)
	}
	return false
}

// Handle the task-failed event.
func handleTaskFailed(event Event, conn *websocket.Conn) {
	if event.Header.ErrorMessage != "" {
		fmt.Printf("Task failed: %s\n", event.Header.ErrorMessage)
	} else {
		fmt.Println("Task failed for an unknown reason")
	}
}

// Close the connection.
func closeConnection(conn *websocket.Conn) {
	if conn != nil {
		conn.Close()
	}
}

// Write binary data to a file.
func writeBinaryDataToFile(data []byte, filePath string) error {
	file, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	defer file.Close()

	_, err = file.Write(data)
	if err != nil {
		return err
	}

	return nil
}

// Send the finish-task instruction.
func sendFinishTaskCmd(conn *websocket.Conn, taskID string) error {
	finishTaskCmd, err := generateFinishTaskCmd(taskID)
	if err != nil {
		return err
	}
	err = conn.WriteMessage(websocket.TextMessage, []byte(finishTaskCmd))
	return err
}

// Generate the finish-task instruction.
func generateFinishTaskCmd(taskID string) (string, error) {
	finishTaskCmd := Event{
		Header: Header{
			Action:    "finish-task",
			TaskID:    taskID,
			Streaming: "duplex",
		},
		Payload: Payload{
			Input: Input{},
		},
	}
	finishTaskCmdJSON, err := json.Marshal(finishTaskCmd)
	return string(finishTaskCmdJSON), err
}

// Clear the output file.
func clearOutputFile(filePath string) error {
	file, err := os.OpenFile(filePath, os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	file.Close()
	return nil
}

C#

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;

class Program {
    // If you have not configured the API key as an environment variable, you can replace the next line with: private const string ApiKey="your_api_key";. We do not recommend hard coding the API key in your code in a production environment to reduce the risk of API key leaks.
    private static readonly string ApiKey = Environment.GetEnvironmentVariable("DASHSCOPE_API_KEY") ?? throw new InvalidOperationException("DASHSCOPE_API_KEY environment variable is not set.");

    // WebSocket server endpoint.
    private const string WebSocketUrl = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/";
    // Output file path.
    private const string OutputFilePath = "output.mp3";

    // WebSocket client.
    private static ClientWebSocket _webSocket = new ClientWebSocket();
    // Cancellation token source.
    private static CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
    // Task ID.
    private static string? _taskId;
    // Whether the task has started.
    private static TaskCompletionSource<bool> _taskStartedTcs = new TaskCompletionSource<bool>();

    static async Task Main(string[] args) {
        try {
            // Clear the output file.
            ClearOutputFile(OutputFilePath);

            // Connect to the WebSocket service.
            await ConnectToWebSocketAsync(WebSocketUrl);

            // Start the task to receive messages.
            Task receiveTask = ReceiveMessagesAsync();

            // Send the run-task instruction.
            _taskId = GenerateTaskId();
            await SendRunTaskCommandAsync(_taskId);

            // Wait for the task-started event.
            await _taskStartedTcs.Task;

            // Continuously send continue-task instructions.
            string[] texts = {
                "A quiet night thought,",
                "I see the moonlight before my bed,",
                "I lift my head and watch the moon,",
                "I lower my head and think of home."
            };
            foreach (string text in texts) {
                await SendContinueTaskCommandAsync(text);
            }

            // Send the finish-task instruction.
            await SendFinishTaskCommandAsync(_taskId);

            // Wait for the receiving task to complete.
            await receiveTask;

            Console.WriteLine("Task finished, connection closed.");
        } catch (OperationCanceledException) {
            Console.WriteLine("Task was canceled.");
        } catch (Exception ex) {
            Console.WriteLine($"An error occurred: {ex.Message}");
        } finally {
            _cancellationTokenSource.Cancel();
            _webSocket.Dispose();
        }
    }

    private static void ClearOutputFile(string filePath) {
        if (File.Exists(filePath)) {
            File.WriteAllText(filePath, string.Empty);
            Console.WriteLine("Output file cleared.");
        } else {
            Console.WriteLine("Output file does not exist, no need to clear.");
        }
    }

    private static async Task ConnectToWebSocketAsync(string url) {
        var uri = new Uri(url);
        if (_webSocket.State == WebSocketState.Connecting || _webSocket.State == WebSocketState.Open) {
            return;
        }

        // Set the header information for the WebSocket connection.
        _webSocket.Options.SetRequestHeader("Authorization", $"bearer {ApiKey}");
        _webSocket.Options.SetRequestHeader("X-DashScope-DataInspection", "enable");

        try {
            await _webSocket.ConnectAsync(uri, _cancellationTokenSource.Token);
            Console.WriteLine("Successfully connected to the WebSocket service.");
        } catch (OperationCanceledException) {
            Console.WriteLine("WebSocket connection was canceled.");
        } catch (Exception ex) {
            Console.WriteLine($"WebSocket connection failed: {ex.Message}");
            throw;
        }
    }

    private static async Task SendRunTaskCommandAsync(string taskId) {
        var command = CreateCommand("run-task", taskId, "duplex", new {
            task_group = "audio",
            task = "tts",
            function = "SpeechSynthesizer",
            model = "cosyvoice-v3-flash",
            parameters = new
            {
                text_type = "PlainText",
                voice = "longanyang",
                format = "mp3",
                sample_rate = 22050,
                volume = 50,
                rate = 1,
                pitch = 1
            },
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("run-task instruction sent.");
    }

    private static async Task SendContinueTaskCommandAsync(string text) {
        if (_taskId == null) {
            throw new InvalidOperationException("Task ID is not initialized.");
        }

        var command = CreateCommand("continue-task", _taskId, "duplex", new {
            input = new {
                text
            }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("continue-task instruction sent.");
    }

    private static async Task SendFinishTaskCommandAsync(string taskId) {
        var command = CreateCommand("finish-task", taskId, "duplex", new {
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("finish-task instruction sent.");
    }

    private static async Task SendJsonMessageAsync(string message) {
        var buffer = Encoding.UTF8.GetBytes(message);
        try {
            await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, _cancellationTokenSource.Token);
        } catch (OperationCanceledException) {
            Console.WriteLine("Message sending was canceled.");
        }
    }

    private static async Task ReceiveMessagesAsync() {
        while (_webSocket.State == WebSocketState.Open) {
            var response = await ReceiveMessageAsync();
            if (response != null) {
                var eventStr = response.RootElement.GetProperty("header").GetProperty("event").GetString();
                switch (eventStr) {
                    case "task-started":
                        Console.WriteLine("Task has started.");
                        _taskStartedTcs.TrySetResult(true);
                        break;
                    case "task-finished":
                        Console.WriteLine("Task has finished.");
                        _cancellationTokenSource.Cancel();
                        break;
                    case "task-failed":
                        Console.WriteLine("Task failed.");
                        _cancellationTokenSource.Cancel();
                        break;
                    default:
                        // result-generated can be handled here.
                        break;
                }
            }
        }
    }

    private static async Task<JsonDocument?> ReceiveMessageAsync() {
        var buffer = new byte[1024 * 4];
        var segment = new ArraySegment<byte>(buffer);

        try {
            WebSocketReceiveResult result = await _webSocket.ReceiveAsync(segment, _cancellationTokenSource.Token);

            if (result.MessageType == WebSocketMessageType.Close) {
                await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", _cancellationTokenSource.Token);
                return null;
            }

            if (result.MessageType == WebSocketMessageType.Binary) {
                // Process binary data.
                Console.WriteLine("Receiving binary data...");

                // Save the binary data to a file.
                using (var fileStream = new FileStream(OutputFilePath, FileMode.Append)) {
                    fileStream.Write(buffer, 0, result.Count);
                }

                return null;
            }

            string message = Encoding.UTF8.GetString(buffer, 0, result.Count);
            return JsonDocument.Parse(message);
        } catch (OperationCanceledException) {
            Console.WriteLine("Message reception was canceled.");
            return null;
        }
    }

    private static string GenerateTaskId() {
        return Guid.NewGuid().ToString("N").Substring(0, 32);
    }

    private static string CreateCommand(string action, string taskId, string streaming, object payload) {
        var command = new {
            header = new {
                action,
                task_id = taskId,
                streaming
            },
            payload
        };

        return JsonSerializer.Serialize(command);
    }
}

PHP

The directory structure of the sample code is as follows:

my-php-project/

├── composer.json

├── vendor/

└── index.php

The content of composer.json is as follows. Determine the version numbers of the dependencies as needed:

{
    "require": {
        "react/event-loop": "^1.3",
        "react/socket": "^1.11",
        "react/stream": "^1.2",
        "react/http": "^1.1",
        "ratchet/pawl": "^0.4"
    },
    "autoload": {
        "psr-4": {
            "App\\": "src/"
        }
    }
}

The content of index.php is as follows:

<?php

require __DIR__ . '/vendor/autoload.php';

use Ratchet\Client\Connector;
use React\EventLoop\Loop;
use React\Socket\Connector as SocketConnector;

# If you have not configured the API key as an environment variable, you can replace the next line with: $api_key="your_api_key";. We do not recommend hard coding the API key in your code in a production environment to reduce the risk of API key leaks.
$api_key = getenv("DASHSCOPE_API_KEY");
$websocket_url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/'; // WebSocket server endpoint.
$output_file = 'output.mp3'; // Output file path.

$loop = Loop::get();

if (file_exists($output_file)) {
    // Clear the file content.
    file_put_contents($output_file, '');
}

// Create a custom connector.
$socketConnector = new SocketConnector($loop, [
    'tcp' => [
        'bindto' => '0.0.0.0:0',
    ],
    'tls' => [
        'verify_peer' => false,
        'verify_peer_name' => false,
    ],
]);

$connector = new Connector($loop, $socketConnector);

$headers = [
    'Authorization' => 'bearer ' . $api_key,
    'X-DashScope-DataInspection' => 'enable'
];

$connector($websocket_url, [], $headers)->then(function ($conn) use ($loop, $output_file) {
    echo "Connected to WebSocket server\n";

    // Generate a task ID.
    $taskId = generateTaskId();

    // Send the run-task instruction.
    sendRunTaskMessage($conn, $taskId);

    // Define the function to send continue-task instructions.
    $sendContinueTask = function() use ($conn, $loop, $taskId) {
        // Text to be sent.
        $texts = ["A quiet night thought,", "I see the moonlight before my bed,", "I lift my head and watch the moon,", "I lower my head and think of home."];
        $continueTaskCount = 0;
        foreach ($texts as $text) {
            $continueTaskMessage = json_encode([
                "header" => [
                    "action" => "continue-task",
                    "task_id" => $taskId,
                    "streaming" => "duplex"
                ],
                "payload" => [
                    "input" => [
                        "text" => $text
                    ]
                ]
            ]);
            echo "Preparing to send continue-task instruction: " . $continueTaskMessage . "\n";
            $conn->send($continueTaskMessage);
            $continueTaskCount++;
        }
        echo "Number of continue-task instructions sent: " . $continueTaskCount . "\n";

        // Send the finish-task instruction.
        sendFinishTaskMessage($conn, $taskId);
    };

    // Flag to indicate if the task-started event has been received.
    $taskStarted = false;

    // Listen for messages.
    $conn->on('message', function($msg) use ($conn, $sendContinueTask, $loop, &$taskStarted, $taskId, $output_file) {
        if ($msg->isBinary()) {
            // Write binary data to a local file.
            file_put_contents($output_file, $msg->getPayload(), FILE_APPEND);
        } else {
            // Process non-binary messages.
            $response = json_decode($msg, true);

            if (isset($response['header']['event'])) {
                handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, $taskStarted);
            } else {
                echo "Unknown message format\n";
            }
        }
    });

    // Listen for connection closure.
    $conn->on('close', function($code = null, $reason = null) {
        echo "Connection closed\n";
        if ($code !== null) {
            echo "Close code: " . $code . "\n";
        }
        if ($reason !== null) {
            echo "Close reason: " . $reason . "\n";
        }
    });
}, function ($e) {
    echo "Could not connect: {$e->getMessage()}\n";
});

$loop->run();

/**
 * Generate a task ID.
 * @return string
 */
function generateTaskId(): string {
    return bin2hex(random_bytes(16));
}

/**
 * Send the run-task instruction.
 * @param $conn
 * @param $taskId
 */
function sendRunTaskMessage($conn, $taskId) {
    $runTaskMessage = json_encode([
        "header" => [
            "action" => "run-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "task_group" => "audio",
            "task" => "tts",
            "function" => "SpeechSynthesizer",
            "model" => "cosyvoice-v3-flash",
            "parameters" => [
                "text_type" => "PlainText",
                "voice" => "longanyang",
                "format" => "mp3",
                "sample_rate" => 22050,
                "volume" => 50,
                "rate" => 1,
                "pitch" => 1
            ],
            "input" => (object) []
        ]
    ]);
    echo "Preparing to send run-task instruction: " . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task instruction sent\n";
}

/**
 * Read an audio file.
 * @param string $filePath
 * @return bool|string
 */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
        echo "Cannot read audio file\n";
    }
    return $voiceData;
}

/**
 * Split audio data.
 * @param string $data
 * @param int $chunkSize
 * @return array
 */
function splitAudioData(string $data, int $chunkSize): array {
    return str_split($data, $chunkSize);
}

/**
 * Send the finish-task instruction.
 * @param $conn
 * @param $taskId
 */
function sendFinishTaskMessage($conn, $taskId) {
    $finishTaskMessage = json_encode([
        "header" => [
            "action" => "finish-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "input" => (object) []
        ]
    ]);
    echo "Preparing to send finish-task instruction: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task instruction sent\n";
}

/**
 * Handle events.
 * @param $conn
 * @param $response
 * @param $sendContinueTask
 * @param $loop
 * @param $taskId
 * @param $taskStarted
 */
function handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, &$taskStarted) {
    switch ($response['header']['event']) {
        case 'task-started':
            echo "Task started, sending continue-task instruction...\n";
            $taskStarted = true;
            // Send the continue-task instruction.
            $sendContinueTask();
            break;
        case 'result-generated':
            // Ignore the result-generated event.
            break;
        case 'task-finished':
            echo "Task finished\n";
            $conn->close();
            break;
        case 'task-failed':
            echo "Task failed\n";
            echo "Error code: " . $response['header']['error_code'] . "\n";
            echo "Error message: " . $response['header']['error_message'] . "\n";
            $conn->close();
            break;
        case 'error':
            echo "Error: " . $response['payload']['message'] . "\n";
            break;
        default:
            echo "Unknown event: " . $response['header']['event'] . "\n";
            break;
    }

    // If the task is finished, close the connection.
    if ($response['header']['event'] == 'task-finished') {
        // Wait for 1 second to ensure all data has been transmitted.
        $loop->addTimer(1, function() use ($conn) {
            $conn->close();
            echo "Client closed connection\n";
        });
    }

    // If the task-started event is not received, close the connection.
    if (!$taskStarted && in_array($response['header']['event'], ['task-failed', 'error'])) {
        $conn->close();
    }
}

Node.js

Install the required dependencies:

npm install ws
npm install uuid

The sample code is as follows:

const WebSocket = require('ws');
const fs = require('fs');
const uuid = require('uuid').v4;

// If you have not configured the API key as an environment variable, you can replace the next line with: apiKey = 'your_api_key'. We do not recommend hard coding the API key in your code in a production environment to reduce the risk of API key leaks.
const apiKey = process.env.DASHSCOPE_API_KEY;
// WebSocket server endpoint.
const url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/';
// Output file path.
const outputFilePath = 'output.mp3';

// Clear the output file.
fs.writeFileSync(outputFilePath, '');

// Create a WebSocket client.
const ws = new WebSocket(url, {
  headers: {
    Authorization: `bearer ${apiKey}`,
    'X-DashScope-DataInspection': 'enable'
  }
});

let taskStarted = false;
let taskId = uuid();

ws.on('open', () => {
  console.log('Connected to WebSocket server');

  // Send the run-task instruction.
  const runTaskMessage = JSON.stringify({
    header: {
      action: 'run-task',
      task_id: taskId,
      streaming: 'duplex'
    },
    payload: {
      task_group: 'audio',
      task: 'tts',
      function: 'SpeechSynthesizer',
      model: 'cosyvoice-v3-flash',
      parameters: {
        text_type: 'PlainText',
        voice: 'longanyang', // Voice
        format: 'mp3', // Audio format
        sample_rate: 22050, // Sample rate
        volume: 50, // Volume
        rate: 1, // Speech rate
        pitch: 1 // Pitch
      },
      input: {}
    }
  });
  ws.send(runTaskMessage);
  console.log('run-task message sent');
});

const fileStream = fs.createWriteStream(outputFilePath, { flags: 'a' });
ws.on('message', (data, isBinary) => {
  if (isBinary) {
    // Write binary data to the file.
    fileStream.write(data);
  } else {
    const message = JSON.parse(data);

    switch (message.header.event) {
      case 'task-started':
        taskStarted = true;
        console.log('Task has started');
        // Send continue-task instructions.
        sendContinueTasks(ws);
        break;
      case 'task-finished':
        console.log('Task has finished');
        ws.close();
        fileStream.end(() => {
          console.log('File stream closed');
        });
        break;
      case 'task-failed':
        console.error('Task failed: ', message.header.error_message);
        ws.close();
        fileStream.end(() => {
          console.log('File stream closed');
        });
        break;
      default:
        // You can handle result-generated here.
        break;
    }
  }
});

function sendContinueTasks(ws) {
  const texts = [
    'A quiet night thought,',
    'I see the moonlight before my bed.',
    'I lift my head and watch the moon,',
    'I lower my head and think of home.'
  ];
  
  texts.forEach((text, index) => {
    setTimeout(() => {
      if (taskStarted) {
        const continueTaskMessage = JSON.stringify({
          header: {
            action: 'continue-task',
            task_id: taskId,
            streaming: 'duplex'
          },
          payload: {
            input: {
              text: text
            }
          }
        });
        ws.send(continueTaskMessage);
        console.log(`continue-task sent, text: ${text}`);
      }
    }, index * 1000); // Send every 1 second.
  });

  // Send the finish-task instruction.
  setTimeout(() => {
    if (taskStarted) {
      const finishTaskMessage = JSON.stringify({
        header: {
          action: 'finish-task',
          task_id: taskId,
          streaming: 'duplex'
        },
        payload: {
          input: {}
        }
      });
      ws.send(finishTaskMessage);
      console.log('finish-task sent');
    }
  }, texts.length * 1000 + 1000); // Send 1 second after all continue-task instructions are sent.
}

ws.on('close', () => {
  console.log('Disconnected from WebSocket server');
});

Java

If you use the Java programming language, we recommend that you use the Java DashScope SDK for development. For more information, see Java SDK.

The following is a Java WebSocket call example. Before you run the example, make sure you have imported the following dependencies:

• Java-WebSocket

• jackson-databind

We recommend that you use Maven or Gradle to manage dependency packages. The configurations are as follows:

<dependencies>
    <!-- WebSocket Client -->
    <dependency>
        <groupId>org.java-websocket</groupId>
        <artifactId>Java-WebSocket</artifactId>
        <version>1.5.3</version>
    </dependency>

    <!-- JSON Processing -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.13.0</version>
    </dependency>
</dependencies>

// Other code omitted.
dependencies {
  // WebSocket Client
  implementation 'org.java-websocket:Java-WebSocket:1.5.3'
  // JSON Processing
  implementation 'com.fasterxml.jackson.core:jackson-databind:2.13.0'
}
// Other code omitted.

The Java code is as follows:

import com.fasterxml.jackson.databind.ObjectMapper;

import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.*;

public class TTSWebSocketClient extends WebSocketClient {
    private final String taskId = UUID.randomUUID().toString();
    private final String outputFile = "output_" + System.currentTimeMillis() + ".mp3";
    private boolean taskFinished = false;

    public TTSWebSocketClient(URI serverUri, Map<String, String> headers) {
        super(serverUri, headers);
    }

    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        System.out.println("Connection successful");

        // Send the run-task instruction.
        String runTaskCommand = "{ \"header\": { \"action\": \"run-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"task_group\": \"audio\", \"task\": \"tts\", \"function\": \"SpeechSynthesizer\", \"model\": \"cosyvoice-v3-flash\", \"parameters\": { \"text_type\": \"PlainText\", \"voice\": \"longanyang\", \"format\": \"mp3\", \"sample_rate\": 22050, \"volume\": 50, \"rate\": 1, \"pitch\": 1 }, \"input\": {} }}";
        send(runTaskCommand);
    }

    @Override
    public void onMessage(String message) {
        System.out.println("Received message from server: " + message);
        try {
            // Parse JSON message
            Map<String, Object> messageMap = new ObjectMapper().readValue(message, Map.class);

            if (messageMap.containsKey("header")) {
                Map<String, Object> header = (Map<String, Object>) messageMap.get("header");

                if (header.containsKey("event")) {
                    String event = (String) header.get("event");

                    if ("task-started".equals(event)) {
                        System.out.println("Received task-started event from server");

                        List<String> texts = Arrays.asList(
                                "A quiet night thought, I see the moonlight before my bed",
                                "I lift my head and watch the moon, I lower my head and think of home"
                        );

                        for (String text : texts) {
                            // Send the continue-task instruction.
                            sendContinueTask(text);
                        }

                        // Send the finish-task instruction.
                        sendFinishTask();
                    } else if ("task-finished".equals(event)) {
                        System.out.println("Received task-finished event from server");
                        taskFinished = true;
                        closeConnection();
                    } else if ("task-failed".equals(event)) {
                        System.out.println("Task failed: " + message);
                        closeConnection();
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("An exception occurred: " + e.getMessage());
        }
    }

    @Override
    public void onMessage(ByteBuffer message) {
        System.out.println("Size of binary audio data received: " + message.remaining());

        try (FileOutputStream fos = new FileOutputStream(outputFile, true)) {
            byte[] buffer = new byte[message.remaining()];
            message.get(buffer);
            fos.write(buffer);
            System.out.println("Audio data has been written to the local file " + outputFile);
        } catch (IOException e) {
            System.err.println("Failed to write audio data to local file: " + e.getMessage());
        }
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("Connection closed: " + reason + " (" + code + ")");
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("Error: " + ex.getMessage());
        ex.printStackTrace();
    }

    private void sendContinueTask(String text) {
        String command = "{ \"header\": { \"action\": \"continue-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": { \"text\": \"" + text + "\" } }}";
        send(command);
    }

    private void sendFinishTask() {
        String command = "{ \"header\": { \"action\": \"finish-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": {} }}";
        send(command);
    }

    private void closeConnection() {
        if (!isClosed()) {
            close();
        }
    }

    public static void main(String[] args) {
        try {
            String apiKey = System.getenv("DASHSCOPE_API_KEY");
            if (apiKey == null || apiKey.isEmpty()) {
                System.err.println("Set the DASHSCOPE_API_KEY environment variable");
                return;
            }

            Map<String, String> headers = new HashMap<>();
            headers.put("Authorization", "bearer " + apiKey);
            TTSWebSocketClient client = new TTSWebSocketClient(new URI("wss://dashscope.aliyuncs.com/api-ws/v1/inference/"), headers);

            client.connect();

            while (!client.isClosed() && !client.taskFinished) {
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            System.err.println("Failed to connect to WebSocket service: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

Python

If you use the Python programming language, we recommend that you use the Python DashScope SDK for development. For more information, see Python SDK.

The following is a Python WebSocket call example. Before you run the example, make sure to import the dependencies as follows:

pip uninstall websocket-client
pip uninstall websocket
pip install websocket-client

import websocket
import json
import uuid
import os
import time


class TTSClient:
    def __init__(self, api_key, uri):
        """
    Initializes the TTSClient instance.

    Parameters:
        api_key (str): The API key for authentication.
        uri (str): The WebSocket service endpoint.
    """
        self.api_key = api_key  # Replace with your API key.
        self.uri = uri  # Replace with your WebSocket endpoint.
        self.task_id = str(uuid.uuid4())  # Generate a unique task ID.
        self.output_file = f"output_{int(time.time())}.mp3"  # Output audio file path.
        self.ws = None  # WebSocketApp instance.
        self.task_started = False  # Whether task-started has been received.
        self.task_finished = False  # Whether task-finished or task-failed has been received.

    def on_open(self, ws):
        """
    Callback function for when the WebSocket connection is established.
    Sends a run-task instruction to start the speech synthesis task.
    """
        print("WebSocket connected")

        # Construct the run-task instruction.
        run_task_cmd = {
            "header": {
                "action": "run-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "task_group": "audio",
                "task": "tts",
                "function": "SpeechSynthesizer",
                "model": "cosyvoice-v3-flash",
                "parameters": {
                    "text_type": "PlainText",
                    "voice": "longanyang",
                    "format": "mp3",
                    "sample_rate": 22050,
                    "volume": 50,
                    "rate": 1,
                    "pitch": 1
                },
                "input": {}
            }
        }

        # Send the run-task instruction.
        ws.send(json.dumps(run_task_cmd))
        print("run-task instruction sent")

    def on_message(self, ws, message):
        """
    Callback function for receiving messages.
    Handles text and binary messages differently.
    """
        if isinstance(message, str):
            # Process JSON text messages.
            try:
                msg_json = json.loads(message)
                print(f"Received JSON message: {msg_json}")

                if "header" in msg_json:
                    header = msg_json["header"]

                    if "event" in header:
                        event = header["event"]

                        if event == "task-started":
                            print("Task has started")
                            self.task_started = True

                            # Send continue-task instructions.
                            texts = [
                                "A quiet night thought, I see the moonlight before my bed",
                                "I lift my head and watch the moon, I lower my head and think of home"
                            ]

                            for text in texts:
                                self.send_continue_task(text)

                            # Send finish-task after all continue-task instructions are sent.
                            self.send_finish_task()

                        elif event == "task-finished":
                            print("Task has finished")
                            self.task_finished = True
                            self.close(ws)

                        elif event == "task-failed":
                            error_msg = msg_json.get("error_message", "Unknown error")
                            print(f"Task failed: {error_msg}")
                            self.task_finished = True
                            self.close(ws)

            except json.JSONDecodeError as e:
                print(f"JSON parsing failed: {e}")
        else:
            # Process binary messages (audio data).
            print(f"Received binary message, size: {len(message)} bytes")
            with open(self.output_file, "ab") as f:
                f.write(message)
            print(f"Audio data has been written to the local file {self.output_file}")

    def on_error(self, ws, error):
        """Callback for when an error occurs."""
        print(f"WebSocket error: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        """Callback for when the connection is closed."""
        print(f"WebSocket closed: {close_msg} ({close_status_code})")

    def send_continue_task(self, text):
        """Sends a continue-task instruction with the text to be synthesized."""
        cmd = {
            "header": {
                "action": "continue-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "input": {
                    "text": text
                }
            }
        }

        self.ws.send(json.dumps(cmd))
        print(f"continue-task instruction sent, text: {text}")

    def send_finish_task(self):
        """Sends a finish-task instruction to end the speech synthesis task."""
        cmd = {
            "header": {
                "action": "finish-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "input": {}
            }
        }

        self.ws.send(json.dumps(cmd))
        print("finish-task instruction sent")

    def close(self, ws):
        """Actively closes the connection."""
        if ws and ws.sock and ws.sock.connected:
            ws.close()
            print("Connection actively closed")

    def run(self):
        """Starts the WebSocket client."""
        # Set request headers (for authentication).
        header = {
            "Authorization": f"bearer {self.api_key}",
            "X-DashScope-DataInspection": "enable"
        }

        # Create a WebSocketApp instance.
        self.ws = websocket.WebSocketApp(
            self.uri,
            header=header,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )

        print("Listening for WebSocket messages...")
        self.ws.run_forever()  # Start the persistent connection listener.


# Example usage
if __name__ == "__main__":
    API_KEY = os.environ.get("DASHSCOPE_API_KEY")  # If you have not configured the API key as an environment variable, set API_KEY to your API key.
    SERVER_URI = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"  # Replace with your WebSocket endpoint.

    client = TTSClient(API_KEY, SERVER_URI)
    client.run()