aiomoqt 0.5.5

Python asyncio implementation of the MoQT protocol

aiomoqt - Media over QUIC Transport (MoQT)

aiomoqt is an implementation of the MoQT protocol, based on asyncio and qh3.

Overview

This package implements the MoQT Specification with dual draft-14 and draft-16 support. It is designed for general use as an MoQT client and server library, supporting both 'publish' and 'subscribe' roles.

The architecture follows the asyncio.Protocol design pattern, and extends the qh3 QuicConnectionProtocol.

Features

• Dual draft-14/draft-16 support — version-conditional serialization following moxygen's branching pattern
• ALPN-based version negotiation (moq-00 for draft-14, moqt-16 for draft-16)
• Supports H3/WebTransport and raw QUIC transports
• Async context manager support for session connection management
• Version-agnostic user API: MOQTRequestError exception (no isinstance checks needed)
• High-level API for control messages with default and custom response handlers
• Low-level API for control and data message serialization/deserialization
• Draft-16 features: delta-encoded parameter keys, track extensions, unified REQUEST_OK/REQUEST_ERROR
• SubgroupHeader stream variants with flag encoding
• ObjectDatagram variants with flag encoding
• Delta-encoded object IDs in subgroup streams
• Interop test client implementing the moq-interop-runner test cases
• Relay version probe tool (relay_probe.py)

Interop Test Results

All 6 standard moq-interop-runner test cases pass against multiple relay implementations on both draft-14 and draft-16:

Relay	Draft	Transport	Tests	Result
OpenMoQ moxygen	draft-16	Raw QUIC	6/6	PASS
OpenMoQ moxygen	draft-14	Raw QUIC	6/6	PASS
Meta moxygen (fb.mvfst.net)	draft-16	Raw QUIC	6/6	PASS
Meta moxygen (fb.mvfst.net)	draft-14	Raw QUIC	6/6	PASS
Cloudflare moq-rs	draft-14	Raw QUIC	6/6	PASS
Red5 Pro	draft-14	Raw QUIC	6/6	PASS
Red5 Pro	draft-14	WebTransport	6/6	PASS
Quicr (libquicr)	draft-14	Raw QUIC	5/6	PASS

Test cases: setup-only, announce-only, publish-namespace-done, subscribe-error, announce-subscribe, subscribe-before-announce

# Run interop tests (draft-14, auto-detected)
python -m aiomoqt.examples.moq_interop_client -r "moqt://moqx-000.ci.openmoq.org:4433"

# Run interop tests (draft-16, explicit)
python -m aiomoqt.examples.moq_interop_client -r "moqt://moqx-000.ci.openmoq.org:4433" --draft 16

# Probe relay for supported versions
python -m aiomoqt.examples.relay_probe

Installation

Requires Python 3.12+ (tested on 3.12, 3.13, and 3.14).

Install using pip:

pip install aiomoqt

Or using uv:

uv pip install aiomoqt

Usage

Basic Client Example

import asyncio
from aiomoqt.client import MOQTClient
from aiomoqt.types import MOQTRequestError

async def main():
    client = MOQTClient('relay.example.com', 4433,
                        endpoint='moq', use_quic=True)

    async with client.connect() as session:
        try:
            await session.client_session_init()
            await session.subscribe(
                'namespace', 'track_name',
                wait_response=True,
            )
            # wait for session close, process data and control messages
            await session.async_closed()
        except MOQTRequestError as e:
            print(f"Request error: {e.error_code} {e.reason}")
            session.close()

asyncio.run(main())

The high-level control message API provides typical default values for most arguments and flexible type handling for input arguments. Messages which expect a response support blocking asyncio await via an optional flag (wait_response=True), returning the response message object. Asynchronous calls return the request message object immediately.

The message serialization/deserialization classes provide <msg>.serialize() which returns a qh3.Buffer with the entire message serialized in buf.data. The <MsgClass>.deserialize() call returns an instance populated from the deserialized data.

Examples

See aiomoqt/examples/ for complete working examples:

Example	Description
pub_example.py	Publisher client — SubgroupHeader streams or ObjectDatagrams
sub_example.py	Subscriber client — receives data from a relay
join_example.py	SUBSCRIBE + FETCH (joining mid-stream)
bench_relay.py	Combined pub/sub benchmark against a relay
bench_pub.py	High-performance publisher with configurable parameters
bench_sub.py	Subscriber with latency/jitter/loss statistics
server_example.py	WebTransport server (origin)
relay_probe.py	Relay version probe — detects draft-14/16 support
moq_interop_client.py	Interop test client (6 standard test cases, --draft flag)
moq_interop_client.py	Interop test client (6 standard test cases, TAP14 output)

Development

To set up a development environment:

git clone https://github.com/gmarzot/aiomoqt.git
cd aiomoqt
./bootstrap_python.sh
source .venv/bin/activate

Local Installation

uv pip install -e ".[test]"

Running Tests

pytest aiomoqt/tests/

Known Limitations

• Raw QUIC data streams: Data stream creation (SubgroupHeader streams, ObjectDatagrams on streams) currently requires WebTransport (_h3). Control plane messages work over raw QUIC, but publishers cannot send data objects in raw QUIC mode. This will be resolved with the aiopquic transport transition.

TODO

• Raw QUIC data stream support (via aiopquic transport transition)
• Dockerize interop test client for moq-interop-runner registration
• Simple relay implementation
• Support for MOQT File Format

Contributing

Contributions are welcome! If you'd like to contribute, please:

• Fork the repository on GitHub.
• Create a new branch for your feature or bug fix.
• Submit a pull request with a clear description of your changes.

For major changes, please open an issue first to discuss your proposal.

Resources

• MoQT Specification
• Media Over QUIC Working Group
• MoQ Interop Runner
• aiomoqt GitHub Repository

---

Acknowledgements

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