transport

MOQ (Media over QUIC) transport implementation for Pipecat.

Uses the upstream moq Python library (moq-rs) for the QUIC connection, MOQ session, announcement discovery, subscription routing, group/frame framing, codec-specific catalog management, and Opus encode/decode + resampling for raw audio tracks. This module just wires it into the pipecat Frame pipeline.

Each participant publishes under a per-participant broadcast path <namespace>/<participant_id> (e.g. pipecat/bot0); the bot subscribes to the peer at <namespace>/<peer_id>. Audio rides on a single Opus track; RTVI JSON rides on a fixed-name transcript.json.z track carried by moq’s JSON stream helper (publish_json_stream / subscribe_json_stream). The stream is an ordered, lossless append-log of records — every message is delivered in order, unlike the JSON snapshot helper (publish_json / subscribe_json) which collapses to the latest value and would drop RTVI events a slow consumer fell behind on. Compression is enabled (hence the .z suffix). The transcript is a side-channel, like moq-boy’s status/command tracks: it deliberately bypasses the catalog (which only describes media renditions), so the browser reads it by the well-known name rather than by catalog discovery.

Both directions carry the same shape: the bot publishes bot-side RTVI events on its own transcript track and subscribes to the client’s transcript track for client-side traffic (client-ready for protocol negotiation, typed text input, function-call results, etc). This makes MoQ a full bidirectional RTVI transport, on par with the Daily and WebSocket transports.

Two modes:

• Server mode (serve=True, currently the only supported mode): the bot binds its own UDP socket via moq.Server and accepts the browser’s direct connection. Removes the need for a separate moq-relay process for local dev. The self-signed cert fingerprints are exposed via MOQTransport.cert_fingerprints so a browser can pin them.

• Client mode (default): MoQ client mode is not yet supported. The bot would dial a relay at relay_url (or the constructor’s host/port/path); transport-level wiring exists but the runner blocks this mode at arg-parse time until the cert-fingerprint plumbing to the browser is finished and we’ve validated the flow against an external relay. See pipecat.runner.moq._validate_moq_args() for the guard.

class pipecat.transports.moq.transport.MOQTrackType(*values)

Bases: IntEnum

Types of media tracks the transport surfaces in event callbacks.

AUDIO = 1

VIDEO = 2

DATA = 3

class pipecat.transports.moq.transport.MOQTrack(broadcast_path: str, name: str, track_type: MOQTrackType = MOQTrackType.DATA)

Bases: object

Identifies a MOQ track for event callbacks.

Parameters:

• broadcast_path – The full broadcast path (e.g. pipecat/bot0).

• name – The track name (e.g. bot-audio).

• track_type – The track media type.

broadcast_path: str

name: str

track_type: MOQTrackType = 3

property full_name: str

Get the full track identifier.

class pipecat.transports.moq.transport.MOQParams(*, audio_out_enabled: bool = False, audio_out_sample_rate: int = 24000, audio_out_channels: int = 1, audio_out_bitrate: int = 96000, audio_out_10ms_chunks: int = 4, audio_out_mixer: Mapping[str | None, ~pipecat.audio.mixers.base_audio_mixer.BaseAudioMixer] | None=None, audio_out_destinations: list[str] = <factory>, audio_out_end_silence_secs: int = 2, audio_out_auto_silence: bool = True, audio_in_enabled: bool = False, audio_in_sample_rate: int = 16000, audio_in_channels: int = 1, audio_in_filter: BaseAudioFilter | None = None, audio_in_stream_on_start: bool = True, audio_in_passthrough: bool = True, video_in_enabled: bool = False, video_out_enabled: bool = False, video_out_is_live: bool = False, video_out_width: int = 1024, video_out_height: int = 768, video_out_bitrate: int | None = None, video_out_framerate: int = 30, video_out_color_format: str = 'RGB', video_out_codec: str | None = None, video_out_destinations: list[str] = <factory>, relay_url: str | None = None, namespace: str = 'pipecat', participant_id: str = 'bot0', peer_id: str = 'client0', audio_out_track: str = 'bot-audio', transcript_track: str = 'transcript.json.z', verify_ssl: bool = True, connection_timeout: float = 30.0, serve: bool = False, serve_bind: str | None = None, serve_tls_host: str = 'localhost', serve_tls_cert: str | None = None, serve_tls_key: str | None = None, audio_in_max_latency_ms: int = 500, audio_out_frame_ms: int = 20, audio_out_max_buffer_ms: int = 25000)

Bases: TransportParams

Configuration parameters for MOQ transport.

Parameters:

• relay_url – Full relay URL (e.g. https://relay.example.com:4080/moq). If unset, the transport composes one from the constructor’s host/port/path. Ignored in serve mode.

• namespace – Top-level namespace shared by all participants.

• participant_id – This bot’s id; the bot publishes under <namespace>/<participant_id>.

• peer_id – The id of the peer (browser/client) the bot subscribes to: <namespace>/<peer_id>.

• audio_out_track – Name of the bot’s outgoing audio track.

• transcript_track – Name of the bot’s outgoing transcript track. A fixed-name JSON stream track carrying RTVI messages as a lossless, ordered append-log (moq’s publish_json_stream / subscribe_json_stream, compression on), discovered by convention rather than via the catalog.

• verify_ssl – Verify the relay’s TLS certificate. Client mode only.

• connection_timeout – Seconds to wait for the peer broadcast to be announced before giving up.

• serve – When True, the bot binds its own UDP socket and accepts incoming MOQ sessions instead of dialing a relay.

• serve_bind – Address to bind in serve mode (e.g. "[::]:4080"). Defaults to "[::]:<port>" based on the constructor’s port.

• serve_tls_host – Hostname to use in the generated self-signed certificate when serve_tls_cert/serve_tls_key aren’t provided. The browser pins this cert via its SHA-256 fingerprint, so the value just needs to match what the browser sees in the URL (typically "localhost").

• serve_tls_cert – Path to a PEM-encoded TLS certificate chain. If unset alongside serve_tls_key, a self-signed cert is generated on startup.

• serve_tls_key – Path to the PEM-encoded private key matching serve_tls_cert.

• audio_out_sample_rate – Sample rate the bot publishes its audio at (Hz). Pipecat hands us audio at whatever rate the TTS produces; the library resamples to the nearest Opus-supported rate before encoding.

• audio_in_sample_rate – Sample rate the pipeline expects to receive user audio at (Hz). Decoded Opus is resampled to this rate before being pushed downstream.

• audio_in_max_latency_ms – How long subscribe_audio() will wait for a late frame before skipping ahead. Lower = more interactive (fewer fills, more drops on bad networks); higher = smoother audio with more glass-to-glass delay.

• audio_out_frame_ms – Opus frame duration for the bot’s audio output. Must be 2, 5, 10, 20, 40, or 60. 20 ms is the real-time default.

• audio_out_max_buffer_ms – How far ahead of real-time the bot is allowed to write audio. The bot writes TTS faster than real-time with future-dated timestamps so the browser player (@moq/watch) can buffer and play at the encoded pace; this paces publish_audio so the in-flight buffer never grows past this many milliseconds. Keep it a little under the player’s buffer ceiling (MoqTransportOptions.audioBufferMaxMs, 30s) so the producer self-limits below the player’s drop ceiling and the player never has to drop. On interruption the pacing clock is re-anchored (see reset_audio_pacing()) so the next utterance isn’t delayed by the previous buffer.

relay_url: str | None

namespace: str

participant_id: str

peer_id: str

audio_out_track: str

transcript_track: str

verify_ssl: bool

connection_timeout: float

serve: bool

serve_bind: str | None

serve_tls_host: str

serve_tls_cert: str | None

serve_tls_key: str | None

audio_out_sample_rate: int

audio_in_sample_rate: int

audio_in_max_latency_ms: int

audio_out_frame_ms: int

audio_out_max_buffer_ms: int

class pipecat.transports.moq.transport.MOQCallbacks(*, on_connected: Callable[[], Awaitable[None]], on_disconnected: Callable[[], Awaitable[None]], on_client_connected: Callable[[], Awaitable[None]], on_client_disconnected: Callable[[], Awaitable[None]], on_track_subscribed: Callable[[MOQTrack], Awaitable[None]], on_error: Callable[[str, Exception], Awaitable[None]], on_audio_received: Callable[[bytes, int], Awaitable[None]], on_message_received: Callable[[dict], Awaitable[None]])

Bases: BaseModel

Callback handlers bridging MOQTransportClient to MOQTransport.

Decouples the moq-rs session/connection logic from pipecat’s Frame pipeline: MOQTransportClient only ever calls these, it never reaches into MOQInputTransport/MOQOutputTransport or pipecat frame types directly.

Parameters:

• on_connected – Called when the MOQ session (client or server) is established.

• on_disconnected – Called when the session ends.

• on_client_connected – Called when the peer’s broadcast is announced.

• on_client_disconnected – Called when the peer’s broadcast goes away.

• on_track_subscribed – Called when a remote track subscription succeeds.

• on_error – Called when the underlying transport errors.

• on_audio_received – Called with decoded mono PCM audio from the peer’s audio track.

• on_message_received – Called with each RTVI message from the peer’s transcript stream.

on_connected: Callable[[], Awaitable[None]]

on_disconnected: Callable[[], Awaitable[None]]

on_client_connected: Callable[[], Awaitable[None]]

on_client_disconnected: Callable[[], Awaitable[None]]

on_track_subscribed: Callable[[MOQTrack], Awaitable[None]]

on_error: Callable[[str, Exception], Awaitable[None]]

on_audio_received: Callable[[bytes, int], Awaitable[None]]

on_message_received: Callable[[dict], Awaitable[None]]

class pipecat.transports.moq.transport.MOQTransportClient(params: MOQParams, url: str, serve_bind: str, callbacks: MOQCallbacks)

Bases: object

Owns the moq-rs QUIC session/connection for a single MOQ transport.

Mirrors the <Provider>TransportClient pattern used by the other transports (e.g. DailyTransportClient, SmallWebRTCClient): it owns dialing/serving, the publish broadcast, and forwarding the peer’s audio/transcript, and talks back to MOQTransport only through callbacks — it has no knowledge of pipecat Frames or the input/output transports.

__init__(params: MOQParams, url: str, serve_bind: str, callbacks: MOQCallbacks)

Initialize the MOQ transport client.

The publish broadcast and its transcript track are created here, synchronously, so the output processor can begin writing into them before the async session bring-up in _run() finishes — otherwise the bot’s first ~hundreds of ms of audio gets silently dropped while we’re still dialing the relay.

Parameters:

• params – MOQ configuration parameters.

• url – Full relay URL to dial in client mode.

• serve_bind – Address to bind in serve mode.

• callbacks – Event/data callbacks back to the owning transport.

async setup(setup: FrameProcessorSetup)

Capture the task manager from the input/output processors.

Both processors forward their setup here; we only need to store the reference once since they share the same task manager.

connect()

Register a holder and start the MOQ session on the first call.

Called by both MOQInputTransport.start() and MOQOutputTransport.start(), after setup() has wired in the task manager. Only the first call actually dials/serves; later calls just record another holder, balanced against stop()/cancel() releasing it.

open_audio_track(sample_rate: int)

Open the bot’s audio track via publish_audio.

Called by MOQOutputTransport.start once it knows the pipeline’s output sample rate. No-op if the track is already open or audio output is disabled.

reset_audio_pacing()

Re-anchor the publish_audio pacing clock to wall-clock now.

Called from MOQOutputTransport.process_frame on InterruptionFrame so the next utterance plays immediately instead of waiting for the (now-cancelled) previous one to finish in pacing time.

Part of the publish_audio pacing workaround; goes away once moq-rs exposes a flush primitive on AudioProducer. See publish_audio().

async publish_audio(audio: bytes)

Push a PCM chunk to the bot’s audio track with real PTS, paced to a cap.

The library does the Opus encode + resample inside the FFI, so we just write S16 PCM bytes.

Each chunk is stamped with a monotonic presentation timestamp so the browser player (@moq/watch) can buffer the future-dated frames and play them at the encoded pace. AudioProducer.write() is fire-and-forget, so to bound how far ahead the bot runs we pace the writes against a virtual clock: each call advances the clock by the chunk’s audio duration, and we sleep until wall-clock is within audio_out_max_buffer_ms (25s) of it. That keeps the in-flight buffer a little under the player’s drop ceiling (MoqTransportOptions.audioBufferMaxMs, 30s), so the producer self-limits below the consumer’s cap. Interruptions are flushed on the browser side (reset() on user-started-speaking); the pacing clock is re-anchored here (see reset_audio_pacing()) so the next utterance isn’t delayed by the previous buffer.

async wait_for_audio_drain(jitter_buffer_margin_s: float = 0.3) → None

Block until the pacing clock catches up to wall-clock, then a bit more.

publish_audio paces each write against _publish_audio_clock, which tracks the presentation-time deadline of the last chunk written to the moq audio stream. When wall-clock reaches that value, the last audio chunk has (approximately) been consumed by the browser’s player. Add a small margin for the client-side jitter buffer to drain past the final sample before we signal session-ending and tear the transport down.

No-op when no audio has been published yet or the clock has already caught up. Capped at audio_out_max_buffer_ms as a safety net so a bogus clock value can’t stall shutdown.

publish_transcript(message)

Append an RTVI message to the transcript JSON stream.

message is a JSON-serializable value (the RTVI message dict); the stream helper serializes and frames it, appending one record to the ordered log so no message is dropped.

property cert_fingerprints: list[str]

SHA-256 fingerprints (hex) of the serving cert (server mode only).

Populated once the bot has bound; empty in client mode or before _run() has reached the listen step. Useful for telling a browser client which self-signed cert to pin.

async disconnect()

Disconnect from the MOQ relay.

Sends an intra-transport session-ending notification on the transcript stream before tearing anything down. This gives the browser-side MoQ transport a chance to disable auto-reconnect, drain its jitter buffer, and close the audio decoder cleanly — without that heads-up, WebTransport just vanishes underneath Chrome’s WebCodecs decoder and can crash the renderer (“Aw snap”).

async cleanup()

Unconditionally tear down the underlying MOQ connection.

Called as a last-resort safety net from MOQInputTransport.cleanup/MOQOutputTransport.cleanup (the FrameProcessor lifecycle hook, always invoked regardless of whether stop()/cancel() already released the session), so we guard against repeating the work. Prefer stop()/ cancel() to end the session during normal shutdown — those wait for every holder to release before tearing anything down.

async stop()

Release this holder’s claim on the session, draining buffered audio first.

Called by both MOQInputTransport.stop() and MOQOutputTransport.stop(). Safe regardless of which one runs first: the session only actually disconnects once both have released their claim, and by the time the last one does, every audio frame written by the output side has already been drained. Draining twice (once per caller) is harmless — the second call just observes the clock has already caught up.

async cancel()

Release this holder’s claim on the session immediately, without draining.

Called by both MOQInputTransport.cancel() and MOQOutputTransport.cancel(): cancellation is a hard stop, so there’s nothing to drain, and the session disconnects as soon as both have released their claim.

class pipecat.transports.moq.transport.MOQInputTransport(client: MOQTransportClient, params: MOQParams, **kwargs)

Bases: BaseInputTransport

MOQ input transport: subscribes to the peer’s audio track.

__init__(client: MOQTransportClient, params: MOQParams, **kwargs)

Initialize the MOQ input transport.

Parameters:

• client – MOQTransportClient instance managing the MoQ session.

• params – MOQ transport configuration parameters.

• **kwargs – Additional arguments passed to the parent class.

async setup(setup: FrameProcessorSetup)

Forward setup to the shared MOQTransportClient so it can create tasks.

Parameters:

setup – Configuration object containing setup parameters.

async start(frame: StartFrame)

Auto-connect to the MOQ relay when the pipeline starts.

Parameters:

frame – The start frame containing initialization parameters.

async stop(frame: EndFrame)

Stop the MOQ input transport.

Releases this transport’s claim on the shared session (see MOQTransportClient.stop()). EndFrame reaches this (upstream) input transport before it reaches MOQOutputTransport downstream, so this call alone doesn’t end the session — the client waits for the output transport to release its own claim (after draining its buffered audio) before actually disconnecting.

Parameters:

frame – The end frame signaling transport shutdown.

async cancel(frame: CancelFrame)

Cancel the MOQ input transport.

Releases this transport’s claim on the shared session. Unlike stop(), cancellation is immediate — there’s no audio to drain.

Parameters:

frame – The cancel frame signaling immediate cancellation.

async cleanup()

Cleanup resources.

async push_received_audio(audio: bytes, sample_rate: int)

Push a received audio frame downstream.

Parameters:

• audio – Raw mono 16-bit PCM audio bytes.

• sample_rate – Sample rate of audio, in Hz.

async push_received_message(message: dict)

Push a received RTVI message to RTVIProcessor upstream.

Called by MOQTransport for every record delivered on the peer’s transcript stream. RTVIProcessor sits upstream of this input transport (it’s prepended by PipelineWorker when enable_rtvi=True), so pushing upstream is the direct route.

Parameters:

message – The parsed RTVI message.

class pipecat.transports.moq.transport.MOQOutputTransport(client: MOQTransportClient, params: MOQParams, **kwargs)

Bases: BaseOutputTransport

MOQ output transport: writes audio + transcript frames to MOQ tracks.

__init__(client: MOQTransportClient, params: MOQParams, **kwargs)

Initialize the MOQ output transport.

Parameters:

• client – MOQTransportClient instance managing the MoQ session.

• params – MOQ transport configuration parameters.

• **kwargs – Additional arguments passed to the parent class.

async setup(setup: FrameProcessorSetup)

Forward setup to the shared MOQTransportClient so it can create tasks.

Parameters:

setup – Configuration object containing setup parameters.

async start(frame: StartFrame)

Start the MOQ output transport.

Parameters:

frame – The start frame containing initialization parameters.

async stop(frame: EndFrame)

Stop the MOQ output transport.

On graceful end (an EndFrame — pushed by EndWorkerFrame from an end_call tool, for example) let the moq audio buffer drain fully before releasing this transport’s claim on the shared session (see MOQTransportClient.stop()). Without this the bot’s final TTS utterance gets clipped: base_output’s stop() unblocks as soon as every audio frame has been written to moq’s pacing queue, but moq buffers up to audio_out_max_buffer_ms (25s) of paced audio that still has to reach the browser and drain the client jitter buffer before it actually plays.

Parameters:

frame – The end frame signaling transport shutdown.

async cancel(frame: CancelFrame)

Cancel the MOQ output transport.

Releases this transport’s claim on the shared session. Unlike stop(), cancellation is immediate — there’s no audio to drain.

Parameters:

frame – The cancel frame signaling immediate cancellation.

async cleanup()

Cleanup resources.

async process_frame(frame: Frame, direction: FrameDirection)

Reset the publish_audio pacing clock on interruption.

The pacing in write_audio_frame keeps moq’s in-flight buffer bounded so InterruptionFrame can actually stop playback in the browser — but the pacing clock needs to be re-anchored to now so the next utterance plays immediately instead of catching up.

Parameters:

• frame – The frame to process.

• direction – The direction of frame flow in the pipeline.

async send_message(frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame)

Publish a transport message (RTVI JSON) on the transcript stream.

Parameters:

frame – The transport message frame to send.

async write_audio_frame(frame: OutputAudioRawFrame) → bool

Write an audio frame to the bot’s audio track.

Parameters:

frame – The output audio frame to write.

Returns:

True if the audio frame was written successfully, False otherwise.

class pipecat.transports.moq.transport.MOQTransport(params: MOQParams, host: str = 'localhost', port: int = 4080, path: str = '/moq', input_name: str | None = None, output_name: str | None = None)

Bases: BaseTransport

MOQ transport that connects to a MOQ relay via the moq library.

Example:

transport = MOQTransport(
    params=MOQParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
        namespace="my-room",
    ),
    host="localhost",
    port=4080,
)

@transport.event_handler("on_connected")
async def on_connected(transport):
    print("Connected to MOQ relay")

Event handlers available:

• on_connected — connection to relay established

• on_disconnected — connection lost / closed

• on_client_connected — peer broadcast announced (client joined)

• on_client_disconnected — peer broadcast went away

• on_track_subscribed — remote track subscription succeeded

• on_error — error in the underlying transport

__init__(params: MOQParams, host: str = 'localhost', port: int = 4080, path: str = '/moq', input_name: str | None = None, output_name: str | None = None)

Initialize the MOQ transport.

Parameters:

• params – MOQ configuration parameters.

• host – Relay host address.

• port – Relay port number.

• path – MOQ endpoint path on the relay.

• input_name – Optional name for the input processor.

• output_name – Optional name for the output processor.

input() → MOQInputTransport

Get the input transport for receiving media.

output() → MOQOutputTransport

Get the output transport for sending media.

property cert_fingerprints: list[str]

SHA-256 fingerprints (hex) of the serving cert (server mode only).

Populated once the bot has bound; empty in client mode or before the session bring-up has reached the listen step. Useful for telling a browser client which self-signed cert to pin.

async disconnect()

Disconnect from the MOQ relay.

See MOQTransportClient.disconnect() for details.