WebSocket ストリーミング（TTS）

WebSocket TTS API では、完全なテキストを送信し、生成された音声を 1 本の接続でストリーミングして受け取れます。合成が完了する前に再生を開始できます。

wss://api.shisa.ai/ws/tts/realtime

完全なテキストを入力、音声をストリーミングで出力

パス名に反して、これはリアルタイム ASR のような入力ストリーミング API ではありません。完全な tts.speak テキストリクエストを送信すると、サービスが生成された音声をバイナリフレームまたは base64 JSON チャンクとしてストリーミングで返します。

音声ファイル全体を一度に返す 1 回のリクエスト/レスポンスが必要な場合は、代わりに POST /tts エンドポイントを使用してください。

要件

選択した音声を使用できる API キー（WebSocket TTS のクォータとレート制限は shisa/tts サービスで管理されます）。
GET /tts/voices から取得した音声の UUID — 音声カタログを参照してください。
Authorization ヘッダーを設定できるサーバーサイドの WebSocket クライアント。API キーはサーバーサイドに保管してください。

ハンドシェイク時に、標準のベアラートークンで認証します:

Authorization: Bearer YOUR_API_KEY

仕組み

Authorization ヘッダーを付けて WebSocket を開きます。
必須の voice_id と format（および任意の sample_rate、temperature、audio_transport）を含む session.update を 1 回送信します。
session.created を待ちます。
完全なテキストを含む tts.speak リクエストを 1 回送信します。
tts.audio.start、音声フレーム、tts.audio.done、tts.usage を読み取ります。
さらに tts.speak リクエストを順番に送信するか、ソケットをクローズします。

一度に 1 つの合成のみ

1 セッションあたりアクティブにできる合成は 1 つだけです。合成が進行中に 2 つ目の tts.speak を送信すると、synthesis_in_progress エラーが返されます。バックエンドで開始された試行では tts.usage を、バックエンド到達前の拒否では終端となる tts.error を受信してから、次のリクエストを送信してください。backend_request_failed の場合は、直後に続く tts.usage も読み取ります。

クライアントメッセージ

`session.update`

合成の前に session.update を 1 回送信します:

{
  "type": "session.update",
  "id": "cfg_0001",
  "session": {
    "voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
    "format": "mp3",
    "audio_transport": "binary"
  }
}

フィールド	必須	備考
`id`	任意	クライアント相関 ID。設定エラー時にエコーされます。
`voice_id`	必須	`GET /tts/voices` から取得した公開音声の UUID。
`format`	必須	出力音声形式 — `mp3`、`wav`、`ogg`、`pcm`、`flac` のいずれか。選択した音声で対応し、かつそのプロバイダーでストリーミング可能な形式を指定します。
`sample_rate`	任意	省略するか`0`を設定するとバックエンドのデフォルト（`24000` Hz）になります。選択した音声で設定可能な値のみを使用します。Qwen系音声の`ogg`では上書きできません。
`temperature`	任意	現在はQwen系音声で受け付けられる、音声のばらつきを制御するプロバイダー固有のパラメータ。その他の音声およびデフォルト値を使う場合は省略します。明示的な`0.0`は実値として送信されます。
`audio_transport`	任意	`binary`（デフォルト）または `base64_json`。音声トランスポートを参照してください。

形式ごとのストリーミング対応

音声カタログのstreaming: trueは必要条件ですが、すべてのformats値がストリーミング可能という意味ではありません。このエンドポイントは常に出力をストリーミングするため、非対応のプロバイダーと形式の組み合わせにはcode: "unsupported_streaming_format"のtts.errorが返ります。

`tts.speak`

合成する完全なテキストを送信します:

{
  "type": "tts.speak",
  "id": "utt_0001",
  "text": "こんにちは。WebSocket TTS のテストです。"
}

フィールド	必須	備考
`id`	任意	クライアント相関 ID。このリクエストの TTS イベントでエコーされます。
`text`	必須	合成する完全なテキスト。デフォルトの上限は 5000 文字です。

サービスイベント

`session.created`

セッションが設定され、tts.speak の準備が整いました:

{
  "type": "session.created",
  "session": {
    "id": "router-session-request-id",
    "service": "shisa/tts",
    "voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
    "format": "mp3",
    "sample_rate": 24000,
    "model": "speech-2.8-hd"
  }
}

`tts.audio.start`

受け入れられた合成リクエストを示し、受け取る音声を記述します:

{
  "type": "tts.audio.start",
  "id": "utt_0001",
  "request_id": "synthesis-request-id",
  "format": "mp3",
  "content_type": "audio/mpeg",
  "sample_rate": 24000,
  "voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
  "model": "speech-2.8-hd",
  "audio_transport": "binary"
}

音声トランスポート

音声バイトの届き方は、session.update で設定した audio_transport によって異なります:

binary（デフォルト） — 生成された音声は生のバイナリ WebSocket フレームとして届きます。順番に連結してください。
base64_json — 音声は tts.audio.delta JSON メッセージとして届き、base64 エンコードされたバイトが audio に、増加する seq が含まれます:

{
  "type": "tts.audio.delta",
  "id": "utt_0001",
  "request_id": "synthesis-request-id",
  "seq": 1,
  "audio": "<base64-audio-bytes>"
}

`tts.audio.done`

このリクエストの合成が完了しました:

{
  "type": "tts.audio.done",
  "id": "utt_0001",
  "request_id": "synthesis-request-id",
  "audio_bytes": 12345,
  "chunks": 4
}

`tts.usage`

バックエンドで開始された合成試行の最終的な利用量です。成功時は tts.audio.done の後に送信されます:

{
  "type": "tts.usage",
  "request_id": "synthesis-request-id",
  "session_id": "router-session-request-id",
  "client_id": "utt_0001",
  "model": "speech-2.8-hd",
  "usage": {
    "input_chars": 24,
    "input_bytes": 72,
    "audio_bytes": 12345,
    "status": "completed",
    "final": true
  }
}

バックエンドリクエストが失敗した場合は、code: "backend_request_failed" の tts.error に続いて tts.usage が送信されます。音声が返らなかった場合の usage.status は backend_error、一部の音声がすでに返った場合は partial_backend_error です。

バックエンドに到達した試行について、ルーターは失敗時も利用記録を作成します（接続が書き込み可能な間は tts.usage も送信します）。一方、入力検証、アクセス制御、レート制限、バックエンド選択の段階で拒否されたリクエストは tts.error のみを返し、利用記録を作成しません。したがって、すべての tts.error の後に tts.usage が続くわけではありません。

`tts.error` と `router.error`

WebSocket のアップグレード前に失敗した場合は、WebSocketイベントではなく構造化されたHTTPエラーが返ります。代表的なコードは、invalid_websocket_upgrade、auth_token_required、invalid_token_format、invalid_token、legacy_api_key_rejected、ws_origin_denied、unknown_service、ws_connection_limit_exceeded、ws_connection_quota_unavailable、websocket_drainingです。

TTS の設定、発話、バックエンドに関するエラーは次の形です。id は、利用可能な場合にクライアントの相関 ID をエコーします:

{
  "type": "tts.error",
  "id": "utt_0001",
  "code": "backend_request_failed",
  "message": "Backend TTS request failed"
}

コード	意味
`invalid_message`, `invalid_id`, `id_too_long`, `invalid_session`	認識されたメッセージ、相関 ID、またはルーターのセッション状態が不正です。
`session_already_configured`	この接続では `session.update` がすでに正常に処理されています。
`missing_voice_id`, `unknown_voice_id`, `tts_service_unavailable`, `service_access_denied`	音声の選択、サービスの可用性、またはアクセス権に問題があります。
`unsupported_audio_format`, `unsupported_sample_rate`, `unsupported_audio_transport`, `unsupported_streaming_format`	要求した出力設定を、選択した音声またはバックエンドで利用できません。
`session_not_configured`, `empty_text`, `tts_text_too_long`, `synthesis_in_progress`	`tts.speak` の状態またはテキストが不正です。
`rate_limited`, `service_rate_limited`	API キーまたは TTS サービスのレート制限を超えました。
`rate_limiter_unavailable`, `service_rate_limiter_unavailable`	レート制限サービスが一時的に利用できません。
`backend_unavailable`	現在利用できる TTS バックエンドがありません。バックエンドリクエストは開始されていないため、利用記録はありません。
`backend_request_failed`	バックエンドで開始された合成が失敗しました。このエラーの後に最終 `tts.usage` が続きます。
`tts_control_frame_too_large`	JSON制御メッセージが設定済みのバイト上限を超えています。読み取り上限で先に検出された場合、接続はコード`1009`で閉じられます。

メッセージ種別を判別する前の制御エラーは、type: "router.error" と、invalid_json、missing_message_type、unknown_message_type などのコードを使用します。クライアントからバイナリフレームを送信すると接続は WebSocket コード 1003 で閉じられ、制御フレームが大きすぎる場合はコード 1009 で閉じられます。

最小限の Python クライアント

依存関係をインストールします:

python -m pip install websockets

キーを設定して実行します:

export SHISA_API_KEY="shsk:..."
export TTS_WS_URL="wss://api.shisa.ai/ws/tts/realtime"
export TTS_VOICE_ID="61ba1141-60aa-4bc3-a3b3-be1ec20700b3"
python websocket_tts.py > output.mp3

websocket_tts.py:

#!/usr/bin/env python3
import asyncio
import json
import os
import sys

import websockets


async def main() -> None:
    url = os.environ.get("TTS_WS_URL", "wss://api.shisa.ai/ws/tts/realtime")
    api_key = os.environ["SHISA_API_KEY"]
    voice_id = os.environ["TTS_VOICE_ID"]
    text = os.environ.get("TTS_TEXT", "こんにちは。WebSocket TTS のテストです。")
    audio_format = os.environ.get("TTS_FORMAT", "mp3")

    headers = [("Authorization", f"Bearer {api_key}")]
    audio = bytearray()

    async with websockets.connect(url, additional_headers=headers, max_size=None) as ws:
        await ws.send(
            json.dumps(
                {
                    "type": "session.update",
                    "id": "cfg_0001",
                    "session": {
                        "voice_id": voice_id,
                        "format": audio_format,
                        "audio_transport": "binary",
                    },
                }
            )
        )

        # session.created を待ってから発話します。
        while True:
            event = json.loads(await ws.recv())
            if event.get("type") == "session.created":
                break
            if event.get("type") in {"tts.error", "router.error", "error"}:
                raise RuntimeError(event)

        await ws.send(json.dumps({"type": "tts.speak", "id": "utt_0001", "text": text}))

        # バイナリフレームは音声、JSON フレームはイベント。
        backend_error = None
        while True:
            msg = await ws.recv()
            if isinstance(msg, bytes):
                audio.extend(msg)
                continue
            event = json.loads(msg)
            event_type = event.get("type")
            if event_type == "tts.error":
                if event.get("code") == "backend_request_failed":
                    # バックエンド失敗時は、この後に最終利用量が続きます。
                    backend_error = event
                    continue
                raise RuntimeError(event)
            if event_type == "tts.usage":
                print(json.dumps(event, ensure_ascii=False), file=sys.stderr)
                if backend_error is not None:
                    raise RuntimeError(backend_error)
                break
            if event_type in {"router.error", "error"}:
                raise RuntimeError(event)

    sys.stdout.buffer.write(audio)


if __name__ == "__main__":
    asyncio.run(main())

次のステップ

音声ファイル全体を 1 回の呼び出しで返すには POST /tts エンドポイントを使用します。
利用可能な音声は音声カタログで閲覧できます。
利用がどのように課金されるかは料金を参照してください。

要​件​

仕組み​

クライアントメッセージ​

session.update​

tts.speak​

サービスイベント​

session.created​

tts.audio.start​

音声トランスポート​

tts.audio.done​

tts.usage​

tts.error と​ router.error​

最小限の​ Python クライアント​

次の​ステップ​

要件

仕組み