简介：PyPI官网发布的aiortc-0.9.27.tar.gz，是面向Python开发者的WebRTC异步实时通信库源码包。aiortc基于asyncio框架实现了RTCPeerConnection、RTCDataChannel、RTCDTMFSender等组件，并内置ICE、STUN、TURN连接协商与NAT穿透支持，可用来构建视频会议、在线教育、实时文件共享、低延迟语音聊天和远程桌面等应用。相比浏览器端WebRTC，它把同一套实时通信能力带到Python服务端，适合需要开发媒体网关、信令后端或嵌入式实时模块的工程师。压缩包共139个文件，以71个Python源码文件为核心，包含实现RTCPeerConnection、SCTP传输等的py模块；39个二进制测试数据覆盖H.264视频帧、RTP扩展头、SDES、SCTP初始化等抓包样本；16个reStructuredText文档说明API与协议细节；另附配置文件、许可证及WAV测试音频。整体仅1.1MB，结构清晰，便于查阅。已有138人学习下载。解包后可直接阅读全部源码与示例，结合测试数据理解H.264封包、RTP消息格式、DTMF和SCTP数据通道等底层实现，为二次开发、故障排查和协议研究提供直观参考。

1. 为什么 WebRTC 场景里会需要 aiortc 0.9.27 这种 Python 库

如果你的工作涉及 WebRTC 网关、实时音视频转发节点或者自动化测试工具，大概率会碰到同一个尴尬：浏览器端 JS 生态的 WebRTC 实现已经非常成熟，但服务端要用 Python 去模拟一个 Peer、抓取 RTP 包、验证 SCTP 数据通道时，却找不到趁手的库。aiortc 0.9.27 就是用来填补这个空白的——它用 asyncio 把 RTCPeerConnection、ICE、DTLS、SCTP 等协议栈在 Python 里重写了一遍，让你能用同步思维写出异步的实时通信逻辑。无论是做流媒体录制、信令网关调试，还是给 WebRTC 客户端写压力测试，这个库都能让你绕过浏览器内核，直接在 Python 进程里操作媒体轨道和 DataChannel。适合有 Python 基础、想深入 WebRTC 内部机制或需要做服务端 RTC 能力的开发者。

2. 从 tar.gz 到可用环境：aiortc 0.9.27 的安装与依赖排查

2.1 解压源码包之前先搞清格式

aiortc 0.9.27 在 PyPI 上以 .tar.gz 形式分发，说明这是源码包而非 wheel。源码包体积通常比 wheel 大，而且安装时需要本地编译 C/C++ 扩展。下载完成后，你拿到的 aiortc-0.9.27.tar.gz 里除了 Python 源码，还带了一批测试资源，文件名类似 h264_0001.bin 、 rtp_with_csrc.bin 、 sctp_init.bin ，这些是官方测试套件里用到的真实抓包数据，后面我们会用到其中几个文件。

在 Linux 或 macOS 上用 tar 解压即可：

tar -zxvf aiortc-0.9.27.tar.gz

参数说明： -z 表示通过 gzip 解压， -x 解包， -v 打印过程， -f 指定文件名。解压后进入目录：

cd aiortc-0.9.27
ls -la

目录下应有 setup.py 、 aiortc/ 源码目录、 tests/ 测试目录和若干 .bin 测试文件。常见的一个小坑是拿到手就执行 pip install aiortc-0.9.27.tar.gz ，但系统提示 tar.gz没有那个文件或目录 ，十有八九是你下载的文件名被浏览器改成了 aiortc-0.9.27 (1).tar.gz 或者放在了别的路径，先把文件全路径确认好。

2.2 pip 安装与常见依赖错误

解压后推荐直接用 pip 安装本地源码包：

pip install ./aiortc-0.9.27

这里用目录而不是 tar.gz 的好处是，如果编译报错，pip 能更清晰地告诉你卡在哪个源文件。aiortc 0.9.27 的核心依赖包括 cryptography 、 pyee 、 av 、 aioice 等。其中 av 是 PyAV，用于音视频帧的解码与封装，它依赖 FFmpeg 库。如果你的系统里没有 FFmpeg，通常会在安装阶段出现类似 Could not find libavcodec 的错误。

我自己在 Ubuntu 20.04 上的处理方式是先装系统级依赖：

sudo apt update
sudo apt install -y libavdevice-dev libavfilter-dev libavformat-dev libavcodec-dev libssl-dev

然后重新执行 pip 安装。如果是 macOS，建议先 brew install ffmpeg ，再把 FFmpeg 的 bin 目录加入 PATH。这里的关键点是：aiortc 的源码安装过程会自动检测 av 的编译环境，而 av 的构建脚本会去找 FFmpeg 头文件。缺了这些，后面即使库装上了，导入时也会报 RuntimeError: libavcodec not found 。

2.3 验证安装并检查自带测试资源

安装完成后，用一段 Python 验证关键模块能否正常导入：

import aiortc
from aiortc import RTCPeerConnection, RTCDataChannel
from aiortc.contrib.media import MediaPlayer, MediaRecorder
print(aiortc.__version__)

如果打印出 0.9.27 ，说明主库可用了。然后检查测试资源文件是否完整，特别是 rtp_with_csrc.bin 、 sctp_init.bin 这几个，后续调试 RTP 打包和 SCTP 初始化时会用到：

file tests/rtp_with_csrc.bin tests/sctp_init.bin

输出一般会显示这是 data 格式，因为它们是直接从网络抓包导出的二进制 RTP/SCTP 报文。

3. aiortc 核心对象拆解：RTCPeerConnection、媒体流与 DataChannel

3.1 RTCPeerConnection 的异步生命周期

在 WebRTC 里， RTCPeerConnection 相当于所有媒体和数据的载体。aiortc 用它模拟浏览器侧的行为，但实例化方式和浏览器 API 略有差异——默认构造函数不会自动开始 ICE 收集，需要显式调用 pc.createOffer() 或 pc.setLocalDescription() 触发。以下是一个最简创建流程：

import asyncio
from aiortc import RTCPeerConnection

async def create_peer():
    pc = RTCPeerConnection()
    # 打印连接状态变化，方便调试
    @pc.on("connectionstatechange")
    def on_connection_state_change():
        print(f"ICE state: {pc.iceConnectionState}, connection state: {pc.connectionState}")
    return pc

pc = asyncio.get_event_loop().run_until_complete(create_peer())

RTCPeerConnection() 内部会创建 aioice.Connection 实例来处理 STUN/TURN 会话。如果你不传任何 ICE 服务器，默认走 host 候选，即局域网直连。生产环境需要传 STUN 地址，可以这样初始化：

pc = RTCPeerConnection(
    iceServers=[{"urls": "stun:stun.l.google.com:19302"}]
)

注意 0.9.27 版本对 ICE 服务器配置的 key 是 urls 而不是 url ，用错会导致 aioice 直接抛异常。

3.2 用 MediaStreamTrack 推送 H264 帧

aiortc 把媒体轨抽象成 MediaStreamTrack 。官方推荐的本地媒体源是 MediaPlayer ，它支持从文件或摄像头读取。如果要从原始 H264 裸流推送，可以自己实现一个 MediaStreamTrack 子类。结合源码包里的 h264_0001.bin ，可以这么写：

from aiortc.mediastreams import MediaStreamTrack, AudioStreamTrack, VideoStreamTrack
from aiortc.rtcrtpparameters import RTCRtpCodecParameters
import av

class H264FileTrack(MediaStreamTrack):
    kind = "video"
    def __init__(self, filepath):
        super().__init__()
        self.container = av.open(filepath)
        self.stream = self.container.streams.video[0]
        self.codec = RTCRtpCodecParameters(
            mimeType="video/H264",
            clockRate=90000,
            payloadType=96
        )
    
    async def recv(self):
        for frame in self.container.decode(self.stream):
            # 返回 VideoFrame
            return frame

这里把 h264_0001.bin 当作容器文件打开，aiortc 会按 RTP 时间戳规则把帧交给发送器。 clockRate 对 H264 固定为 90000，这样接收端才能正确计算播放时间。实际使用中，如果你只是重放 PPS/SPS + IDR 裸流，建议先通过 av.open 确认文件封装格式，否则可能解码失败。

3.3 RTCDataChannel 与 SCTP 缓冲区的关联

RTCDataChannel 底层走 SCTP over DTLS，aiortc 里由 aiortc.sctp 模块实现。当我们创建 DataChannel 时，实际上是在 SCTP 关联上建立一个 stream，并配置消息分发策略。一个容易忽略的参数是 maxRetransmits 和 ordered ，它们直接影响消息可靠性和顺序。示例：

from aiortc import RTCPeerConnection, RTCDataChannel

pc = RTCPeerConnection()

@pc.on("datachannel")
def on_datachannel(channel: RTCDataChannel):
    print(f"DataChannel created: {channel.label}")
    @channel.on("message")
    def on_message(message):
        print(f"Received: {message}")

dc = pc.createDataChannel("test", ordered=True, maxRetransmits=30)

maxRetransmits=30 表示在 SCTP 层最多重传 30 次，超过后消息被丢弃。如果你需要低延迟且能容忍少量丢包，可以设置 ordered=False ，此时 SCTP 会关闭对顺序的强制等待，但注意接收端的消息顺序就不能保证。

4. 动手搭建一个最小可跑的 WebRTC 信令示例

4.1 信令设计

WebRTC 本身的协议栈不包含信令，aiortc 也不例外。最常用的信令方案是用 WebSocket 转发 SDP 和 ICE candidate。以下示例里，两个 Python 进程分别扮演 Offerer 和 Answerer，用 asyncio 的 Queue 模拟信令通道，便于理解流程。

4.2 offer/answer 交换流程

Offerer 代码核心部分：

import asyncio
import json
from aiortc import RTCPeerConnection, RTCSessionDescription

async def run_offerer(pc, signaling):
    # 1. 创建本地 SDP
    offer = await pc.createOffer()
    await pc.setLocalDescription(offer)
    
    # 2. 把 offer 通过信令发给远端
    await signaling.put({
        "sdp": pc.localDescription.sdp,
        "type": pc.localDescription.type
    })
    
    # 3. 等待 answer
    answer_data = await signaling.get()
    answer = RTCSessionDescription(sdp=answer_data["sdp"], type=answer_data["type"])
    await pc.setRemoteDescription(answer)

Answerer 侧则相反，先收到 offer，再创建 answer：

async def run_answerer(pc, signaling):
    offer_data = await signaling.get()
    offer = RTCSessionDescription(sdp=offer_data["sdp"], type=offer_data["type"])
    await pc.setRemoteDescription(offer)
    
    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)
    
    await signaling.put({
        "sdp": pc.localDescription.sdp,
        "type": pc.localDescription.type
    })

关键点在于 createOffer() 必须在 addTrack 或 createDataChannel 之后调用，否则生成的 SDP 里没有媒体描述，对端会报 InvalidStateError 。

4.3 数据通道双向通信实测

加入 DataChannel 后，Offerer 和 Answerer 都能发送消息。一个简单但完整的测试如下：

import asyncio
from aiortc import RTCPeerConnection, RTCDataChannel

async def test_datachannel():
    pc1 = RTCPeerConnection()
    pc2 = RTCPeerConnection()
    
    # 连接 pc1 和 pc2 的 ICE 事件
    @pc1.on("icecandidate")
    async def on_ice_candidate(candidate):
        if candidate:
            await pc2.addIceCandidate(candidate)
    @pc2.on("icecandidate")
    async def on_ice_candidate2(candidate):
        if candidate:
            await pc1.addIceCandidate(candidate)
    
    received = asyncio.Future()
    @pc2.on("datachannel")
    def on_datachannel(channel):
        @channel.on("message")
        def on_message(message):
            received.set_result(message)
    
    dc = pc1.createDataChannel("chat")
    await dc.open()
    
    # 交换 offer/answer
    offer = await pc1.createOffer()
    await pc1.setLocalDescription(offer)
    await pc2.setRemoteDescription(offer)
    answer = await pc2.createAnswer()
    await pc2.setLocalDescription(answer)
    await pc1.setRemoteDescription(answer)
    
    dc.send("hello from pc1")
    print(await received)  # 输出 hello from pc1

这里直接调用 addIceCandidate 而不是通过信令转发，是为了让示例在单进程内跑通。实际部署时，候选要随信令通道传到对端，而且 icecandidate 事件可能在 setRemoteDescription 之前触发，所以最好先缓存候选，等远端描述设置后再添加。

5. 用 rtp_with_csrc.bin 这类测试文件验证 RTP 打包逻辑

5.1 测试文件到底在测什么

源码包里的 rtp_with_csrc.bin 、 rtp_only_padding.bin 、 rtp_with_sdes_mid.bin 是 aiortc 官方测试套件里的协议分析样本。它们分别对应 RTP 包的三种特殊场景：携带 CSRC 标识、携带 padding 字段、携带 SDES MID 扩展头。这些文件被 tests/test_rtp.py 读取后，通过 RtpPacket.parse() 反序列化，用来验证 RTP 头解析是否正确。

如果你打开 tests/test_rtp.py ，会发现类似这样的代码：

from aiortc.rtp import RtpPacket

def test_rtp_with_csrc():
    data = open("rtp_with_csrc.bin", "rb").read()
    packet = RtpPacket.parse(data)
    assert packet.csrc == [0x12345678]

也就是说，如果你要修改 aiortc 的 RTP 处理逻辑，或者想理解带 CSRC 和 padding 的包长怎么计算，直接拿这些文件来做单元测试是最快的验证方式。

5.2 复现 aiortc 的 RTP 发送路径

更实际的用途是，用这些抓包文件来校验自己写的 RTP 打包代码是否符合规范。比如你要把一个 H264 帧封装成多个 RTP 分片，可以这样检查长度和头标志：

from aiortc.rtp import RtpPacket, RtpHeader
import hashlib

payload_size = 1200
frame_data = open("h264_0001.bin", "rb").read()

seq = 0
timestamp = 0
ssrc = 12345

while frame_data:
    chunk = frame_data[:payload_size]
    packet = RtpPacket(
        header=RtpHeader(
            marker=0,
            payload_type=96,
            sequence_number=seq,
            timestamp=timestamp,
            ssrc=ssrc,
        ),
        payload=chunk,
    )
    encoded = packet.serialize()
    print(f"seq={seq}, payload_len={len(chunk)}, total_len={len(encoded)}")
    seq += 1
    frame_data = frame_data[payload_size:]

这段代码里， marker 标志是否为 1 表示是否是最后一片，aiortc 内部会根据帧的边界自动设置。你可以对比 rtp_with_csrc.bin 的解包结果来验证自己的 serialize 是否正确。另外， rtp_with_sdes_mid.bin 这类文件常被用来测试 RTP 扩展头的序列化，aiortc 0.9.27 里对扩展头的处理集中在 rtp.py 的 RtpHeaderExtension 类中。

实际调通之后，建议你再做一件事：用 MediaRecorder 把收到的流保存成 WebM 文件，然后去查 ffprobe 的“V：h264”字段有没有正确出现。这样才算把“RTP 打包正确”和“视频能解出来”闭环起来。

提示：如果你在修改 aiortc 源码后跑测试报错，先确认你是不是直接从源码目录 import，而不是从 site-packages。用 python setup.py develop 以开发者模式安装是避免这个坑最省事的方式。

本文还有配套的精品资源，点击获取