如何在Python FastAPI服务器音频流中发送ICY格式消息？

在FastAPI中通过异步生成器实现了基于音频字节的StreamingResponse，需要在流中插入消息，供React Native端的音频播放器接收并触发Event.MetadataCommonReceived事件。已知ICY格式适用，需明确：

1. 流端点所需的HTTP头部配置
2. 触发目标事件的消息格式
3. FastAPI StreamingResponse的分块传输编码是否会引发问题及解决办法

附带当前代码示例：

@router.get(
    "/stream/{session_id}",
    response_class=StreamingResponse,
    responses={200: {"content": {"audio/mpeg": {}}, "description": "An audio file in MP3 format"}},
)
async def stream_audio(session_id: str):
...
    return StreamingResponse(
        stream_from_queue(<some asyncio.Queue>, session_id),
        headers={
            "content-type": "audio/mpeg",
            "icy-metaint": "16000",
            "icy-name": "MyAudioStream",
            "icy-genre": "Podcast",
            "icy-url": "http://localhost:8000"
        },
        media_type="audio/mpeg",
    )


async def stream_from_queue(queue: Queue, session_id: str):
     ... # 从队列获取音频块
     ... # 插入元数据逻辑待实现

一、ICY流必备HTTP头部 #

ICY协议依赖以下HTTP头部，部分为必填项：

• icy-metaint：必填，指定每多少字节音频后插入一次元数据块（比如你设置的16000）。播放器会根据这个值定位元数据位置。
• content-type：建议设为audio/mpeg或audio/mp3，匹配音频格式。
• icy-name：可选，流的名称，会被播放器识别展示。
• icy-genre：可选，流的分类标签。
• icy-url：可选，流的关联网页地址。
• icy-br：可选，流的比特率（单位：kbps），帮助播放器优化播放。
• Transfer-Encoding: identity：关键，禁用FastAPI默认的分块传输编码，因为ICY协议基于HTTP/1.0，不依赖分块传输。

修改后的headers配置示例：

headers={
    "content-type": "audio/mpeg",
    "icy-metaint": "16000",
    "icy-name": "MyAudioStream",
    "icy-genre": "Podcast",
    "icy-url": "http://localhost:8000",
    "icy-br": "128",
    "Transfer-Encoding": "identity"
}

二、触发Event.MetadataCommonReceived的元数据格式 #

要触发React Native播放器的该事件，元数据必须严格遵循ICY规范：

1. 元数据块以1个字节开头，代表后续元数据的长度（单位：16字节块）。比如要传输16字节的元数据，这个字节值为1；如果是32字节，值为2，以此类推。如果没有元数据要发送，这个字节设为0。
2. 后续的元数据内容采用key='value';的格式，比如StreamTitle='今日更新';StreamUrl='http://xxx';，需要用单引号包裹值。
3. 元数据总长度必须是16字节的倍数，不足部分用**空字节（\x00）**填充。

代码实现示例 #

修改stream_from_queue函数，添加元数据插入逻辑：

import asyncio
from asyncio import Queue

async def stream_from_queue(queue: Queue, session_id: str):
    # 记录已发送的音频字节数，匹配icy-metaint设置的16000
    sent_bytes = 0
    icy_metaint = 16000

    while True:
        audio_chunk = await queue.get()
        if audio_chunk is None:
            break
        
        # 发送音频块
        yield audio_chunk
        sent_bytes += len(audio_chunk)

        # 检查是否到达元数据插入点
        if sent_bytes >= icy_metaint:
            # 构造元数据内容
            metadata_str = "StreamTitle='新节目上线';StreamUrl='http://localhost:8000';"
            # 计算需要填充的空字节数，确保总长度是16的倍数
            metadata_len = (len(metadata_str) + 15) // 16
            padded_metadata = metadata_str.ljust(metadata_len * 16, '\x00')
            # 先发送长度字节，再发送填充后的元数据
            yield bytes([metadata_len]) + padded_metadata.encode('utf-8')
            # 重置计数
            sent_bytes = 0

三、分块传输编码的问题解决 #

FastAPI的StreamingResponse默认会启用Transfer-Encoding: chunked分块传输，但ICY协议不兼容这种模式——分块传输会给每个数据块添加长度前缀，破坏ICY的音频+元数据的固定间隔结构，导致播放器无法识别元数据。

解决办法就是在响应头部显式设置Transfer-Encoding: identity，强制禁用分块传输。同时注意不要设置Content-Length头部，因为音频流的总长度是动态的。

内容的提问来源于stack exchange，提问作者Ivan Vinitskyi