接入文档

curl -s /v1/tts \
  -H "Authorization: Bearer sk_live_你的平台密钥" \
  -H "Content-Type: application/json" \
  -d '{ "voice_id": "voice_xxxx", "text": "今天天气不错，适合出门走走。", "format": "mp3", "sample_rate": 24000, "speed_ratio": 1.0, "volume_ratio": 1.0 }' \
  --output speech.mp3 -w '%{http_code}\n'

import requests

resp = requests.post(
    "/v1/tts",
    headers={"Authorization": "Bearer sk_live_你的平台密钥"},
    json={
  "voice_id": "voice_xxxx",
  "text": "今天天气不错，适合出门走走。",
  "format": "mp3",
  "sample_rate": 24000,
  "speed_ratio": 1.0,
  "volume_ratio": 1.0
},
    timeout=30,
)
resp.raise_for_status()
print("计费字数:", resp.headers.get("X-Word-Count"))
open("speech.mp3", "wb").write(resp.content)

package main

import (
	"bytes"
	"io"
	"net/http"
	"os"
)

func main() {
	payload := []byte(`{
  "voice_id": "voice_xxxx",
  "text": "今天天气不错，适合出门走走。",
  "format": "mp3",
  "sample_rate": 24000,
  "speed_ratio": 1.0,
  "volume_ratio": 1.0
}`)
	req, _ := http.NewRequest("POST", "/v1/tts", bytes.NewReader(payload))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer sk_live_你的平台密钥")

	res, err := http.DefaultClient.Do(req)
	if err != nil {
		panic(err)
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(res.Body)
		panic(string(b)) // {"code":1002,"message":"余额不足"} 之类的信封
	}
	audio, _ := io.ReadAll(res.Body)
	_ = os.WriteFile("speech.mp3", audio, 0o600)
	_, _ = io.WriteString(os.Stderr, "word_count="+res.Header.Get("X-Word-Count")+"\n")
}

const res = await fetch('/v1/tts', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    Authorization: 'Bearer sk_live_你的平台密钥',
  },
  body: JSON.stringify({ "voice_id": "voice_xxxx", "text": "今天天气不错，适合出门走走。", "format": "mp3", "sample_rate": 24000, "speed_ratio": 1.0, "volume_ratio": 1.0 }),
});
if (!res.ok) throw new Error(await res.text());
console.log('计费字数:', res.headers.get('x-word-count'));
await writeFile('speech.mp3', Buffer.from(await res.arrayBuffer()));

curl -s /v1/audio/speech \
  -H "Authorization: Bearer sk_live_你的平台密钥" \
  -H "Content-Type: application/json" \
  -d '{"model":"tts-1","input":"今天天气不错，适合出门走走。","voice":"voice_xxxx","response_format":"mp3","speed":1.0}' \
  --output speech.mp3

# 回执头与 /v1/tts 同源：curl -i 能看到 X-Word-Count / X-Word-Count-Source / X-Request-Id

from openai import OpenAI

client = OpenAI(
    api_key="sk_live_你的平台密钥",          # 平台密钥，不是 OpenAI 的 sk-
    base_url="https://你的域名/v1",   # SDK 自己拼出 /audio/speech
)

with client.audio.speech.with_streaming_response.create(
    model="tts-1",                          # 收下但不参与计费与路由
    voice="voice_xxxx",                     # 平台音色 ID，不是 alloy / nova
    input="今天天气不错，适合出门走走。",
    response_format="mp3",                  # 只支持 mp3 / wav / pcm
) as r:
    r.save("speech.mp3")

# AstrBot 的 TTS 里选内置的 openai_tts_api（源码里读的键名就是这几个），填：
#
#   api_base:  https://你的域名/v1     # SDK 拼成 {api_base}/audio/speech
#   api_key:   sk_live_你的平台密钥
#   model:     tts-1                  # 我们按音色计费，这个值不参与任何判断
#   openai-tts-voice:  voice_xxxx     # 注意：这里填平台音色 ID，不是 alloy / nova
#
# 它固定用 response_format="wav" 落盘，本服务的 wav 是真 wav（RIFF 头完整），直接播即可。
# 想换密钥或音色只改这两行；不必为它写任何适配代码。

curl -N -s /v1/tts/stream \
  -H "Authorization: Bearer sk_live_你的平台密钥" \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{"voice_id":"voice_xxxx","text":"流式合成一段话","format":"mp3","sample_rate":24000}'

# event: audio
# data: {"chunk_id":1,"data":"<base64>"}
# ...
# event: done
# data: {"status":"ok","word_count":11,"word_count_source":"upstream"}

// 一条连接 = 一次合成；文本可以分多段送，音频边收边播。
const ws = new WebSocket('ws://localhost:8080/v1/tts/stream/ws?token=sk_live_你的平台密钥');
ws.binaryType = 'arraybuffer';
const chunks = [];                       // 收完 done 后拼起来就是整段音频

ws.onopen = () => {
  // config 必须第一帧、5 秒内到达；format 只开放 mp3 / pcm
  ws.send(JSON.stringify({ type: 'config', voice_id: 'voice_xxxx', format: 'mp3', sample_rate: 24000 }));
  ws.send(JSON.stringify({ type: 'text', text: '第一句先合成。' }));
  ws.send(JSON.stringify({ type: 'text', text: '第二句接着来。' }));
  ws.send(JSON.stringify({ type: 'finish' }));   // 文本发完了，接下来只收
};

ws.onmessage = (ev) => {
  if (typeof ev.data !== 'string') {              // 二进制 = 原始音频分片（不是 base64）
    chunks.push(ev.data);
    return;
  }
  const f = JSON.parse(ev.data);
  if (f.type === 'sentence_start' || f.type === 'sentence_end') console.log(f.type, f.seq, f.text);
  if (f.type === 'done') console.log('计费字数', f.word_count, f.word_count_source, '取消', f.canceled);
  if (f.type === 'error') console.error(f.code, f.message);
};

# tts_duplex.py —— 边发文本边收音频；依赖：pip install websockets
import asyncio, json
import websockets

WS = "ws://localhost:8080/v1/tts/stream/ws?token=sk_live_你的平台密钥"

async def main():
    async with websockets.connect(WS) as ws:
        # config 必须第一帧、5 秒内到达
        await ws.send(json.dumps({
            "type": "config", "voice_id": "voice_xxxx", "format": "mp3", "sample_rate": 24000,
        }))
        for text in ["第一句先合成。", "第二句接着来。"]:
            await ws.send(json.dumps({"type": "text", "text": text}))
        await ws.send(json.dumps({"type": "finish"}))   # 文本发完了，接下来只收

        audio = bytearray()
        async for msg in ws:
            if isinstance(msg, bytes):       # 原始音频分片，直接追加
                audio.extend(msg)
                continue
            f = json.loads(msg)
            if f["type"] == "sentence_start":
                print("开始合成第", f["seq"], "句:", f.get("text", ""))
            elif f["type"] == "done":
                print("计费字数", f["word_count"], f["word_count_source"], "取消", f["canceled"])
                break
            elif f["type"] == "error":
                print("合成失败:", f["code"], f["message"])
                break
        open("speech.mp3", "wb").write(audio)

asyncio.run(main())

const ws = new WebSocket('ws://localhost:8080/v1/asr/stream?token=sk_live_你的平台密钥');

ws.addEventListener('open', () => {
  // 5 秒内必须发 config，否则平台下发 error(1006) 后关闭连接
  ws.send(JSON.stringify({
    type: 'config',
    audio: { format: 'pcm', rate: 16000, bits: 16, channel: 1, codec: 'raw' },
    language: 'zh',
  }));
});

// 上行：每帧 100–200ms 的裸音频字节
// ws.send(pcmChunk);
// 结束：ws.send(JSON.stringify({ type: 'finish' }));

ws.addEventListener('message', (ev) => {
  const frame = JSON.parse(String(ev.data));
  if (frame.type === 'partial') console.log('中间结果', frame.text);
  if (frame.type === 'final') console.log('最终结果', frame.text, frame.duration_ms, frame.duration_source);
  if (frame.type === 'error') console.error(frame.code, frame.message);
});

// 依赖只这一个：go get github.com/gorilla/websocket
package main

import (
	"fmt"
	"log"
	"os"
	"time"

	"github.com/gorilla/websocket"
)

// 下行三种帧（partial / final / error）共用一套字段，用 type 区分。
type frame struct {
	Type           string `json:"type"`
	Text           string `json:"text"`
	Seq            int    `json:"seq"`
	DurationMS     int64  `json:"duration_ms"`
	DurationSource string `json:"duration_source"`
	Code           int    `json:"code"`
	Message        string `json:"message"`
}

func main() {
	// token 走查询参数，与 REST 的 Authorization: Bearer 同权。
	conn, _, err := websocket.DefaultDialer.Dial(
		"ws://localhost:8080/v1/asr/stream?token=sk_live_你的平台密钥", nil)
	if err != nil {
		// token 缺失/无效/账户被限制时，连接在握手完成前就被关掉，看到的正是这一条
		log.Fatal("建连失败: ", err)
	}
	defer conn.Close()

	// 1) 5 秒内发 config；这里声明的格式整条连接期间不可变更。
	if err := conn.WriteJSON(map[string]any{
		"type": "config",
		"audio": map[string]any{
			"format": "pcm", "rate": 16000, "bits": 16, "channel": 1, "codec": "raw",
		},
		"language": "zh",
	}); err != nil {
		log.Fatal("发 config 失败: ", err)
	}

	// 先转成 16k/16bit/单声道：ffmpeg -i in.wav -f s16le -ar 16000 -ac 1 voice.pcm
	pcm, err := os.ReadFile("voice.pcm")
	if err != nil {
		log.Fatal(err)
	}

	go func() {
		// 2) 每帧 3200 字节 = 100ms；单帧上限 65536 字节。
		for off := 0; off < len(pcm); off += 3200 {
			end := off + 3200
			if end > len(pcm) {
				end = len(pcm)
			}
			if err := conn.WriteMessage(websocket.BinaryMessage, pcm[off:end]); err != nil {
				return
			}
			time.Sleep(20 * time.Millisecond) // 略快于实时就行；两帧之间最长静默 60 秒
		}
		conn.WriteJSON(map[string]string{"type": "finish"}) // 3) 说「我说完了」
	}()

	// 4) 一路读到 final 或 error。error 之后不会再补 final，连接随即关闭。
	for {
		var f frame
		if err := conn.ReadJSON(&f); err != nil {
			log.Println("连接结束: ", err)
			return
		}
		switch f.Type {
		case "partial":
			fmt.Println("中间结果:", f.Text)
		case "final":
			// duration_ms 就是账单上那个数，别自己按发送字节猜
			fmt.Printf("最终结果: %s\n计费时长: %dms（来源 %s）\n", f.Text, f.DurationMS, f.DurationSource)
			return
		case "error":
			log.Fatalf("识别失败: %d %s", f.Code, f.Message)
		}
	}
}

# push_asr.py —— 把一份本地 PCM 当流推给识别，拿回整句文本。
# 依赖：pip install websockets
# 音频先转成 16k / 16bit / 单声道的裸 PCM（QQ 的 silk、Telegram 的 ogg 都走这一步，在 shell 里执行）：
#   ffmpeg -i voice.wav -f s16le -ar 16000 -ac 1 voice.pcm
import asyncio, json, sys
import websockets

WS = "ws://localhost:8080/v1/asr/stream?token=sk_live_你的平台密钥"

async def main():
    data = open(sys.argv[1], "rb").read()
    async with websockets.connect(WS) as ws:
        await ws.send(json.dumps({
            "type": "config",
            "audio": {"format": "pcm", "rate": 16000, "bits": 16, "channel": 1, "codec": "raw"},
            "language": "zh",
        }))

        async def pump():
            for i in range(0, len(data), 3200):        # 3200 字节 = 100ms
                await ws.send(data[i:i + 3200])
            await ws.send(json.dumps({"type": "finish"}))

        async def read():
            while True:
                frame = json.loads(await ws.recv())
                if frame["type"] == "final":
                    print(frame["text"], frame["duration_ms"], frame["duration_source"])
                    return
                if frame["type"] == "error":
                    print("识别失败:", frame["code"], frame["message"])
                    return

        await asyncio.gather(pump(), read())

asyncio.run(main())
# python push_asr.py voice.pcm

# ① 提交：只拿任务号，这一步不产生费用
curl -s /v1/asr/file \
  -H "Authorization: Bearer sk_live_你的平台密钥" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: my-meeting-001" \
  -d '{"audio_url":"https://example.com/meeting.mp3","format":"mp3"}'

# {"code":0,"message":"ok","data":{"task_id":"my-meeting-001","status":"pending","replayed":false}}
#   ↑ 任务号就是上面那把键（后端拿它当发给识别侧的请求号）。
#     同一个键重发只会得到同一个任务，不会受理两次、也不会扣两次钱。

# ② 轮询：跑完才有 text 与计费时长
curl -s /v1/asr/file/my-meeting-001 \
  -H "Authorization: Bearer sk_live_你的平台密钥"

# {"code":0,"data":{"task_id":"my-meeting-001","status":"succeeded",
#  "duration_ms":6312,"text":"……",
#  "utterances":[{"start_time":480,"end_time":5920,"text":"……",
#                 "words":[{"start_time":480,"end_time":600,"text":"…"}]}]}}
#   status: pending / processing / succeeded / silent（整段没人说话，成功但不计费）/ failed
#   账单上的时长取这里回传的 duration_ms，不足 1 秒按 1 秒；受理与失败都不计费。

curl -s /v1/audio/transcriptions \
  -H "Authorization: Bearer sk_live_你的平台密钥" \
  -F "file=@voice.wav" \
  -F "language=zh"

# {"text":"刚刚还在想你……"}
#   回执头：X-Usage-Duration-Ms / X-Usage-Duration-Source / X-Request-Id
#   只要纯文本：加一个 -F "response_format=text"
#   逐段出字（长音频想看进度）：加一个 -F "stream=true"，响应变成 SSE

import requests

with open("voice.wav", "rb") as f:
    r = requests.post(
        "/v1/audio/transcriptions",
        headers={"Authorization": "Bearer sk_live_你的平台密钥"},
        files={"file": ("voice.wav", f, "audio/wav")},
        data={"language": "zh"},
        timeout=60,
    )
r.raise_for_status()
print(r.json()["text"])
# 账单上的时长从这里取，别按文件大小自己算：
print(r.headers["X-Usage-Duration-Ms"], r.headers["X-Usage-Duration-Source"])

# AstrBot 的 STT 提供商选内置的 openai_whisper_api，配置项就是它源码里读的那几个：
#
#   provider_type:  openai_whisper_api
#   api_base:  https://你的域名/v1     # 它自己拼出 {api_base}/audio/transcriptions
#   api_key:   sk_live_你的平台密钥
#   model:     任意值即可（本服务按音频时长计费，这个值不参与路由与计费）
#   timeout:   60
#
# 然后在「STT 设置」里把 provider_id 指到上面这条、enable 打开即可。
# 它收到语音后自己落成本地 wav（QQ 的 silk、Telegram 的 ogg 都由它转），
# 再把字节 multipart 传给我们 —— 采样率不用你管，wav / mp3 直接收。

curl -s /v1/voices/clone \
  -H "Authorization: Bearer sk_live_你的平台密钥" \
  -F "name=我的播报音色" \
  -F "audio=@reference.wav"

# {"code":0,"message":"ok","data":{"voice_id":"voice_xxxx","status":"cloning","created_at":"..."}}
curl -s /v1/voices/voice_xxxx -H "Authorization: Bearer sk_live_你的平台密钥"