接入准备
- 在 MaiToken 控制台 创建 API Key。
- 确保 Key 所选分组允许使用
qwen-audio-3.0-tts-plus，且可用余额充足。 - 选择当前模型支持的音色。本文使用
longanlingxin，该音色已在本文所述正式环境 HTTP 测试中成功使用；其他音色参见官方音色列表。
示例仅使用环境变量或占位 Key。管理后台登录 Token 不能替代模型调用 Key。
接口地址与调用方式
| 方式 | 正式环境入口 | 输入与输出 |
|---|---|---|
| WebSocket 双向流式 | wss://api.maitoken.com/api-ws/v1/inference |
分段提交文本，接收 JSON 事件和二进制音频帧 |
| HTTP 非实时 | https://api.maitoken.com/api/v1/services/audio/tts/SpeechSynthesizer |
一次提交文本，返回包含音频 URL 的 JSON |
使用完整原生路径，不要额外添加 /v1，也不要把 HTTP 地址直接交给 WebSocket 客户端。本文模型的 WebSocket 路由是 /api-ws/v1/inference，模型名位于首个 run-task 消息中。
/api-ws/v1/realtime?model=... 属于另一套 Qwen-TTS Realtime 会话协议，使用 session.update 等事件，不能与本文的 run-task 协议混用。其他模型是否已在账号中开放，以模型列表及 Key 权限为准。
鉴权
在 WebSocket 握手的 HTTP 请求头或 HTTP 合成请求头中传入：
Authorization: Bearer YOUR_MAITOKEN_API_KEYWebSocket 不需要 X-Api-Resource-Id、X-DashScope-Async 或 X-DashScope-SSE。JSON 控制消息以 WebSocket 文本帧发送，音频以二进制帧接收。
浏览器原生 WebSocket 构造函数不能设置自定义 Authorization 请求头。网页应用应通过自己的服务端接入；不要把 Key 拼进 URL 代替请求头。下方 Python 示例可直接设置握手头。
WebSocket 交互流程
- 建立连接，在握手头中携带 Key。
- 生成新的任务 UUID，发送
run-task。 - 等待服务端返回
task-started，再发送文本。 - 使用一个或多个
continue-task消息依次提交文本。 - 文本提交完毕后发送
finish-task，继续接收剩余音频。 - 收到
task-finished才判定任务成功结束；收到task-failed则按失败处理。
同一任务的所有消息使用同一个 task_id。MaiToken 当前一条 inference 连接只允许一个合成任务；下一次合成重新建立连接。不要照搬上游文档中的同连接多任务复用方式。
1. 启动任务：run-task
{ "header": { "action": "run-task", "task_id": "dbe53a5f-257f-4302-a9c1-1cf1d4790b52", "streaming": "duplex" }, "payload": { "task_group": "audio", "task": "tts", "function": "SpeechSynthesizer", "model": "qwen-audio-3.0-tts-plus", "parameters": { "voice": "longanlingxin", "format": "mp3", "sample_rate": 24000 }, "input": {} }}| 参数 | 类型 | 说明 |
|---|---|---|
header.action |
string | 启动时为run-task |
header.task_id |
string | 每次新任务生成一个 UUID，后续消息沿用 |
header.streaming |
string | 双向流式使用duplex |
payload.task_group |
string | 固定为audio |
payload.task |
string | 固定为tts |
payload.function |
string | 固定为SpeechSynthesizer |
payload.model |
string | 准确的模型 ID，本文为qwen-audio-3.0-tts-plus |
payload.parameters.voice |
string | 当前模型支持且可访问的音色 ID |
payload.parameters.format |
string | 本文选择mp3；使用其他格式时同步调整音频处理方式 |
payload.parameters.sample_rate |
integer | 本文选择24000 Hz，支持范围以模型文档为准 |
payload.input |
object | 启动双向流式任务时传{}，文本通过后续消息提交 |
模型和音色在启动时确定，不能在后续 continue-task 中切换。音色权限受当前上游资源影响；自定义音色应先通过 MaiToken 对应音色管理接口创建并登记。
2. 提交文本：continue-task
收到 task-started 后发送：
{ "header": { "action": "continue-task", "task_id": "dbe53a5f-257f-4302-a9c1-1cf1d4790b52", "streaming": "duplex" }, "payload": { "input": { "text": "你好，欢迎使用 MaiToken。" } }}后续文本可以继续发送相同结构的消息。建议按自然短句发送；生产应用需要同时接收音频，避免仅发送而不读取导致缓冲积压。不要假定每次提交文本恰好对应一个音频帧。
3. 完成输入：finish-task
{ "header": { "action": "finish-task", "task_id": "dbe53a5f-257f-4302-a9c1-1cf1d4790b52", "streaming": "duplex" }, "payload": { "input": {} }}finish-task 表示文本已发送完毕，不表示音频已经接收完毕。此时不要立即断开连接。
响应与音频还原
| 返回内容 | 含义与处理 |
|---|---|
JSON：header.event=task-started |
任务已启动，可以提交文本 |
| 二进制帧 | 音频片段，按接收顺序写入文件或交给解码器 |
JSON：header.event=result-generated |
合成过程中的结果或用量事件，按需读取payload |
JSON：header.event=task-finished |
合成成功结束，可关闭连接并完成文件保存 |
JSON：header.event=task-failed |
读取header.error_code、header.error_message 排查 |
完成事件的结构示意：
{ "header": { "task_id": "dbe53a5f-257f-4302-a9c1-1cf1d4790b52", "event": "task-finished" }, "payload": { "usage": { "characters": 26 } }}以上仅展示事件结构，实际字段及用量出现位置以上游响应为准。payload.usage.characters 是字符用量；同任务可能多次报告累计值，不要把每个累计值重复相加。
本文选择 MP3，可顺序拼接二进制帧。二进制帧不是 Base64，无需解码；JSON 控制帧不能写入音频文件。若改用 PCM，应按模型返回的采样率、声道和采样格式配置播放器，不能仅修改扩展名将 PCM 当作 MP3 或 WAV。
Python 完整示例：流式接收并保存 MP3
安装依赖：
python -m pip install websocket-client保存为 qwen_tts_stream.py。此示例通过 WebSocket 接收并保存音频，不包含音频设备播放逻辑。失败时保留 .part 文件用于检查，只有收到成功结束事件才生成最终 MP3。
import jsonimport osimport timeimport uuidfrom pathlib import Path import websocket def main(): key = os.environ.get("MAITOKEN_API_KEY", "").strip() if not key: raise RuntimeError("请先设置 MAITOKEN_API_KEY") task_id = str(uuid.uuid4()) target = Path(f"qwen-{task_id}.mp3") partial = target.with_suffix(".mp3.part") texts = ["你好，欢迎使用 MaiToken。", "这是千问实时语音合成示例。"] started = False finished = False audio_bytes = 0 latest_usage = None ws = websocket.create_connection( "wss://api.maitoken.com/api-ws/v1/inference", header={"Authorization": f"Bearer {key}"}, timeout=30, ) def send(action, payload): ws.send(json.dumps({ "header": { "action": action, "task_id": task_id, "streaming": "duplex", }, "payload": payload, }, ensure_ascii=False)) try: print("任务 ID：", task_id) send("run-task", { "task_group": "audio", "task": "tts", "function": "SpeechSynthesizer", "model": "qwen-audio-3.0-tts-plus", "parameters": { "voice": "longanlingxin", "format": "mp3", "sample_rate": 24000, }, "input": {}, }) deadline = time.monotonic() + 120 with partial.open("xb") as output: while not finished: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("任务超过示例设置的 120 秒总超时") ws.settimeout(min(30, remaining)) frame = ws.recv() if isinstance(frame, bytes): output.write(frame) audio_bytes += len(frame) continue if not frame: raise RuntimeError("连接提前关闭，未收到 task-finished") event = json.loads(frame) header = event.get("header") or {} if header.get("task_id") not in (None, "", task_id): raise RuntimeError("收到不属于当前任务的事件") payload = event.get("payload") or {} usage = payload.get("usage") if isinstance(usage, dict): latest_usage = usage name = header.get("event") if name == "task-started" and not started: started = True # 两个短句可立即提交；接入持续 LLM 输出时使用独立发送逻辑， # 保持本接收循环运行，并在文本结束后才发送 finish-task。 for text in texts: send("continue-task", {"input": {"text": text}}) send("finish-task", {"input": {}}) elif name == "task-failed": raise RuntimeError( f"任务失败：{header.get('error_code')}，" f"{header.get('error_message')}" ) elif name == "task-finished": finished = True if not audio_bytes: raise RuntimeError("任务结束但未收到音频数据") partial.replace(target) print(f"已保存 {target}，共 {audio_bytes} 字节") print("上游用量：", latest_usage) finally: ws.close() if __name__ == "__main__": main()Bash / Git Bash：
export MAITOKEN_API_KEY='替换为你的 MaiToken API Key'python qwen_tts_stream.pyWindows PowerShell：
$env:MAITOKEN_API_KEY = '替换为你的 MaiToken API Key'python qwen_tts_stream.py示例设置 30 秒单次接收超时和 120 秒任务总超时，用于短文本接入验证；长文本业务应按实际需求调整。超时或中断不代表一定没有产生费用，不要自动无限重试。
HTTP 非实时调用
如果不需要边输入边接收，可以使用 HTTP 路由。注意参数位置：HTTP 的 voice、format、sample_rate 放在 input，不同于上面 WebSocket 的 payload.parameters。
Bash / cURL
export MAITOKEN_API_KEY='替换为你的 MaiToken API Key' curl --fail-with-body --silent --show-error --max-time 120 \ 'https://api.maitoken.com/api/v1/services/audio/tts/SpeechSynthesizer' \ -H "Authorization: Bearer ${MAITOKEN_API_KEY}" \ -H 'Content-Type: application/json; charset=utf-8' \ --data-binary '{"model":"qwen-audio-3.0-tts-plus","input":{"text":"你好，这是千问语音合成测试。","voice":"longanlingxin","format":"mp3","sample_rate":24000}}'常见问题
| 现象 | 排查方式 |
|---|---|
| 握手 HTTP 401 | 检查 MaiToken Key 是否有效、是否带Bearer，不要使用后台登录 Token |
模型不可用或task-failed |
检查 Key 分组权限、首帧模型 ID、音色权限及错误消息；握手成功不代表模型任务已经成功 |
404 或unsupported_endpoint |
检查完整原生路径，避免重复添加/v1，区分 inference 与 realtime |
| 发送文本后没有音频 | 确认已收到task-started；所有帧使用同一 task_id；最后发送 finish-task |
| 音频被截断 | 不要在发送 finish-task 后立即关闭连接，等待 task-finished |
| MP3 无法播放 | 仅拼接二进制音频帧，不能写入 JSON 文本，也不要对二进制帧做 Base64 解码 |
| 更换模型或音色被拒绝 | 新建连接并在新的 run-task 中指定 |
| 第二次 run-task 被拒绝 | MaiToken 当前一条连接只允许一个任务，新任务重新连接 |
| 429 或并发限制 | 降低并发并退避；已收到音频后发生中断时，不要无条件重放整个任务 |
| 字符用量与文本长度不同 | 使用上游 characters，不按字节数、音频时长或本地字符串长度替代 |
| 自定义音色无法调用 | 检查是否通过 MaiToken 登记，及创建音色的目标模型与当前模型是否匹配 |
SSML、指令控制及其他高级参数的适用模型、输入限制请遵循千问文档；本文示例使用普通文本，不默认开启这些功能。千问文档提到的 AOQ 不属于本文验证的 MaiToken 接入方式。
文档依据与验证范围
- 千问实时语音合成：实时合成模式与原生 WebSocket 事件流程。
- MaiToken 当前语音合成文档：现有 Seed TTS 文档结构及接口差异对照。
- 千问 Qwen-Audio-TTS 音色列表：音色选择参考。
- MaiToken gateway 的路由与 WebSocket 校验实现：确认 inference 路由、首帧模型选择，以及一连接一任务的限制。
核对日期：2026-09-29。正式环境 HTTP 调用已返回 200、音频 URL 和 26 字符用量。本轮文档编写未再次发起付费合成；WebSocket 示例依据官方协议及网关实现编写并做静态语法检查，未在正式环境实测，不能将 HTTP 成功等同于 WebSocket 已验收。