KNX MQTT Integration: Bridge Group Addresses to MQTT Topics

A KNX MQTT bridge turns telegrams on a KNX bus into MQTT messages and turns MQTT commands back into KNX group writes. In this tutorial you will build that bridge in Python with the open-source xknx library, map group addresses to readable topics, decode datapoint types into JSON, and connect it to a ConnectMQ broker so dashboards and other buildings can use your KNX data.

Why Bridge KNX to MQTT

KNX is excellent at room-level automation. Push buttons, dimmers, blind actuators and presence detectors talk directly to each other through group addresses, so the lights keep working even if every server in the building is off. That decentralized design is also why KNX data rarely leaves the building: the bus speaks group telegrams with compact binary datapoint types (DPTs), and KNX/IP routing uses multicast that does not cross the internet.

An MQTT bridge adds a northbound interface without changing how the installation works. Each group address of interest becomes a topic, values become JSON, and any client connected to the MQTT broker can read them. This is the same integration layer described in our guide to building automation system MQTT.

What You Need

• A KNX/IP interface or router reachable on the LAN. Tunneling to an interface is the most common setup.
• The group address list from the ETS project, with the DPT of each address. Ask the KNX integrator for an export if you do not have ETS.
• A Linux box, Raspberry Pi or PC on the same network.
• Python 3.10 or newer.

python -m venv knx-gw
source knx-gw/bin/activate
pip install xknx "paho-mqtt>=2.0"

Most KNX/IP interfaces support only a limited number of simultaneous tunnels, often between one and a handful. If ETS, a visualization and your bridge all connect at once, one of them may be refused. Check how many tunnels your interface offers before you start.

Step 1: Map Group Addresses

KNX group addresses look like 1/2/3. Using them directly as MQTT topic levels creates deep hierarchies and confusing topic names, so map each one to a meaningful name and a DPT:

Write the same mapping into the bridge as a Python dictionary. Keeping status and command addresses separate is standard KNX practice and fits MQTT nicely: state topics are published by the bridge, command topics are subscribed.

Step 2: The KNX MQTT Bridge

The bridge connects to KNX through tunneling, listens to all telegrams, decodes the ones in the mapping, and publishes them. It also subscribes to .../set topics and converts them into group writes.

import asyncio
import json
import uuid
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from xknx import XKNX
from xknx.dpt import DPTArray, DPTBinary, DPTScaling, DPTTemperature
from xknx.io import ConnectionConfig, ConnectionType
from xknx.telegram import GroupAddress, Telegram
from xknx.telegram.apci import GroupValueResponse, GroupValueWrite

KNX_GATEWAY_IP = "192.168.20.10"
BROKER = "broker.connectmq.com"
PORT = 1883                         # 8883 + TLS for a private broker
USERNAME = None                     # "YOUR_USERNAME"
PASSWORD = None                     # "YOUR_PASSWORD"
CLIENT_ID = f"knx-gw-{uuid.uuid4().hex[:8]}"   # "YOUR_CLIENT_ID"
PREFIX = "connectmq/demo/knx-4mq8"

# group address -> (topic suffix, dpt kind)
STATE_MAP = {
    "1/1/2": ("f2/office-201/light/state", "switch"),
    "1/1/5": ("f2/office-201/light/brightness-state", "percent"),
    "3/1/1": ("f2/office-201/temperature", "temperature"),
    "4/1/1": ("f2/office-201/presence", "switch"),
}
# topic suffix -> (group address, dpt kind)
COMMAND_MAP = {
    "f2/office-201/light/switch": ("1/1/1", "switch"),
    "f2/office-201/light/brightness": ("1/1/4", "percent"),
}

loop: asyncio.AbstractEventLoop | None = None
xknx: XKNX | None = None
mqttc: mqtt.Client | None = None


def decode(kind, value):
    if kind == "switch" and isinstance(value, DPTBinary):
        return bool(value.value)
    if kind == "percent" and isinstance(value, DPTArray):
        return DPTScaling.from_knx(value)
    if kind == "temperature" and isinstance(value, DPTArray):
        return round(DPTTemperature.from_knx(value), 2)
    return None


def encode(kind, value):
    if kind == "switch":
        return DPTBinary(1 if value in (True, 1, "ON", "on", "true") else 0)
    if kind == "percent":
        return DPTScaling.to_knx(max(0, min(100, float(value))))
    raise ValueError(kind)


async def telegram_received(telegram: Telegram):
    ga = str(telegram.destination_address)
    if ga not in STATE_MAP:
        return
    if not isinstance(telegram.payload, (GroupValueWrite, GroupValueResponse)):
        return
    suffix, kind = STATE_MAP[ga]
    value = decode(kind, telegram.payload.value)
    if value is None:
        return
    payload = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
               "value": value, "ga": ga, "source": str(telegram.source_address)}
    # retained so dashboards see the last state immediately
    mqttc.publish(f"{PREFIX}/{suffix}", json.dumps(payload), qos=1, retain=True)


async def send_knx(ga, kind, value):
    telegram = Telegram(destination_address=GroupAddress(ga),
                        payload=GroupValueWrite(encode(kind, value)))
    await xknx.telegrams.put(telegram)


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/set", qos=1)
    client.subscribe(f"{PREFIX}/+/+/+/set", qos=1)
    client.publish(f"{PREFIX}/gateway/status", "online", qos=1, retain=True)


def on_message(client, userdata, msg):
    suffix = msg.topic[len(PREFIX) + 1:].removesuffix("/set")
    if suffix not in COMMAND_MAP:
        print("Ignored command for", suffix)
        return
    ga, kind = COMMAND_MAP[suffix]
    try:
        value = json.loads(msg.payload)
    except json.JSONDecodeError:
        value = msg.payload.decode()
    if isinstance(value, dict):
        value = value.get("value")
    # paho callbacks run in another thread; schedule the KNX write on the asyncio loop
    asyncio.run_coroutine_threadsafe(send_knx(ga, kind, value), loop)


async def main():
    global loop, xknx, mqttc
    loop = asyncio.get_running_loop()

    mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mqttc.username_pw_set(USERNAME, PASSWORD)
        mqttc.tls_set()
    mqttc.will_set(f"{PREFIX}/gateway/status", "offline", qos=1, retain=True)
    mqttc.on_connect = on_connect
    mqttc.on_message = on_message
    mqttc.connect(BROKER, PORT, keepalive=60)
    mqttc.loop_start()

    xknx = XKNX(
        connection_config=ConnectionConfig(
            connection_type=ConnectionType.TUNNELING,
            gateway_ip=KNX_GATEWAY_IP,
        ),
        telegram_received_cb=telegram_received,
    )
    await xknx.start()
    print("KNX connected")
    try:
        await asyncio.Event().wait()   # run forever
    finally:
        await xknx.stop()
        mqttc.loop_stop()

asyncio.run(main())

A few details matter here. The two subscriptions cover command topics at both four and three levels below the prefix, which matches the names in COMMAND_MAP. The bridge also ignores any group address it does not know, so a mistake in a topic can never write to an arbitrary address. Finally, run_coroutine_threadsafe is required because paho delivers messages on its network thread, while xknx runs inside asyncio.

Reading Initial States at Startup

KNX devices only send a telegram when something changes, so a freshly started bridge knows nothing about the current state of the building. Retained MQTT messages hide this for dashboards, but after a long outage the retained values may be stale. The fix is to send a GroupValueRead to every status address once the KNX connection is up. Actuators with the read flag set in ETS answer with a GroupValueResponse, which the bridge already handles because telegram_received accepts both writes and responses.

from xknx.telegram.apci import GroupValueRead

async def read_all_states():
    for ga in STATE_MAP:
        await xknx.telegrams.put(Telegram(destination_address=GroupAddress(ga),
                                          payload=GroupValueRead()))
        await asyncio.sleep(0.2)   # be gentle with the bus

Call await read_all_states() right after await xknx.start(). The short pause between reads matters: a KNX twisted-pair line has limited bandwidth, and firing dozens of reads at once can delay real telegrams from push buttons. If a status address never answers, the read flag is probably not set on the actuator's communication object; ask the integrator to enable it in ETS rather than working around it in the bridge.

Step 3: Test the Bridge

Watch the published states:

mosquitto_sub -h broker.connectmq.com -p 1883 -i knx-view-4mq8 -t "connectmq/demo/knx-4mq8/#" -v

Press the wall switch in office 201. You should see:

connectmq/demo/knx-4mq8/f2/office-201/light/state {"ts": "2026-10-16T09:02:11+00:00", "value": true, "ga": "1/1/2", "source": "1.1.12"}

Now switch the light from MQTT and dim it to 40 percent:

mosquitto_pub -h broker.connectmq.com -p 1883 -i knx-cmd-4mq8 -q 1 \
  -t "connectmq/demo/knx-4mq8/f2/office-201/light/switch/set" -m '{"value": "ON"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i knx-cmd-4mq8 -q 1 \
  -t "connectmq/demo/knx-4mq8/f2/office-201/light/brightness/set" -m '{"value": 40}'

The public MQTT test broker is fine for this first round, but anyone can read and publish under any public topic. Use a lab KNX kit, not a real office, while testing in public.

Step 4: Run It Against a Private Broker

For a live installation, create a broker in the ConnectMQ portal and use its credentials:

BROKER = "broker.connectmq.com"   # or the host shown for your broker
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"
PREFIX = "acme/ist01/hq"

Then install the bridge as a systemd service so it restarts after reboots:

sudo tee /etc/systemd/system/knx-mqtt.service > /dev/null <<'EOF'
[Unit]
Description=KNX MQTT bridge
After=network-online.target
Wants=network-online.target

[Service]
WorkingDirectory=/opt/knx-gw
ExecStart=/opt/knx-gw/bin/python /opt/knx-gw/bridge.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload
sudo systemctl enable --now knx-mqtt

With a managed MQTT broker the only process you maintain on site is the bridge itself.

Keeping the Bridge Secure

The bridge holds a connection into the KNX installation, so treat it like any other building controller. Run it on a dedicated device or VLAN, keep its operating system updated, and store the MQTT password in an environment file readable only by the service user rather than in the script itself. If your KNX installation uses KNX Secure, xknx can use secure tunneling with the keyring exported from ETS; ask the integrator for the credentials instead of disabling security on the interface.

Troubleshooting

1. Tunnel connection refused. All tunnels on the interface are in use. Close ETS or another visualization, or use an interface with more tunnels.
2. No telegrams arrive. The interface may filter group telegrams. Check its filter table in ETS, or connect to a KNX/IP router instead.
3. Values decode to nonsense. The DPT in your mapping is wrong. A 2-byte float (DPT 9) decoded as a scaling value (DPT 5) produces meaningless numbers.
4. States missing after a bridge restart. KNX only sends telegrams on change. Send a GroupValueRead for each state address at startup, or rely on the retained MQTT messages.
5. Duplicate commands. QoS 1 can deliver a message twice; switching a light on twice is harmless, but avoid toggle semantics and always send absolute values.

Alternatives Worth Knowing

If you prefer a visual tool, Node-RED with a KNX node package can perform the same mapping, and Home Assistant has a native KNX integration that can publish to MQTT through automations. The Python bridge above is the most transparent option and is easy to version-control. The official xknx documentation at xknx.github.io lists every supported DPT.

For lighting scenes on top of this bridge, see smart lighting with MQTT, and if the same building runs BACnet HVAC, combine it with the BACnet to MQTT gateway. Our IoT automation guide collects related projects.

Connect Your KNX Installation to ConnectMQ

Create a free Serverless broker at app.connectmq.com, no credit card required, and you will have credentials for the bridge in a few minutes. The live topic explorer in ConnectMQ shows every group address as it changes, which makes commissioning and fault-finding much faster.

Frequently Asked Questions

Do I need a KNX/IP router or is an interface enough?

An IP interface with tunneling is enough for a single bridge. A router helps when several applications need simultaneous access, because routing uses multicast rather than a limited number of tunnels.

Will the KNX installation still work if MQTT goes down?

Yes. KNX devices communicate directly over group addresses, so the bridge only adds visibility and remote control; local switching never depends on it.

Should state topics be retained?

Yes. KNX sends values only on change, so retained MQTT messages let a dashboard show the current state immediately after it connects.

Can I send any group address from MQTT?

You should not. Keep an explicit command map in the bridge so only approved addresses can be written from MQTT.