aiomoqt - Media over QUIC Transport (MoQT)

aiomoqt is an implementation of MoQT for asyncio, layered on aiopquic.

It follows the standard asyncio transport/protocol pattern via subclasses of aiopquic.asyncio.QuicConnectionProtocol for both raw QUIC and H3/WT sessions. Message encode/decode in aiomoqt.messages is sans-I/O. The session exposes APIs for all MoQT control-plane operations and supports a range of publish/subscribe workflows for both client and server roles. The package includes example clients, benchmarking and track simulation tools, a relay probe, and a moq-interop-runner-compatible test client.

Support matrix

Draft-16 adds delta-encoded param keys, track extensions, and unified request/response. Draft-18 adds vi64 varints, a uni-stream control pair, per-request bidi streams, and Request-ID-less replies. Wire-format conformance for all three is covered by aiomoqt/tests/test_wire_conformance.py.

Installation

uv pip install aiomoqt    # or: pip install aiomoqt

Pure Python. aiopquic (the QUIC transport) installs as a binary wheel automatically. Prebuilt wheels exist for Linux (glibc 2.34+, RHEL 9 / Ubuntu 22.04+) and macOS arm64; other platforms build aiopquic from sdist and need a C toolchain — see aiopquic install notes.

HLS/DASH ingest (pub_media --input) needs PyAV, an optional extra: uv pip install 'aiomoqt[media]'. Without it, --input exits with that hint.

Quick start

1. Verify the install and reach a relay

Report the installed stack:

python -m aiomoqt.versions

Probe a relay over raw QUIC:

python -m aiomoqt.tools.relay_probe --url moqt://moqx-main.ci.openmoq.org:4433
# moqt://moqx-main.ci.openmoq.org:4433              QUIC   ✓  draft-14,draft-16,draft-18  (540ms)

Probe the same relay over H3/WebTransport:

python -m aiomoqt.tools.relay_probe --url https://moqx-main.ci.openmoq.org:4433/moq-relay
# https://moqx-main.ci.openmoq.org:4433/moq-relay   H3/WT  ✓  draft-14,draft-16,draft-18  (435ms)

The probe exits 0 if any draft handshakes, so it drops straight into a shell conditional.

Clients verify the relay's certificate against certifi's roots; pass -k to accept a self-signed one (the loopback tools do), or MOQTClient(ca_file=...) to trust a private CA.

2. Subscribe

import asyncio
from aiomoqt.client import MOQTClient

def on_object(msg, size, recv_time_us, group_id=None, subgroup_id=None):
    print(f"g={group_id} obj={msg.object_id} {size}B payload={msg.payload}")

async def main():
    client = MOQTClient('relay.example.com', 443, path='moq',
                        use_quic=True, supported_drafts=16)
    async with client.connect() as session:
        await session.client_session_init()
        session.on_object_received = on_object
        await session.subscribe('ns', 'track', wait_response=True)
        await session.async_closed()

asyncio.run(main())

3. Publish

import asyncio
from aiomoqt.client import MOQTClient
from aiomoqt.track import PublishedTrack

class HelloTrack(PublishedTrack):
    async def produce(self, out):
        for group_id in range(60):
            await out.write(group_id, 0, b"hello", group_start=True)
            await asyncio.sleep(1)

async def main():
    client = MOQTClient('relay.example.com', 443, path='moq',
                        use_quic=True, supported_drafts=16)
    async with client.connect() as session:
        await session.client_session_init()
        track = HelloTrack(session, 'ns', 'track')
        # bare PUBLISH; produce() starts when a subscriber arrives
        await track.publish()
        await track.wait_closed()

asyncio.run(main())

PublishedTrack owns stream setup, subgroup headers, object numbering, pacing, and fan-out: produce() numbers the sequence once however many subscribers there are. SubscribedTrack is the receive side. Both are in aiomoqt/track.py; the raw stream API underneath is the session's open_uni_stream() and stream_write() (aiomoqt/protocol.py) and SubgroupHeader (aiomoqt/messages/data.py).

API guide

Control messages

Every control message takes wait_response:

resp = await session.subscribe('ns', 'track', wait_response=True)   # awaits the response
req  = session.subscribe('ns', 'track')       # returns the SUBSCRIBE sent; response via handler

Register handlers for peer-initiated messages:

from aiomoqt.types import MOQTMessageType

client.register_handler(MOQTMessageType.SUBSCRIBE, my_subscribe_handler)

A SUBSCRIBE, PUBLISH_OK or subscription update addressed to a track the session publishes goes to that track; the handler sees the rest. With no SUBSCRIBE handler, a SUBSCRIBE for a track the session does not publish is refused with DOES_NOT_EXIST.

Request failures raise MOQTRequestError regardless of negotiated draft.

Tracks

Several tracks can share a session; each gets the control messages that name it or its requests. publish() announces a track (PUBLISH, PUBLISH_NAMESPACE, or both). attach() serves it without announcing, for tracks announced together under one namespace:

for name in ('video', 'audio'):
    MyTrack(session, 'ns', name).attach()
await session.publish_namespace(namespace='ns', wait_response=True)

A full track name (namespace + name) serves one track per session.

StreamMapping selects the data-plane shape: PER_GROUP (subgroup stream per group) or DATAGRAM (one object per datagram, raw QUIC or WebTransport; the object must fit one packet).

Auth

AUTH_TOKEN rides the SETUP handshake (session-level) and any request message (namespace- or track-level). Values are arbitrary bytes; the codec wraps them in the spec Token structure. Read a peer's token back from the message's parameters.

from aiomoqt.types import SetupParamType, ParamType

await session.client_session_init(parameters={SetupParamType.AUTH_TOKEN: b"session-tok"})

await session.publish_namespace('ns', parameters={ParamType.AUTH_TOKEN: b"ns-tok"},
                                wait_response=True)

ok = await session.subscribe('ns', 'track',
                             parameters={ParamType.AUTH_TOKEN: b"track-tok"},
                             wait_response=True)
print(ok.parameters.get(ParamType.AUTH_TOKEN))

Auth is control-plane only; there is no per-object authentication.

Media

aiomoqt.media implements MSF catalogs with three packagings: LOC (per-frame payloads, metadata in MOQ object properties), CMSF/CMAF (moof+mdat chunks, CMAF header in the catalog initDataList) and LOCMAF (draft-einarsson-moq-locmaf, locmafVersion 0.3: the same chunks with each moof carried as a compact header, a delta of the previous one within a group; the receiver rebuilds a canonical CMAF chunk).

The media guide covers every source, packaging, timing and delivery option, with recipes.

Command line

# Publish an mp4 as an MSF broadcast — LOC packaging, catalog + video + audio
python -m aiomoqt.tools.pub_media $RELAY -N demo/live --mp4 clip.mp4 --loop \
    --target-latency 500 --draft 16 -t 3600

# Keep 1 MiB of recent objects per media track to answer FETCH (default 0 refuses it)
python -m aiomoqt.tools.pub_media $RELAY -N demo/live --mp4 clip.mp4 --loop --fetch-history 1048576

# Same content, CMAF packaging (requires --mp4)
python -m aiomoqt.tools.pub_media $RELAY -N cmsf/live --mp4 clip.mp4 --packaging cmaf --loop

# Same chunks as LOCMAF (also with --input); audio rides ~1 s groups
python -m aiomoqt.tools.pub_media $RELAY -N cmsf/live --mp4 clip.mp4 --packaging locmaf --loop

# HLS or DASH (anything FFmpeg opens) as CMAF (needs aiomoqt[media]); --rendition all
# (experimental) = every video rendition as one alternate group, fetched in turn, so
# large ladders fall behind real time
python -m aiomoqt.tools.pub_media $RELAY -N vod --input https://cdn.example/master.m3u8 -t 3600

# Live H.264 Annex-B ingest (OBS / ffmpeg pipe); frames stamped on arrival
ffmpeg -i 'srt://0.0.0.0:9000?mode=listener' -map 0:v -c:v copy -f h264 - \
  | python -m aiomoqt.tools.pub_media $RELAY -N obs --h264 - --no-audio

# Subscribe: catalog-driven, writes playable files to ./media-out
python -m aiomoqt.tools.sub_media $RELAY -N demo/live --inspect 5 --show-catalog
ffplay media-out/video.h264      # LOC → elementary stream
ffplay media-out/video.mp4       # CMAF or LOCMAF → fMP4

# Only the LOCMAF tracks; the summary line per track counts rejected objects
python -m aiomoqt.tools.sub_media $RELAY -N cmsf/live --packaging locmaf -t 30

# Measure delivery instead: latency, jitter, loss, playout model; no decode, no files
python -m aiomoqt.tools.sub_media $RELAY -N demo/live --analyze -t 60 -i 5 --report run.csv

Source material should use short GOPs and no B-frames (-g 2×fps -sc_threshold 0 -bf 0) for low join latency. CMAF chunks carry one sample each with no composition-time offsets, so B-frame sources will not present correctly.

--inspect N prints per-frame group/object ids, size, keyframe flag, ts_skew_ms (wire latency), and extension properties. --show-catalog prints the catalog JSON as subscribers receive it.

API

from aiomoqt.media import (Catalog, CatalogTrack, InitData, MediaPublisher,
                           MediaSubscriber, LocTrackPublisher, StreamMapping)

# publisher: catalog track + one LOC track per medium
catalog = Catalog(generatedAt=..., tracks=[CatalogTrack(
    name='video', packaging='loc', isLive=True, role='video',
    codec='avc1.42C01E', width=1280, height=720, initRef='v0')],
    initDataList=[InitData.from_bytes('v0', avcc_extradata)])
pub = MediaPublisher(session, 'demo/live', catalog)
video = pub.add_track(LocTrackPublisher(session, 'demo/live', 'video',
                                        config=avcc_extradata,
                                        mapping=StreamMapping.PER_GROUP))
await pub.start()
await video.send_frame(payload, key_frame=True, timestamp=epoch_us)

# subscriber: reads the catalog, subscribes every track it describes
sub = MediaSubscriber(session, 'demo/live', on_frame=handle, on_catalog=handle_catalog)
catalog = await sub.start()

MediaSubscriber.start() joins the catalog track with SUBSCRIBE + joining FETCH (msf-01 §5) so a late joiner gets the relay-cached catalog, falling back to plain SUBSCRIBE when the peer cannot serve the fetch. on_catalog fires on the first catalog and every applied delta.

End-to-end pipeline walkthroughs — file to browser, live OBS ingest, CMAF, cross-implementation consumers — are in docs/demo-runbook.md. Load generation, ramps to a relay ceiling and host tuning for accurate runs are in docs/bench-runbook.md.

Server

# WebTransport origin
python -m aiomoqt.examples.server_example --cert cert.pem --key key.pem -p 4433

# Standalone publisher server (no relay), raw QUIC or WT
python -m aiomoqt.tools.pub_server --cert cert.pem --key key.pem -p 4433 -Q

MOQTServer serves one transport with serve(), or both on a single UDP port with serve_dual():

from aiomoqt.server import MOQTServer

server = MOQTServer('0.0.0.0', 4433, certificate='cert.pem', private_key='key.pem',
                    path='moq', supported_drafts=[18, 16])
await server.serve_dual()   # raw QUIC + H3/WT on one port

serve_dual() routes each connection by negotiated ALPN (aiopquic serve_dispatch): raw connections select a draft via the per-draft MoQT ALPNs, WebTransport connections via WT-Protocol. It replaces two-listener arrangements that split transports across ports.

Tools

Each tool runs as a module (python -m aiomoqt.tools.NAME) and most also install a console script. Every tool prints its full option set with -h (also -? or --help). Bench and media tools take a positional relay URL: moqt://host[:port] for raw QUIC, https://host[:port]/[path] for H3/WebTransport.

Examples under aiomoqt.examples are minimal, readable clients rather than instrumented tools:

Common options across clients: --namespace, --trackname, --path, --draft, --debug, --keylogfile.

python -m aiomoqt.examples.pub_example moqt://relay.ex.com
python -m aiomoqt.examples.sub_example moqt://relay.ex.com
python -m aiomoqt.tools.loopback_bench -s 4096 -P 4 -t 20
python -m aiomoqt.tools.loopback_bench --fanout -s 4096 -t 20   # through the send lane
python -m aiomoqt.tools.pub_bench moqt://relay.ex.com -s 4096 -P 4 -r 120 -t 60

Interop

Validated against live public relays — OpenMoQ moqx, Meta moxygen, Cloudflare moq-rs, Quicr libquicr, Meetecho imquic, OzU moqtail, Nokia — across draft-14/16/18 and both transports, using the moq-interop-runner cases plus a multi-subscriber pub-sub bench. The point-in-time matrix is in PERFORMANCE.md; the relay catalog with per-endpoint notes is tests/relays.json.

Interop client

Runs the six standard runner cases (setup-only, announce-only, publish-namespace-done, subscribe-error, announce-subscribe, subscribe-before-announce) plus fetch and join, emitting TAP.

python -m aiomoqt.tools.moq_interop_client -r "moqt://relay.ex.com:4433"              # all, draft auto
python -m aiomoqt.tools.moq_interop_client -r "moqt://relay.ex.com:4433" --draft 16
python -m aiomoqt.tools.moq_interop_client -r "moqt://relay.ex.com:4433" -t subscribe-error
python -m aiomoqt.tools.moq_interop_client -l                                         # list cases

aiomoqt-relay

aiomoqt-relay (module moq_interop_relay, also installed as moq-interop-relay) is a MoQT relay built on aiomoqt, serving raw QUIC and WebTransport at drafts 14 to 22. It forwards objects from an upstream publisher to downstream subscribers, fans one upstream subscription out to several subscribers, serves both publish flows (PUBLISH_NAMESPACE and bare PUBLISH), offers PUBLISH to SUBSCRIBE_TRACKS subscribers, and dials upstream origins with --upstream. A recent-object cache per track (8 groups, 32 MiB) answers standalone and joining FETCH; one relay-wide budget (--cache-bytes, default 256 MiB) caps all tracks together by dropping the oldest group of the least recently used track, and past the budget plus 32 MiB the largest live group goes uncached. Cache and send lanes charge each object its payload, its properties and about 224 bytes of record overhead. A late subscriber starts at the live edge. For --startup-hold seconds after start (default 2, 0 = off), a SUBSCRIBE for a track no publisher serves yet waits for one, so viewers that reconnect after a restart before their publishers are answered, not refused. It runs the public relay at aiomoqt-relay.marzresearch.net, and moxygen's moqtest conformance suite and the interop runner test against it.

Limits: no authentication, authorization or rate limiting; each subscriber's send lane holds up to 4 MiB, counting what its streams still hold in the transport, and a subscriber that falls further behind loses whole groups, their streams reset, and a lane that completes no write for 10 s logs one lane stalled WARNING; upstream Forward State is raised for a viewer and lowered once no viewer has forwarded the track for --pause-after seconds (default 60, 0 = never); a subscriber's objects older than its OBJECT_DELIVERY_TIMEOUT (the smaller of the publisher's and the subscriber's) or the track's MAX_CACHE_DURATION, or with neither set its live objects older than --lag-limit (default 5 s, 0 = never), are dropped, the subgroup reset with DELIVERY_TIMEOUT and the rest of the group skipped; past MAX_CACHE_DURATION a cached object is not served and a FETCH for it goes upstream. Written in Python, it carries less traffic than the Rust and C++ relays. Because the relay re-encodes everything it forwards, a conformance client checking those objects is checking aiomoqt's encoder, not merely its routing.

aiomoqt-relay --port 4443 \
    --cert cert.pem --key key.pem --dual     # raw QUIC + WT on one port

--quic serves raw QUIC only, --quic-port N runs the legacy second listener for runners that expect distinct endpoints, and --upstream URL (repeatable) dials an origin for tracks no inbound publisher serves — the arrangement the moq-test conformance suite expects.

Relay probe

Reads a relay list, performs a real CLIENT_SETUP / SERVER_SETUP handshake per (endpoint × draft), and writes a JSON status report. Accepts CLI flags, environment variables, or both (CLI overrides env).

python -m aiomoqt.tools.relay_probe --relays-file relays.json -o status.json
RELAYS_FILE=relays.json OUTPUT_FILE=status.json python -m aiomoqt.tools.relay_probe
python -m aiomoqt.tools.relay_probe --relays-file relays.json -o status.json --interval 300

Performance

Throughput at this layer is bounded by aiopquic, picoquic, and the kernel UDP path beneath it. Methodology, observed figures, the paced-vs-unpaced distinction, TX budget tuning, and the full tool matrix are in PERFORMANCE.md.

Development

git clone https://github.com/gmarzot/aiomoqt.git
cd aiomoqt
python3 -m venv .venv && source .venv/bin/activate
uv pip install -e ".[test]"    # or: pip install -e ".[test]"
pytest aiomoqt/tests/

Install editable. The standalone bench scripts and the cross-importing loopback tests resolve certs/ and sibling modules relative to the working tree; a non-editable install skips the loopback suites with "TLS certs not found in certs/". pytest generates certs/ on first run. loopback_bench and adaptive_bench (loopback) need no setup: when they find no cert, they generate a self-signed localhost cert once into ~/.cache/aiomoqt/ ($XDG_CACHE_HOME), using the openssl CLI or the cryptography package. Servers (pub_server, aiomoqt-relay) never self-sign; give them --cert/--key or generate one:

mkdir -p certs && openssl req -x509 -newkey rsa:2048 -nodes -days 3650 \
  -keyout certs/key.pem -out certs/cert.pem -subj "/CN=localhost" \
  -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"

Building against a local aiopquic

The PyPI aiopquic wheel is portable (any CPU of the architecture). A locally compiled aiopquic is host-tuned (-O3 -march=native -flto, plus picotls Fusion AES-GCM on x86_64); build from source when benchmarking or targeting known hardware.

git clone https://github.com/gmarzot/aiopquic.git
cd aiopquic && git submodule update --init --recursive && ./build_picoquic.sh

With a separate venv per repo, install the local aiopquic into the aiomoqt venv first, so the dependency is already satisfied and no wheel is fetched:

# in the aiomoqt venv:
uv pip install -e ~/aiopquic    # local source, editable — BEFORE aiomoqt
uv pip install -e '.[test]'

Confirm which build is in use:

python -c "import aiopquic; print(aiopquic.__file__)"   # must be <repo>/src/aiopquic/…
python -m aiomoqt.versions                              # paths + picoquic/picotls SHAs

Both venvs must use the same Python version. C/Cython changes need a rebuild (./build_picoquic.sh, then uv pip install -e ~/aiopquic); pure-Python edits are live.

Reporting issues

Include the version report — it captures aiomoqt, aiopquic, and the picoquic + picotls submodule SHAs aiopquic was built from:

python -m aiomoqt.versions

aiomoqt:   0.14.0 (~/.venv/lib/python3.13/site-packages/aiomoqt) [2026-10-27 17:17]
aiopquic:  0.7.0 (~/.venv/lib/python3.13/site-packages/aiopquic) [2026-10-27 17:17]
  - picoquic:  1.1.53.0 (2026.10.22.70b5fd5-0-70b5fd5a) [2026-10-22]
  - picotls:   master (f06553b8) [2026-09-30]
  - openssl:   OpenSSL 3.5.8 (~/.venv/lib/python3.13/site-packages/aiopquic.libs/libcrypto-c37e28d8.so.3.5.8)

Limitations

• relay_probe -f and tests/relays.json use different schemas. The probe expects a top-level array of {id, name, endpoints:[{url}]}; the interop catalog is {relays:[{name, urls:{...}, drafts:[...]}]}. Use --url for single-endpoint probes until the loader accepts both.
• aiomoqt-relay has no authentication or rate limiting; a subscriber that falls more than its send lane's bound behind loses whole groups. One session holds at most --max-namespaces-per-session (256) namespace requests and --max-subscriptions-per-session (1024) SUBSCRIBEs; more are refused with EXCESSIVE_LOAD (INTERNAL_ERROR at d14/d16). Its cache budget (--cache-bytes) never drops a subscribed track's live group, so live groups can exceed it by up to 32 MiB before the largest is dropped from the cache. See aiomoqt-relay.

Roadmap

• Single-port interop-runner adapter — one relay image and one port via serve_dual()
• Binding-deadline reaper for data streams whose track-alias binding never completes
• Broader CI test coverage — the workflow currently names a subset of the suite explicitly

Contributing

Fork, branch, and open a pull request. For major changes, open an issue first.

Resources

• MoQT Specification
• Media Over QUIC Working Group
• MoQ Interop Runner
• OpenMOQ — github.com/openmoq
• aiomoqt GitHub Repository

Author

Giovanni Marzot — gmarzot@marzresearch.net

A Marz Research project.

Acknowledgements

This project takes inspiration from, and has benefited from the great work done by the OpenMOQ team (github.com/openmoq), and the continued efforts of the MOQ IETF WG.