Normalize MQTT URI schemes from /notifiers (Postel's Law)

normalize_broker_uri now accepts bare hosts (broker.example.com),
host:port forms, and strips unknown schemes — previously these
either misparsed or were rejected.

extract_mqtt_broker_uris() handles both the spec notifiersResponse
shape ({"MQTT": {"URIS": [...]}}) and the VTN-RI list shape
([{"transport": "MQTT", "url": ...}]) with field-name tolerance.

vtn_supports_mqtt() rewritten to recognize both shapes properly
instead of substring-matching the str() of the response.

Closes OA3C-7q7.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: dcj

README.md

@@ -208,8 +208,20 @@ pid = ven.resolve_program_id("residential-pricing")
 ```python
 notifiers = ven.discover_notifiers()
 supports_mqtt = ven.vtn_supports_mqtt()
+
+# Extract broker URIs from /notifiers — handles both spec
+# ({"MQTT": {"URIS": [...]}}) and VTN-RI ([{"transport": "MQTT", "url": ...}])
+# response shapes. Returns [] if MQTT isn't advertised.
+uris = ven.get_mqtt_broker_uris()
+if uris:
+    mqtt = ven.add_mqtt(uris[0])
 ```
 
+URIs from `/notifiers` may use any of the `mqtt://`, `mqtts://`, `tcp://`, or
+`ssl://` schemes (or no scheme at all — `broker.example.com:1883`).
+`MqttChannel` and `MQTTConnection` accept all of these via
+`normalize_broker_uri`.
+
 ### VEN-scoped topic methods
 
 Default to the registered `ven_id` when called without arguments:

src/openadr3_client/__init__.py

@@ -6,7 +6,12 @@
     advertise_vtn,
     discover_vtns,
 )
-from openadr3_client.mqtt import MQTTConnection, MQTTMessage, normalize_broker_uri
+from openadr3_client.mqtt import (
+    MQTTConnection,
+    MQTTMessage,
+    extract_mqtt_broker_uris,
+    normalize_broker_uri,
+)
 from openadr3_client.notifications import (
     MqttChannel,
     NotificationChannel,
@@ -37,5 +42,6 @@
     # Helpers
     "extract_topics",
     "normalize_broker_uri",
+    "extract_mqtt_broker_uris",
     "detect_lan_ip",
 ]

src/openadr3_client/mqtt.py

@@ -24,24 +24,83 @@
 log = logging.getLogger(__name__)
 
 
+_KNOWN_MQTT_SCHEMES = frozenset({"mqtt", "mqtts", "tcp", "ssl"})
+_TLS_MQTT_SCHEMES = frozenset({"mqtts", "ssl"})
+
+
 def normalize_broker_uri(uri: str) -> tuple[str, int, bool]:
     """Translate an MQTT URI into (host, port, use_tls).
 
-    Supports mqtt://, mqtts://, tcp://, ssl:// schemes.
-    Adds default ports (1883 for plain, 8883 for TLS) when omitted.
+    Liberal in what is accepted (Postel's Law):
+
+    - Recognized schemes: ``mqtt://`` and ``tcp://`` (plain, default 1883),
+      ``mqtts://`` and ``ssl://`` (TLS, default 8883). Case-insensitive.
+    - Bare or unknown-scheme inputs (``broker.example.com``,
+      ``broker.example.com:1883``) are interpreted as plain MQTT.
     """
     parsed = urlparse(uri)
     scheme = parsed.scheme.lower()
+    if scheme not in _KNOWN_MQTT_SCHEMES:
+        # Unknown or missing scheme — strip any "<scheme>://" prefix and
+        # re-parse as plain MQTT. Be liberal in what we accept.
+        rest = uri.split("://", 1)[1] if "://" in uri else uri
+        parsed = urlparse(f"mqtt://{rest}")
+        scheme = "mqtt"
+
     host = parsed.hostname or "127.0.0.1"
+    use_tls = scheme in _TLS_MQTT_SCHEMES
+    port = parsed.port or (8883 if use_tls else 1883)
+    return host, port, use_tls
 
-    if scheme in ("mqtts", "ssl"):
-        use_tls = True
-        port = parsed.port or 8883
-    else:
-        use_tls = False
-        port = parsed.port or 1883
 
-    return host, port, use_tls
+def extract_mqtt_broker_uris(notifiers: Any) -> list[str]:
+    """Extract MQTT broker URIs from a ``/notifiers`` response.
+
+    Accepts both response shapes seen in the wild:
+
+    - **Spec shape** (`notifiersResponse`): ``{"WEBHOOK": true, "MQTT": {"URIS": [...], ...}}``
+    - **VTN-RI shape**: ``[{"transport": "MQTT", "url": "..."}, ...]``
+
+    Returns an empty list if no MQTT URI is advertised. Schemes are returned
+    as-is (callers should pass them through :func:`normalize_broker_uri`).
+    """
+    if not notifiers:
+        return []
+
+    if isinstance(notifiers, dict):
+        mqtt = notifiers.get("MQTT") or notifiers.get("mqtt")
+        if not isinstance(mqtt, dict):
+            return []
+        uris = mqtt.get("URIS") or mqtt.get("uris")
+        if isinstance(uris, list):
+            return [u for u in uris if isinstance(u, str)]
+        single = mqtt.get("URI") or mqtt.get("uri") or mqtt.get("url") or mqtt.get("broker")
+        return [single] if isinstance(single, str) else []
+
+    if isinstance(notifiers, list):
+        result: list[str] = []
+        for item in notifiers:
+            if not isinstance(item, dict):
+                continue
+            transport = (item.get("transport") or item.get("Transport") or "").upper()
+            if transport and transport != "MQTT":
+                continue
+            uri = (
+                item.get("url")
+                or item.get("uri")
+                or item.get("URI")
+                or item.get("URL")
+                or item.get("broker")
+                or item.get("endpoint")
+            )
+            if isinstance(uri, str):
+                result.append(uri)
+            uris = item.get("URIS") or item.get("uris")
+            if isinstance(uris, list):
+                result.extend(u for u in uris if isinstance(u, str))
+        return result
+
+    return []
 
 
 def _parse_payload(raw: bytes, topic: str) -> Any:

src/openadr3_client/ven.py

@@ -11,6 +11,7 @@
 from openadr3.entities.models import Event, Program
 
 from openadr3_client.base import BaseClient
+from openadr3_client.mqtt import extract_mqtt_broker_uris
 from openadr3_client.notifications import (
     MqttChannel,
     NotificationChannel,
@@ -130,15 +131,32 @@ def discover_notifiers(self) -> dict[str, Any] | None:
         return None
 
     def vtn_supports_mqtt(self) -> bool:
-        """Check if the VTN advertises MQTT notification support."""
+        """Check if the VTN advertises MQTT notification support.
+
+        Handles both the spec ``notifiersResponse`` dict (presence of an
+        ``MQTT`` key with a non-null binding object) and the VTN-RI list
+        shape (``[{"transport": "MQTT", ...}, ...]``).
+        """
         notifiers = self.discover_notifiers()
         if not notifiers:
             return False
-        # VTN-RI returns a list of notifier dicts with "transport" field
+        if isinstance(notifiers, dict):
+            mqtt = notifiers.get("MQTT") or notifiers.get("mqtt")
+            return bool(mqtt)
         if isinstance(notifiers, list):
-            return any(n.get("transport", "").upper() == "MQTT" for n in notifiers)
-        # Or it might be a dict with transport info
-        return "mqtt" in str(notifiers).lower()
+            return any(
+                isinstance(n, dict) and (n.get("transport") or "").upper() == "MQTT"
+                for n in notifiers
+            )
+        return False
+
+    def get_mqtt_broker_uris(self) -> list[str]:
+        """Return MQTT broker URIs the VTN advertises via ``/notifiers``.
+
+        Empty if MQTT is not advertised. Use :func:`normalize_broker_uri`
+        before passing to a connection if you need (host, port, use_tls).
+        """
+        return extract_mqtt_broker_uris(self.discover_notifiers())
 
     # -- Channel management --
 

tests/test_mqtt.py

@@ -8,6 +8,7 @@
 from openadr3_client.mqtt import (
     MQTTConnection,
     _parse_payload,
+    extract_mqtt_broker_uris,
     normalize_broker_uri,
 )
 
@@ -31,6 +32,83 @@ def test_tcp_scheme(self):
     def test_ssl_scheme(self):
         assert normalize_broker_uri("ssl://broker.local") == ("broker.local", 8883, True)
 
+    def test_uppercase_scheme(self):
+        assert normalize_broker_uri("MQTTS://broker.local") == ("broker.local", 8883, True)
+
+    def test_bare_host(self):
+        assert normalize_broker_uri("broker.example.com") == ("broker.example.com", 1883, False)
+
+    def test_bare_host_with_port(self):
+        assert normalize_broker_uri("broker.example.com:9883") == (
+            "broker.example.com",
+            9883,
+            False,
+        )
+
+    def test_bare_ip_with_port(self):
+        assert normalize_broker_uri("127.0.0.1:1883") == ("127.0.0.1", 1883, False)
+
+    def test_unknown_scheme_treated_as_plain(self):
+        # An unrecognized scheme falls back to mqtt:// — be liberal in what we accept.
+        assert normalize_broker_uri("foo://broker.local:1234") == ("broker.local", 1234, False)
+
+
+class TestExtractMqttBrokerUris:
+    def test_spec_dict_shape(self):
+        notifiers = {
+            "WEBHOOK": True,
+            "MQTT": {
+                "URIS": ["mqtts://broker.vtn.example.com", "mqtt://broker.vtn.example.com:1883"],
+                "serialization": "JSON",
+                "authentication": {},
+            },
+        }
+        assert extract_mqtt_broker_uris(notifiers) == [
+            "mqtts://broker.vtn.example.com",
+            "mqtt://broker.vtn.example.com:1883",
+        ]
+
+    def test_spec_dict_shape_lowercase_keys(self):
+        notifiers = {"mqtt": {"uris": ["mqtt://broker:1883"]}}
+        assert extract_mqtt_broker_uris(notifiers) == ["mqtt://broker:1883"]
+
+    def test_spec_dict_shape_no_mqtt(self):
+        assert extract_mqtt_broker_uris({"WEBHOOK": True}) == []
+
+    def test_spec_dict_shape_single_uri_field(self):
+        notifiers = {"MQTT": {"uri": "mqtt://broker:1883"}}
+        assert extract_mqtt_broker_uris(notifiers) == ["mqtt://broker:1883"]
+
+    def test_vtn_ri_list_shape(self):
+        notifiers = [
+            {"transport": "MQTT", "url": "mqtt://broker:1883"},
+            {"transport": "WEBHOOK", "url": "https://example.com"},
+        ]
+        assert extract_mqtt_broker_uris(notifiers) == ["mqtt://broker:1883"]
+
+    def test_vtn_ri_list_uri_field_variants(self):
+        # Different VTNs may use different field names for the broker URI.
+        for field in ("url", "uri", "URI", "URL", "broker", "endpoint"):
+            notifiers = [{"transport": "MQTT", field: "mqtt://broker:1883"}]
+            assert extract_mqtt_broker_uris(notifiers) == ["mqtt://broker:1883"], field
+
+    def test_vtn_ri_list_with_uris_array_in_item(self):
+        notifiers = [
+            {"transport": "MQTT", "URIS": ["mqtts://b1:8883", "tcp://b2:1883"]},
+        ]
+        assert extract_mqtt_broker_uris(notifiers) == ["mqtts://b1:8883", "tcp://b2:1883"]
+
+    def test_empty_inputs(self):
+        assert extract_mqtt_broker_uris(None) == []
+        assert extract_mqtt_broker_uris({}) == []
+        assert extract_mqtt_broker_uris([]) == []
+
+    def test_normalizable_tcp_scheme_round_trip(self):
+        # The whole point: VTN advertises tcp://, we accept it through to normalization.
+        uris = extract_mqtt_broker_uris([{"transport": "MQTT", "url": "tcp://broker:1883"}])
+        assert uris == ["tcp://broker:1883"]
+        assert normalize_broker_uri(uris[0]) == ("broker", 1883, False)
+
 
 class TestParsePayload:
     def test_json_object(self):

tests/test_ven.py

@@ -196,6 +196,59 @@ def test_vtn_supports_mqtt_error(self, mock_create):
         with VenClient(url="http://test", token="tok") as ven:
             assert ven.vtn_supports_mqtt() is False
 
+    @patch("openadr3_client.base.create_ven_client")
+    def test_vtn_supports_mqtt_spec_dict_shape(self, mock_create):
+        # Spec notifiersResponse: dict with MQTT binding object
+        mock_api = MagicMock()
+        mock_api.get_notifiers.return_value = _make_response(
+            200,
+            {
+                "WEBHOOK": True,
+                "MQTT": {
+                    "URIS": ["mqtts://broker.vtn.example.com"],
+                    "serialization": "JSON",
+                    "authentication": {},
+                },
+            },
+        )
+        mock_create.return_value = mock_api
+
+        with VenClient(url="http://test", token="tok") as ven:
+            assert ven.vtn_supports_mqtt() is True
+
+    @patch("openadr3_client.base.create_ven_client")
+    def test_vtn_supports_mqtt_spec_dict_no_mqtt(self, mock_create):
+        mock_api = MagicMock()
+        mock_api.get_notifiers.return_value = _make_response(200, {"WEBHOOK": True})
+        mock_create.return_value = mock_api
+
+        with VenClient(url="http://test", token="tok") as ven:
+            assert ven.vtn_supports_mqtt() is False
+
+    @patch("openadr3_client.base.create_ven_client")
+    def test_get_mqtt_broker_uris_vtn_ri_shape(self, mock_create):
+        mock_api = MagicMock()
+        mock_api.get_notifiers.return_value = _make_response(
+            200,
+            [{"transport": "MQTT", "url": "tcp://broker:1883"}],
+        )
+        mock_create.return_value = mock_api
+
+        with VenClient(url="http://test", token="tok") as ven:
+            assert ven.get_mqtt_broker_uris() == ["tcp://broker:1883"]
+
+    @patch("openadr3_client.base.create_ven_client")
+    def test_get_mqtt_broker_uris_spec_shape(self, mock_create):
+        mock_api = MagicMock()
+        mock_api.get_notifiers.return_value = _make_response(
+            200,
+            {"MQTT": {"URIS": ["mqtts://b:8883"]}},
+        )
+        mock_create.return_value = mock_api
+
+        with VenClient(url="http://test", token="tok") as ven:
+            assert ven.get_mqtt_broker_uris() == ["mqtts://b:8883"]
+
 
 class TestVenClientMqttTopics:
     @patch("openadr3_client.base.create_ven_client")