连接AssemblyAI实时转录WebSocket API出现4001未授权错误的原因及解决方法

问题 #

连接AssemblyAI实时转录WebSocket端点时出现4001: Not authorized错误，错误日志如下：

WebSocket closed with code 4001: Not authorized
Connecting to WebSocket at wss://api.assemblyai.com/v2/realtime/ws?sample_rate=16000
WebSocket connected. Receiving SessionBegins...
SessionBegins: {"error":"Not authorized"}
Starting to send and receive messages...
Receive: WebSocket closed with code 4001: Not authorized
Send: WebSocket closed with code 4001: Not authorized

使用的Python代码：

import websockets
import asyncio
import base64
import json
from configure import auth_key
import pyaudio

# Audio configuration
FRAMES_PER_BUFFER = 3200
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000

# Initialize PyAudio
p = pyaudio.PyAudio()

# Start recording
stream = p.open(
    format=FORMAT,
    channels=CHANNELS,
    rate=RATE,
    input=True,
    frames_per_buffer=FRAMES_PER_BUFFER
)

# AssemblyAI WebSocket endpoint
URL = "wss://api.assemblyai.com/v2/realtime/ws?sample_rate=16000"
async def send_receive():
    print(f"Connecting to WebSocket at {URL}")

    try:
        auth_url = f"{URL}&auth_key={auth_key}"  # Append API key to URL
        async with websockets.connect(
            auth_url,
            ping_interval=5,
            ping_timeout=20
        ) as _ws:
            await asyncio.sleep(0.1)
            print("WebSocket connected. Receiving SessionBegins...")

            # Receive SessionBegins
            session_begins = await _ws.recv()
            print(f"SessionBegins: {session_begins}")
            print("Starting to send and receive messages...")

            # Define the send coroutine
            async def send():
                while True:
                    try:
                        # Read audio data, encode and send
                        data = stream.read(FRAMES_PER_BUFFER)
                        data = base64.b64encode(data).decode("utf-8")
                        json_data = json.dumps({"audio_data": data})
                        await _ws.send(json_data)
                        print("Sent audio data")
                    except websockets.ConnectionClosedError as e:
                        print(f"Send: WebSocket closed with code {e.code}: {e.reason}")
                        break
                    except Exception as e:
                        print(f"Send: Unexpected error: {e}")
                        break
                    await asyncio.sleep(0.01)

            # Define the receive coroutine
            async def receive():
                while True:
                    try:
                        # Receive and print text data
                        result_str = await _ws.recv()
                        print(f"Received: {json.loads(result_str).get('text', '')}")
                    except websockets.ConnectionClosedError as e:
                        print(f"Receive: WebSocket closed with code {e.code}: {e.reason}")
                        break
                    except Exception as e:
                        print(f"Receive: Unexpected error: {e}")
                        break

            # Run send and receive concurrently
            await asyncio.gather(send(), receive())

    except Exception as e:
        print(f"Connection failed: {e}")

while True:
    try:
        asyncio.run(send_receive())
    except KeyboardInterrupt:
        print("Program interrupted by user. Exiting...")
        break
    except Exception as e:
        print(f"Unexpected error: {e}")
        break

已采取的排查步骤：

• 验证API密钥（auth_key）正确且已正确包含在请求中
• 检查URL格式正确，已将auth_key作为查询参数追加
• 确认API密钥拥有访问实时转录服务的必要权限

需要解决的问题：为何出现该未授权错误，以及如何修复？

解决方案 #

1. 修正API密钥传递方式 #

AssemblyAI实时WebSocket API要求将API密钥放在请求头中，而非URL查询参数——这是导致未授权错误的核心原因。修改连接逻辑，通过extra_headers参数传递密钥：

async with websockets.connect(
    URL,
    ping_interval=5,
    ping_timeout=20,
    extra_headers={"Authorization": auth_key}  # 正确的密钥传递方式
) as _ws:

2. 排查密钥本身问题 #

• 检查auth_key是否存在多余空格、换行符，建议直接从AssemblyAI控制台复制粘贴
• 确认密钥对应账号状态正常，无欠费、权限限制等情况

3. 验证音频参数匹配 #

确保音频配置（采样率16000、单声道、16位整数格式）与API要求完全一致，避免后续出现其他异常。

修改后的核心代码片段 #

# AssemblyAI WebSocket endpoint
URL = "wss://api.assemblyai.com/v2/realtime/ws?sample_rate=16000"
async def send_receive():
    print(f"Connecting to WebSocket at {URL}")

    try:
        async with websockets.connect(
            URL,
            ping_interval=5,
            ping_timeout=20,
            extra_headers={"Authorization": auth_key}
        ) as _ws:
            await asyncio.sleep(0.1)
            print("WebSocket connected. Receiving SessionBegins...")

            # 后续代码保持不变...

内容的提问来源于stack exchange，提问作者Rouaa Belhaj ali