Skip to content

MQTT routing#

FastStream MQTT support is implemented on top of zmqtt — a pure asyncio MQTT 3.1.1 / 5.0 client with no extra runtime dependencies. You can use the underlying zmqtt.MQTTClient via the broker connection when you need APIs not wrapped by FastStream.

Why MQTT#

MQTT is a lightweight publish/subscribe protocol designed for constrained networks and high fan-out. Messages are addressed by topic strings; brokers route publishes to subscribers whose topic filters match (including + and # wildcards).

Compared to Kafka or RabbitMQ, MQTT emphasizes simple topic namespaces, optional persistent sessions, and QoS levels built into the protocol. Choose MQTT when your infrastructure or devices already speak MQTT, or when you want broker-mediated pub/sub without managing exchanges or partitions yourself.

FastStream MQTTBroker#

Import the broker and optional helpers from faststream.mqtt:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from faststream import FastStream
from faststream.mqtt import MQTTBroker, MQTTMessage, QoS

broker = MQTTBroker("mqtt://localhost:1883", version="5.0")
app = FastStream(broker)


@broker.subscriber(
    "sensors/+/temp",
    qos=QoS.AT_LEAST_ONCE,
    # shared="workers",  # optional: $share/workers/... for load-balanced consumers
    # max_workers=4,     # optional: concurrent handler tasks
)
async def on_temp(degrees: float, message: MQTTMessage) -> None:
    print(message.raw_message.topic)


@app.after_startup
async def publish_demo() -> None:
    await broker.publish(21.5, "sensors/room1/temp", qos=QoS.AT_LEAST_ONCE)

Connection parameters#

The broker constructor mirrors common zmqtt.MQTTClient options:

Parameter Role
url Broker URL. mqtt:// uses plain TCP and port 1883; mqtts:// uses TLS and port 8883. Username and password can be included in the URL.
host, port Legacy parameters retained for backward compatibility.
version "3.1.1" or "5.0" — selects protocol features and how FastStream maps metadata (see MQTT versions).
client_id Client identity string.
security Pass SASLPlaintext(username, password) or BaseSecurity(ssl_context) for credentials and TLS (see Security).
keepalive, clean_session Session behaviour.
will Optional Will (from faststream.mqtt) published by the broker after an unexpected disconnect. WillProperties are supported with MQTT 5.0.
reconnect Optional ReconnectConfig (from faststream.mqtt) for automatic reconnect with backoff.
on_connection_recovery_failed Optional async callback invoked after a running connection cannot be restored. FastStream passes the callback directly to zmqtt.
session_expiry_interval MQTT 5.0 session expiry (seconds).
session_replay_buffer_size Maximum unmatched messages held while a resumed persistent session waits for local subscriptions. The default is 1000; 0 is unbounded.
session_replay_timeout Seconds to wait for local subscriptions before dropping unmatched replay messages without acknowledging them. The default is 30.
mqtt_connect_timeout Seconds to wait for the broker's CONNACK during the MQTT connect handshake (default 30); raises MQTTTimeoutError (from zmqtt), and is retried when reconnect is enabled.

Routers reuse the same API via MQTTRouter / MQTTRoute (see routers).

Terminal connection recovery failure#

When zmqtt exhausts the configured runtime reconnect attempts, it invokes the user-provided on_connection_recovery_failed callback and raises the terminal error from active subscription iterators. FastStream stops the failed consumer task instead of restarting it with the same disconnected client.

FastStream does not stop the application or create a new client automatically. The broker remains part of the running application, and await broker.ping() returns False. Use the callback or your application's health check to report the failure and let your deployment policy decide whether to restart the process.

Persistent-session startup replay#

With a stable client_id, clean_session=False, and a positive MQTT 5.0 session_expiry_interval, a broker can replay queued messages immediately after CONNACK, before FastStream starts its local subscribers. zmqtt temporarily holds those messages in its session replay buffer and routes them after the matching subscriptions are ready. Use session_replay_buffer_size and session_replay_timeout to size that startup window for the expected backlog.