媒体处理 WebSocket 流式语音合成（TTS）协议

测试工具: 单击打开测试工具 Demo

一、1.1 URL 格式

wss://mps.cloud.tencent.com/tts/v1/{appid}?voiceId={音色ID}&language=zh&secretId={SecretId}&nonce={随机串}&timeStamp={签名生效 unix 秒}&expired={签名失效 unix 秒}&signature={签名}

注意：说明：

注意：

二、1.2 URL 参数

注意：说明：

注意：任一必填参数缺失或非法，服务下发 Handshake 错误包后立即关闭连接。

三、1.3 签名算法

签名沿用腾讯云 CAM 签名标准。客户端按下列规则生成 signature：

1. CanonicalQueryString：取 URL 上除 signature 之外的所有 query 参数，按 key 字典序升序，对每个 value 做 URL 编码，以 & 拼接。

2. CanonicalRequest：

·HTTPRequestMethod：post

·CanonicalURI：/tts/v1/{appid}

·CanonicalQueryString：第 1 步结果

·CanonicalHeaders：content-type:application/json; charset=utf-8\\nhost:mps.cloud.tencent.com\\n

·SignedHeaders：content-type;host

·HashedRequestPayload：sha256("")

3. 流程参考：签名生成。

4. 把参与签名的 query 参数与 signature 一并附在 URL Query 上。

注意：警告：

注意：SecretKey 严禁出现在客户端 / 浏览器侧的明文环境。生产环境签名必须由接入方后端代签，再下发给前端用于连接。

四、1.4 握手包（Handshake）

服务在完成参数校验、鉴权、配额检查、上游会话建立等动作后，下发 Handshake 结果：

{ "NotificationType": "Handshake", "TaskId": "125xxxxxx9-tts-xxxxxxxxxxxxxxxx", "HandshakeResult": { "Code": 0, "Message": "success", "Format": "mp3", "SampleRate": 24000 }}

·成功：Code = 0，Format / SampleRate 为协商后的最终值，可能与请求值不同。

·失败：Code 取自 错误码表，Message 为简要文案；服务随后主动关闭连接。

·客户端必须等本消息到达且 Code = 0 后再开始发送上行业务消息。

·任意握手失败：客户端应停止本次会话，不要在同一连接重试。

五、2.1 Format / SampleRate 协商

服务对于音色 ID (不同的音色 ID 引擎可能不一样) 内部维护一张「支持的 (format, sampleRate) 组合」表。

·若 URL 请求的 (format, sampleRate) 命中该音色支持表，则原样使用。

·若不命中，则整体回退到该音色的默认 (format, sampleRate)。

·协商结果通过握手包 HandshakeResult.Format / HandshakeResult.SampleRate 字段告知。

客户端必须以握手包返回值为准初始化播放器 / 解码器，不要假设其等于自己请求的值。

推荐组合（覆盖度最高）：

六、2.2 上行：合成请求

握手成功后，客户端通过 WebSocket 文本发送 JSON：

{ "Text": "你好，世界", "Final": false }

上行仅接受文本消息 + JSON，二进制消息会被服务忽略。

七、2.3 下行：通知包

所有下行消息分为文本消息（JSON）和二进制消息（音频数据）。

文本 JSON 消息：

{ "NotificationType": "Handshake | ProcessEof", "TaskId": "…", "HandshakeResult": { … }, "ProcessEofInfo": { … }}

两种 NotificationType 互斥，每条消息仅携带一种对应的子字段。

八、2.3.1 Handshake —— 握手结果

详情请参见 1.4 内容。

九、2.3.2 Audio —— 音频消息

音频按协商后的 Format / SampleRate 编码：

·mp3 / wav / opus：已含容器封装。

·pcm：裸 PCM，s16le 单声道。

·ulaw / alaw：8 kHz 电话编码，需相应解码。

十、2.3.3 ProcessEof —— 会话结束

{ "NotificationType": "ProcessEof", "TaskId": "…", "ProcessEofInfo": { "Code": 0, "Message": "finished normally" }}

·触发条件：

·合成自然完成：Code = 0

·空闲超时： Code = 4002

·上行消息不合法：Code = 4003 / 4001

·服务异常：Code = 5000

·ProcessEof 严格排在所有 Audio 消息之后下发，客户端可据此安全结束播放。

·客户端收到本消息后应主动 close WebSocket，不要再发送任何消息；服务也会随后关闭连接。

十一、3. 超时与异常

客户端建议：

·连接成功后，间隔小于 timeoutSec / 2 发送有效文本或保证有持续下行音频，避免空闲超时。

·收到任意 ProcessEof 即视为终态，不要在同一连接上重试；如需重试请新建 WebSocket。

·等到 ProcessEof 再 close，否则可能漏掉消息。

·网络异常重连建议采用指数退避，每次重连重新生成 nonce 与签名。

十二、4. 限流

·服务按账号（uin）维度做并发连接数限制，默认并发数为 2。超出限额的新连接握手返回 4004。

·timeoutSec 最大值 120 秒。

注意：说明：

注意：如需调整默认并发数等配额，请 联系我们。

十三、5. 错误码

4xxx 为客户端可纠正错误，5xxx 为服务端错误。

十四、6. 一次完整会话示例

URL：

wss://mps.cloud.tencent.com/tts/v1/1258344699?expired=1782293076&format=pcm&language=zh&nonce=6751517593&sampleRate=16000&secretId=AAABBBCCCDDD&timeStamp=1782289476&timeoutSec=60&voiceId=s1_sL%2FWCc2KmDIEKzDpmRZ2tDeHlWXN0k79tv%2BNRBWzM%2B93Oo0N0m1rRPqycpiXzSo%3D&signature=xxxxxxxxx

时序：

十五、7. 接入 Checklist

URL 参数按 1.2内容 拼齐，签名由接入方后端代签，SecretKey 不出现在客户端。

连接建立后先等待 Handshake 结果再发送业务消息，并以 HandshakeResult.Format / SampleRate 初始化播放器。

上行字段严格使用 Text / Final / Flush，UTF-8 编码，单条 ≤ 5000 字符。

收到 ProcessEof 即终态，主动 close，不在同连接重试；需要重试时新建 WebSocket 并重新签名。

鉴权 / 配额错误（4xxx）不要无限重试，避免触发风控。

十六、8. 支持语种

详情请参见 文档说明。

十七、9.示例代码

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""WebSocket TTS demo client.Sign the request with TC3-HMAC-SHA256, open a WSS connection, send text,collect audio frames (binary ws messages) and save the merged audio to alocal file. Control signals (Handshake / ProcessEof) still come as JSON textframes; audio payload is delivered as raw binary frames without JSON wrapping.Dependency: pip install websocketsUsage: export TTS_SECRET_ID=AKIDxxxx export TTS_SECRET_KEY=xxxxxxxx python3 wss-tts.py \\ –host mps.cloud.tencent.com \\ –appid your_appid \\ –voice s1_iX4D1zb9hyem/Bp2GpyJ7cD0miMrFVWDbtKap71N8cF8M7raq9RAupL+bRWn \\ –format mp3 \\ –sample-rate 22050 \\ –language zh \\ –text "你好，世界。这是一条测试合成的文本。" \\ –output out.mp3"""import argparseimport asyncioimport hashlibimport hmacimport jsonimport loggingimport osimport randomimport sslimport structimport sysimport timefrom datetime import datetime, timezonefrom typing import Dict, List, Tuplefrom urllib.parse import quoteimport websocketsLOG = logging.getLogger("wss-tts")# CAM 签名链路所识别的服务名（需与服务端配置保持一致）SERVICE_NAME = "mps"# 规范请求中使用的 HTTP 请求方法（服务端要求小写 post）HTTP_METHOD = "post"# 构造规范请求头时默认使用的内容类型CONTENT_TYPE = "application/json; charset=utf-8"def build_signed_url( host: str, appid: str, secret_id: str, secret_key: str, biz_params: Dict[str, str], valid_seconds: int = 3600, use_ssl: bool = True,) -> str: """Return a signed wss/ws URL ready to be passed to websockets.connect. biz_params holds voiceId / format / sampleRate / language / timeoutSec / speed / vol / resId. The function appends secretId / nonce / timeStamp / expired / signature with TC3-HMAC-SHA256 signing. """ timestamp = int(time.time()) expired = timestamp + valid_seconds nonce = str(random.randint(10 ** 9, 10 ** 10 – 1)) params = {k: str(v) for k, v in biz_params.items() if v not in (None, "")} params["secretId"] = secret_id params["nonce"] = nonce params["timeStamp"] = str(timestamp) params["expired"] = str(expired) # 规范查询字符串：键名升序排列，值进行 URL 编码处理。 keys_sorted = sorted(params.keys()) canonical_qs = "&".join(f"{k}={quote(params[k], safe='')}" for k in keys_sorted) path = f"/tts/v1/{appid}" # 规范请求格式必须与服务端（CAM api_sha256_v4）及可用网页客户端完全一致: # post\\n # /tts/v1/{appid}\\n # {canonical_qs}\\n # content-type:application/json; charset=utf-8\\n # host:{host}\\n # \\n <- blank line that separates headers and signed_headers # content-type;host\\n <- signed_headers, followed by an empty trailing line canonical_request = "\\n".join([ HTTP_METHOD, path, canonical_qs, f"content-type:{CONTENT_TYPE}", f"host:{host}", "", "content-type;host", "", ]) date = datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d") credential_scope = f"{date}/{SERVICE_NAME}/tc3_request" string_to_sign = "\\n".join([ "TC3-HMAC-SHA256", str(timestamp), credential_scope, hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(), ]) k_date = hmac.new(("TC3" + secret_key).encode("utf-8"), date.encode("utf-8"), hashlib.sha256).digest() k_service = hmac.new(k_date, SERVICE_NAME.encode("utf-8"), hashlib.sha256).digest() k_signing = hmac.new(k_service, b"tc3_request", hashlib.sha256).digest() signature = hmac.new(k_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest() # 拼接签名，生成最终查询字符串 final_qs = "&".join(f"{k}={quote(params[k], safe='')}" for k in keys_sorted) final_qs += f"&signature={signature}" scheme = "wss" if use_ssl else "ws" return f"{scheme}://{host}{path}?{final_qs}"def wrap_pcm_to_wav(pcm: bytes, sample_rate: int, sample_width: int = 2, channels: int = 1) -> bytes: """Wrap raw PCM s16le bytes with a minimal RIFF/WAVE header for playback.""" byte_rate = sample_rate * channels * sample_width block_align = channels * sample_width data_size = len(pcm) riff_size = 36 + data_size header = b"RIFF" + struct.pack("<I", riff_size) + b"WAVE" fmt_chunk = b"fmt " + struct.pack("<IHHIIHH", 16, 1, channels, sample_rate, byte_rate, block_align, sample_width * 8) data_chunk = b"data" + struct.pack("<I", data_size) return header + fmt_chunk + data_chunk + pcmdef redact_for_log(payload: dict) -> str: """Render a control-signal JSON payload for logging. Audio bytes are delivered as binary ws frames now, so this only needs to pretty-print Handshake / ProcessEof JSON dicts. """ if not isinstance(payload, dict): return json.dumps(payload, ensure_ascii=False) return json.dumps(payload, ensure_ascii=False)async def run_session(url: str, texts: List[str], output: str) -> int: """Run one TTS session and persist the merged audio.""" audio_chunks: List[bytes] = [] handshake_format = "" handshake_sample_rate = 0 final_code = -1 final_message = "" ssl_ctx = ssl.create_default_context() if url.startswith("wss://") else None LOG.info("connect %s", url) ws = await websockets.connect(url, ssl=ssl_ctx, max_size=None) try: # 步骤 1：需等待握手应答（Handshake ack）返回后，再发送业务文本数据 first = await ws.recv() first_obj = json.loads(first) if isinstance(first, str) else {} LOG.info("recv handshake: %s", redact_for_log(first_obj)) if first_obj.get("NotificationType") != "Handshake": LOG.error("first frame is not Handshake: %s", first_obj) return 1 hs = first_obj.get("HandshakeResult") or {} if hs.get("Code") != 0: LOG.error("handshake failed code=%s message=%s", hs.get("Code"), hs.get("Message")) return 1 handshake_format = (hs.get("Format") or "").lower() handshake_sample_rate = int(hs.get("SampleRate") or 0) LOG.info("handshake ok format=%s sampleRate=%d", handshake_format, handshake_sample_rate) # 步骤 2：发送全部文本分片，最后发送空数据包标识数据流结束（EOS） for idx, seg in enumerate(texts): pkt = {"Text": seg, "Final": False} await ws.send(json.dumps(pkt, ensure_ascii=False)) LOG.info("send #%d: %s", idx + 1, pkt) eos = {"Text": "", "Final": True} await ws.send(json.dumps(eos)) LOG.info("send EOS: %s", eos) # 步骤 3：持续接收数据帧，直至收到结束帧（ProcessEof） # – 二进制 WebSocket 帧 -> 原始音频字节流，直接追加拼接 # – 文本 WebSocket 帧 -> 控制类 JSON 报文（握手消息 / 结束指令 ProcessEof） while True: try: msg = await ws.recv() except websockets.ConnectionClosed as exc: LOG.info("ws closed: code=%s reason=%s", exc.code, exc.reason) break if isinstance(msg, (bytes, bytearray)): chunk = bytes(msg) if not chunk: continue audio_chunks.append(chunk) LOG.info("audio chunk %d bytes (total=%d)", len(chunk), sum(len(c) for c in audio_chunks)) continue try: obj = json.loads(msg) except json.JSONDecodeError as exc: LOG.warning("ignore non-json text frame: %s err=%s", msg, exc) continue LOG.info("recv: %s", redact_for_log(obj)) ntype = obj.get("NotificationType") if ntype == "ProcessEof": info = obj.get("ProcessEofInfo") or {} final_code = int(info.get("Code", -1)) final_message = info.get("Message", "") LOG.info("process eof code=%d message=%s, exit", final_code, final_message) break else: LOG.warning("unknown NotificationType: %s", ntype) finally: # 强制关闭连接，不再等待服务端关闭帧； # 收到 ProcessEof 后进程立即退出， # 避免在优雅关闭握手上阻塞，防止卡死 asyncio 事件循环 try: await asyncio.wait_for(ws.close(), timeout=1.0) except Exception: try: ws.transport.abort() except Exception: pass if not audio_chunks: LOG.error("no audio received, code=%d message=%s", final_code, final_message) return 2 merged = b"".join(audio_chunks) out_path = output # 原始 PCM 响应数据需封装 WAV 文件头，以便直接播放 if handshake_format == "pcm": merged = wrap_pcm_to_wav(merged, handshake_sample_rate or 16000) if not out_path.lower().endswith(".wav"): out_path = out_path + ".wav" with open(out_path, "wb") as fp: fp.write(merged) LOG.info("saved %d bytes to %s (format=%s sampleRate=%d code=%d)", len(merged), out_path, handshake_format, handshake_sample_rate, final_code) return 0 if final_code == 0 else 3def parse_args() -> Tuple[argparse.Namespace, Dict[str, str]]: """Parse CLI arguments and assemble the business URL query params.""" parser = argparse.ArgumentParser(description="WebSocket TTS demo client") parser.add_argument("–host", default="mps.cloud.tencent.com", help="server host") parser.add_argument("–appid", required=True, help="tencent cloud appid") parser.add_argument("–secret-id", default=os.environ.get("TTS_SECRET_ID", ""), help="cam secret id (env TTS_SECRET_ID)") parser.add_argument("–secret-key", default=os.environ.get("TTS_SECRET_KEY", ""), help="cam secret key (env TTS_SECRET_KEY)") parser.add_argument("–voice", required=True, help="voiceId, e.g. ws_voice_101") parser.add_argument("–format", default="mp3", choices=["pcm", "mp3", "wav", "flac", "opus", "ulaw", "alaw"], help="audio format (server may downgrade, see Handshake ack)") parser.add_argument("–sample-rate", type=int, default=24000, help="audio sample rate (Hz)") parser.add_argument("–language", default="zh", help="language hint, e.g. zh / en / auto") parser.add_argument("–timeout-sec", type=int, default=60, help="session idle timeout (<=120)") parser.add_argument("–speed", type=float, default=0.0, help="speech speed, 0 = use upstream default") parser.add_argument("–vol", type=float, default=0.0, help="speech volume, 0 = use upstream default") parser.add_argument("–res-id", default="", help="optional resource tag for billing") parser.add_argument("–text", action="append", default=None, help="text segment, can be repeated to send multiple segments") parser.add_argument("–output", default="out.bin", help="output file path") parser.add_argument("–no-ssl", action="store_true", help="use ws:// instead of wss://") parser.add_argument("–verbose", action="store_true", help="enable debug logging") args = parser.parse_args() if not args.secret_id or not args.secret_key: parser.error("secret-id/secret-key required (or via env TTS_SECRET_ID/TTS_SECRET_KEY)") if not args.text: args.text = ["真正的危险不是计算机开始像人一样思考，而是人开始像计算机一样思考。"] biz: Dict[str, str] = { "voiceId": args.voice, "format": args.format, "sampleRate": str(args.sample_rate), "timeoutSec": str(args.timeout_sec), } if args.language: biz["language"] = args.language if args.res_id: biz["resId"] = args.res_id if args.speed > 0: biz["speed"] = str(args.speed) if args.vol > 0: biz["vol"] = str(args.vol) return args, bizdef main() -> int: args, biz = parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) url = build_signed_url( host=args.host, appid=args.appid, secret_id=args.secret_id, secret_key=args.secret_key, biz_params=biz, use_ssl=not args.no_ssl, ) return run_async(run_session(url, args.text, args.output))def run_async(coro) -> int: """Run an async coroutine on Python >=3.6 (asyncio.run only exists in 3.7+).""" if hasattr(asyncio, "run"): return asyncio.run(coro) loop = asyncio.new_event_loop() try: asyncio.set_event_loop(loop) return loop.run_until_complete(coro) finally: # 关闭事件循环前取消所有未完成任务，否则 # 后台任务（如 WebSocket 关闭连接）可能在已关闭的循环上执行回调， # 导致程序退出时产生大量冗余报错堆栈 try: pending = [t for t in asyncio.Task.all_tasks(loop) if not t.done()] for t in pending: t.cancel() if pending: loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) except Exception: pass try: loop.run_until_complete(loop.shutdown_asyncgens()) except Exception: pass loop.close() asyncio.set_event_loop(None)if __name__ == "__main__": sys.exit(main())