实时音频

// npm install ws；node tts.mjs
import WebSocket from 'ws';
import { writeFileSync } from 'node:fs';
import { randomUUID } from 'node:crypto';

const key = process.env.TOKENTOSEA_API_KEY;
const model = 'qwen3-tts-flash-realtime';
const requestKey = randomUUID();
const audio = [];

const ws = new WebSocket(`wss://api.tokentosea.com/v1/realtime?model=${encodeURIComponent(model)}`, {
  headers: { Authorization: `Bearer ${key}`, 'Idempotency-Key': requestKey },
});
const send = (event) => ws.send(JSON.stringify({ event_id: `evt-${randomUUID()}`, ...event }));

ws.on('open', () => {
  send({
    type: 'session.update',
    session: { voice: 'Cherry', mode: 'commit', language_type: 'Chinese', response_format: 'pcm', sample_rate: 24000 },
  });
});

ws.on('message', (raw, isBinary) => {
  if (isBinary) return;
  const event = JSON.parse(raw.toString());
  if (event.type === 'session.updated') {
    send({ type: 'input_text_buffer.append', text: '你好，欢迎使用 TokenToSea 实时语音合成。' });
    send({ type: 'input_text_buffer.commit' });
    send({ type: 'session.finish' });
  } else if (event.type?.endsWith('audio.delta') && typeof event.delta === 'string') {
    audio.push(Buffer.from(event.delta, 'base64'));
  } else if (event.type === 'error') {
    console.error('会话错误：', event.error);
  } else if (event.type === 'tokentosea.session.closed') {
    console.log('会话结果：', event.result?.status, event.result?.billing_status);
  }
});

ws.on('error', (error) => console.error('连接失败：', error.message));

ws.on('close', async (code) => {
  writeFileSync('speech.pcm', Buffer.concat(audio));
  // 无论是否收到 tokentosea.session.closed，都以结果查询为准
  const result = await fetch(`https://api.tokentosea.com/v1/results/${requestKey}`, {
    headers: { Authorization: `Bearer ${key}` },
  }).then((response) => response.json());
  console.log('关闭码', code, '最终状态', result.status, result.billing_status ?? '');
});

# pip install "websockets>=14" requests；python tts.py
import asyncio
import base64
import json
import os
import uuid

import requests
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed, InvalidStatus

KEY = os.environ["TOKENTOSEA_API_KEY"]
MODEL = "qwen3-tts-flash-realtime"


def event(payload):
    return json.dumps({"event_id": f"evt-{uuid.uuid4()}", **payload}, ensure_ascii=False)


async def main():
    request_key = str(uuid.uuid4())
    audio = bytearray()
    url = f"wss://api.tokentosea.com/v1/realtime?model={MODEL}"
    headers = {"Authorization": f"Bearer {KEY}", "Idempotency-Key": request_key}
    try:
        async with connect(url, additional_headers=headers) as ws:
            await ws.send(event({
                "type": "session.update",
                "session": {"voice": "Cherry", "mode": "commit", "language_type": "Chinese",
                            "response_format": "pcm", "sample_rate": 24000},
            }))
            async for raw in ws:
                if isinstance(raw, bytes):
                    continue
                message = json.loads(raw)
                kind = message.get("type", "")
                if kind == "session.updated":
                    await ws.send(event({"type": "input_text_buffer.append", "text": "你好，欢迎使用 TokenToSea 实时语音合成。"}))
                    await ws.send(event({"type": "input_text_buffer.commit"}))
                    await ws.send(event({"type": "session.finish"}))
                elif kind.endswith("audio.delta") and isinstance(message.get("delta"), str):
                    audio += base64.b64decode(message["delta"])
                elif kind == "error":
                    print("会话错误：", message.get("error"))
                elif kind == "tokentosea.session.closed":
                    print("会话结果：", message["result"].get("status"))
    except InvalidStatus as rejected:
        print("握手被拒绝：HTTP", rejected.response.status_code)
    except ConnectionClosed as closed:
        print("连接关闭：", closed.rcvd.code if closed.rcvd else None)

    with open("speech.pcm", "wb") as output:
        output.write(audio)
    # 无论是否收到 tokentosea.session.closed，都以结果查询为准
    result = requests.get(
        f"https://api.tokentosea.com/v1/results/{request_key}",
        headers={"Authorization": f"Bearer {KEY}"},
        timeout=30,
    ).json()
    print("最终状态：", result.get("status"), result.get("billing_status"))


asyncio.run(main())