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 only — see Limitations).

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).

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

# 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.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

Interop relay

moq_interop_relay is a test fixture, not a production relay. 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), and dials upstream origins with --upstream. It has no group cache — so no joining FETCH and a late subscriber sees only what arrives next — no forward-state propagation, no PUBLISH forwarding to SUBSCRIBE_TRACKS subscribers, no authentication, and no backpressure. Use moxygen, moq-rs, or another real relay for any workload.

Its purpose is to exercise aiomoqt's server-side primitives — MOQTServer, the announce and subscribe handlers, serve_dual() — and to be driven by an external conformance suite, so both sides of the stack are covered. Because the relay re-encodes everything it forwards, a conformance client checking those objects is checking aiomoqt's encoder, not merely its routing.

python -m aiomoqt.tools.moq_interop_relay --bind 0.0.0.0 --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. For the standalone bench scripts, generate them directly:

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.13.1 (~/.venv/lib/python3.13/site-packages/aiomoqt) [2026-10-21 16:43]
aiopquic:  0.6.1 (~/.venv/lib/python3.13/site-packages/aiopquic) [2026-10-21 16:43]
  - picoquic:  1.1.53.0 (2026.10.08.c908872-0-c9088724) [2026-10-08]
  - 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

• WebTransport datagram TX is not implemented. Datagram publishing and reception work over raw QUIC; on WebTransport, datagram reception works but StreamMapping.DATAGRAM publishing raises. Use PER_GROUP mapping over WebTransport.
• 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.
• moq_interop_relay is a conformance fixture, not a production relay — it forwards and fans out, but has no group cache (hence no joining FETCH), no forward-state propagation, no authz, and no backpressure. See Interop relay.

Roadmap

• Single-port interop-runner adapter — one relay image and one port via serve_dual()
• WebTransport datagram TX
• 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.