Skip to content

build_mqtt_url

faststream.mqtt.utils.build_mqtt_url #

build_mqtt_url(*, host: str, port: int, tls: bool) -> str

Build a credential-free MQTT connection URL.

Source code in faststream/mqtt/utils.py 
48
49
50
51
52
def build_mqtt_url(*, host: str, port: int, tls: bool) -> str:
    """Build a credential-free MQTT connection URL."""
    formatted_host = f"[{host}]" if ":" in host else host
    scheme = "mqtts" if tls else "mqtt"
    return f"{scheme}://{formatted_host}:{port}"