diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py
index e4280e6bddc..8716f03838c 100644
--- a/homeassistant/components/knx/__init__.py
+++ b/homeassistant/components/knx/__init__.py
@@ -5,7 +5,6 @@ import logging
 import voluptuous as vol
 from xknx import XKNX
 from xknx.core.telegram_queue import TelegramQueue
-from xknx.devices import DateTime, ExposeSensor
 from xknx.dpt import DPTArray, DPTBase, DPTBinary
 from xknx.exceptions import XKNXException
 from xknx.io import (
@@ -18,26 +17,21 @@ from xknx.telegram import AddressFilter, GroupAddress, Telegram
 from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
 
 from homeassistant.const import (
-    CONF_ENTITY_ID,
     CONF_HOST,
     CONF_PORT,
     EVENT_HOMEASSISTANT_STOP,
     SERVICE_RELOAD,
-    STATE_OFF,
-    STATE_ON,
-    STATE_UNAVAILABLE,
-    STATE_UNKNOWN,
 )
-from homeassistant.core import callback
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import discovery
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import async_get_platforms
-from homeassistant.helpers.event import async_track_state_change_event
 from homeassistant.helpers.reload import async_integration_yaml_config
 from homeassistant.helpers.service import async_register_admin_service
 from homeassistant.helpers.typing import ServiceCallType
 
 from .const import DOMAIN, SupportedPlatforms
+from .expose import create_knx_exposure
 from .factory import create_knx_device
 from .schema import (
     BinarySensorSchema,
@@ -75,6 +69,7 @@ SERVICE_KNX_ATTR_PAYLOAD = "payload"
 SERVICE_KNX_ATTR_TYPE = "type"
 SERVICE_KNX_ATTR_REMOVE = "remove"
 SERVICE_KNX_EVENT_REGISTER = "event_register"
+SERVICE_KNX_EXPOSURE_REGISTER = "exposure_register"
 SERVICE_KNX_READ = "read"
 
 CONFIG_SCHEMA = vol.Schema(
@@ -183,12 +178,27 @@ SERVICE_KNX_EVENT_REGISTER_SCHEMA = vol.Schema(
     }
 )
 
+SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA = vol.Any(
+    ExposeSchema.SCHEMA.extend(
+        {
+            vol.Optional(SERVICE_KNX_ATTR_REMOVE, default=False): cv.boolean,
+        }
+    ),
+    vol.Schema(
+        # for removing only `address` is required
+        {
+            vol.Required(SERVICE_KNX_ATTR_ADDRESS): cv.string,
+            vol.Required(SERVICE_KNX_ATTR_REMOVE): vol.All(cv.boolean, True),
+        },
+        extra=vol.ALLOW_EXTRA,
+    ),
+)
+
 
 async def async_setup(hass, config):
     """Set up the KNX component."""
     try:
         hass.data[DOMAIN] = KNXModule(hass, config)
-        hass.data[DOMAIN].async_create_exposures()
         await hass.data[DOMAIN].start()
     except XKNXException as ex:
         _LOGGER.warning("Could not connect to KNX interface: %s", ex)
@@ -196,6 +206,12 @@ async def async_setup(hass, config):
             f"Could not connect to KNX interface: <br><b>{ex}</b>", title="KNX"
         )
 
+    if CONF_KNX_EXPOSE in config[DOMAIN]:
+        for expose_config in config[DOMAIN][CONF_KNX_EXPOSE]:
+            hass.data[DOMAIN].exposures.append(
+                create_knx_exposure(hass, hass.data[DOMAIN].xknx, expose_config)
+            )
+
     for platform in SupportedPlatforms:
         if platform.value in config[DOMAIN]:
             for device_config in config[DOMAIN][platform.value]:
@@ -235,6 +251,14 @@ async def async_setup(hass, config):
         schema=SERVICE_KNX_EVENT_REGISTER_SCHEMA,
     )
 
+    async_register_admin_service(
+        hass,
+        DOMAIN,
+        SERVICE_KNX_EXPOSURE_REGISTER,
+        hass.data[DOMAIN].service_exposure_register_modify,
+        schema=SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA,
+    )
+
     async def reload_service_handler(service_call: ServiceCallType) -> None:
         """Remove all KNX components and load new ones from config."""
 
@@ -269,6 +293,7 @@ class KNXModule:
         self.config = config
         self.connected = False
         self.exposures = []
+        self.service_exposures = {}
 
         self.init_xknx()
         self._knx_event_callback: TelegramQueue.Callback = self.register_callback()
@@ -340,34 +365,6 @@ class KNXModule:
             auto_reconnect=True,
         )
 
-    @callback
-    def async_create_exposures(self):
-        """Create exposures."""
-        if CONF_KNX_EXPOSE not in self.config[DOMAIN]:
-            return
-        for to_expose in self.config[DOMAIN][CONF_KNX_EXPOSE]:
-            expose_type = to_expose.get(ExposeSchema.CONF_KNX_EXPOSE_TYPE)
-            entity_id = to_expose.get(CONF_ENTITY_ID)
-            attribute = to_expose.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE)
-            default = to_expose.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT)
-            address = to_expose.get(ExposeSchema.CONF_KNX_EXPOSE_ADDRESS)
-            if expose_type.lower() in ["time", "date", "datetime"]:
-                exposure = KNXExposeTime(self.xknx, expose_type, address)
-                exposure.async_register()
-                self.exposures.append(exposure)
-            else:
-                exposure = KNXExposeSensor(
-                    self.hass,
-                    self.xknx,
-                    expose_type,
-                    entity_id,
-                    attribute,
-                    default,
-                    address,
-                )
-                exposure.async_register()
-                self.exposures.append(exposure)
-
     async def telegram_received_cb(self, telegram):
         """Call invoked after a KNX telegram was received."""
         data = None
@@ -417,6 +414,37 @@ class KNXModule:
                 group_address,
             )
 
+    async def service_exposure_register_modify(self, call):
+        """Service for adding or removing an exposure to KNX bus."""
+        group_address = call.data.get(SERVICE_KNX_ATTR_ADDRESS)
+
+        if call.data.get(SERVICE_KNX_ATTR_REMOVE):
+            try:
+                removed_exposure = self.service_exposures.pop(group_address)
+            except KeyError as err:
+                raise HomeAssistantError(
+                    f"Could not find exposure for '{group_address}' to remove."
+                ) from err
+            else:
+                removed_exposure.shutdown()
+            return
+
+        if group_address in self.service_exposures:
+            replaced_exposure = self.service_exposures.pop(group_address)
+            _LOGGER.warning(
+                "Service exposure_register replacing already registered exposure for '%s' - %s",
+                group_address,
+                replaced_exposure.device.name,
+            )
+            replaced_exposure.shutdown()
+        exposure = create_knx_exposure(self.hass, self.xknx, call.data)
+        self.service_exposures[group_address] = exposure
+        _LOGGER.debug(
+            "Service exposure_register registered exposure for '%s' - %s",
+            group_address,
+            exposure.device.name,
+        )
+
     async def service_send_to_knx_bus(self, call):
         """Service for sending an arbitrary KNX message to the KNX bus."""
         attr_payload = call.data.get(SERVICE_KNX_ATTR_PAYLOAD)
@@ -448,93 +476,3 @@ class KNXModule:
                 payload=GroupValueRead(),
             )
             await self.xknx.telegrams.put(telegram)
-
-
-class KNXExposeTime:
-    """Object to Expose Time/Date object to KNX bus."""
-
-    def __init__(self, xknx: XKNX, expose_type: str, address: str):
-        """Initialize of Expose class."""
-        self.xknx = xknx
-        self.expose_type = expose_type
-        self.address = address
-        self.device = None
-
-    @callback
-    def async_register(self):
-        """Register listener."""
-        self.device = DateTime(
-            self.xknx,
-            name=self.expose_type.capitalize(),
-            broadcast_type=self.expose_type.upper(),
-            localtime=True,
-            group_address=self.address,
-        )
-
-
-class KNXExposeSensor:
-    """Object to Expose Home Assistant entity to KNX bus."""
-
-    def __init__(self, hass, xknx, expose_type, entity_id, attribute, default, address):
-        """Initialize of Expose class."""
-        self.hass = hass
-        self.xknx = xknx
-        self.type = expose_type
-        self.entity_id = entity_id
-        self.expose_attribute = attribute
-        self.expose_default = default
-        self.address = address
-        self.device = None
-
-    @callback
-    def async_register(self):
-        """Register listener."""
-        if self.expose_attribute is not None:
-            _name = self.entity_id + "__" + self.expose_attribute
-        else:
-            _name = self.entity_id
-        self.device = ExposeSensor(
-            self.xknx,
-            name=_name,
-            group_address=self.address,
-            value_type=self.type,
-        )
-        async_track_state_change_event(
-            self.hass, [self.entity_id], self._async_entity_changed
-        )
-
-    async def _async_entity_changed(self, event):
-        """Handle entity change."""
-        new_state = event.data.get("new_state")
-        if new_state is None:
-            return
-        if new_state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
-            return
-
-        if self.expose_attribute is not None:
-            new_attribute = new_state.attributes.get(self.expose_attribute)
-            old_state = event.data.get("old_state")
-
-            if old_state is not None:
-                old_attribute = old_state.attributes.get(self.expose_attribute)
-                if old_attribute == new_attribute:
-                    # don't send same value sequentially
-                    return
-            await self._async_set_knx_value(new_attribute)
-        else:
-            await self._async_set_knx_value(new_state.state)
-
-    async def _async_set_knx_value(self, value):
-        """Set new value on xknx ExposeSensor."""
-        if value is None:
-            if self.expose_default is None:
-                return
-            value = self.expose_default
-
-        if self.type == "binary":
-            if value == STATE_ON:
-                value = True
-            elif value == STATE_OFF:
-                value = False
-
-        await self.device.set(value)
diff --git a/homeassistant/components/knx/expose.py b/homeassistant/components/knx/expose.py
new file mode 100644
index 00000000000..93abd7d7b43
--- /dev/null
+++ b/homeassistant/components/knx/expose.py
@@ -0,0 +1,152 @@
+"""Exposures to KNX bus."""
+from typing import Union
+
+from xknx import XKNX
+from xknx.devices import DateTime, ExposeSensor
+
+from homeassistant.const import (
+    CONF_ENTITY_ID,
+    STATE_OFF,
+    STATE_ON,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.event import async_track_state_change_event
+from homeassistant.helpers.typing import ConfigType
+
+from .schema import ExposeSchema
+
+
+@callback
+def create_knx_exposure(
+    hass: HomeAssistant, xknx: XKNX, config: ConfigType
+) -> Union["KNXExposeSensor", "KNXExposeTime"]:
+    """Create exposures from config."""
+    expose_type = config.get(ExposeSchema.CONF_KNX_EXPOSE_TYPE)
+    entity_id = config.get(CONF_ENTITY_ID)
+    attribute = config.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE)
+    default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT)
+    address = config.get(ExposeSchema.CONF_KNX_EXPOSE_ADDRESS)
+
+    exposure: Union["KNXExposeSensor", "KNXExposeTime"]
+    if expose_type.lower() in ["time", "date", "datetime"]:
+        exposure = KNXExposeTime(xknx, expose_type, address)
+    else:
+        exposure = KNXExposeSensor(
+            hass,
+            xknx,
+            expose_type,
+            entity_id,
+            attribute,
+            default,
+            address,
+        )
+    exposure.async_register()
+    return exposure
+
+
+class KNXExposeSensor:
+    """Object to Expose Home Assistant entity to KNX bus."""
+
+    def __init__(self, hass, xknx, expose_type, entity_id, attribute, default, address):
+        """Initialize of Expose class."""
+        self.hass = hass
+        self.xknx = xknx
+        self.type = expose_type
+        self.entity_id = entity_id
+        self.expose_attribute = attribute
+        self.expose_default = default
+        self.address = address
+        self.device = None
+        self._remove_listener = None
+
+    @callback
+    def async_register(self):
+        """Register listener."""
+        if self.expose_attribute is not None:
+            _name = self.entity_id + "__" + self.expose_attribute
+        else:
+            _name = self.entity_id
+        self.device = ExposeSensor(
+            self.xknx,
+            name=_name,
+            group_address=self.address,
+            value_type=self.type,
+        )
+        self._remove_listener = async_track_state_change_event(
+            self.hass, [self.entity_id], self._async_entity_changed
+        )
+
+    @callback
+    def shutdown(self) -> None:
+        """Prepare for deletion."""
+        if self._remove_listener is not None:
+            self._remove_listener()
+        if self.device is not None:
+            self.device.shutdown()
+
+    async def _async_entity_changed(self, event):
+        """Handle entity change."""
+        new_state = event.data.get("new_state")
+        if new_state is None:
+            return
+        if new_state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
+            return
+
+        if self.expose_attribute is None:
+            await self._async_set_knx_value(new_state.state)
+            return
+
+        new_attribute = new_state.attributes.get(self.expose_attribute)
+        old_state = event.data.get("old_state")
+
+        if old_state is not None:
+            old_attribute = old_state.attributes.get(self.expose_attribute)
+            if old_attribute == new_attribute:
+                # don't send same value sequentially
+                return
+        await self._async_set_knx_value(new_attribute)
+
+    async def _async_set_knx_value(self, value):
+        """Set new value on xknx ExposeSensor."""
+        if value is None:
+            if self.expose_default is None:
+                return
+            value = self.expose_default
+
+        if self.type == "binary":
+            if value == STATE_ON:
+                value = True
+            elif value == STATE_OFF:
+                value = False
+
+        await self.device.set(value)
+
+
+class KNXExposeTime:
+    """Object to Expose Time/Date object to KNX bus."""
+
+    def __init__(self, xknx: XKNX, expose_type: str, address: str):
+        """Initialize of Expose class."""
+        self.xknx = xknx
+        self.expose_type = expose_type
+        self.address = address
+        self.device = None
+
+    @callback
+    def async_register(self):
+        """Register listener."""
+        self.device = DateTime(
+            self.xknx,
+            name=self.expose_type.capitalize(),
+            broadcast_type=self.expose_type.upper(),
+            localtime=True,
+            group_address=self.address,
+        )
+
+    @callback
+    def shutdown(self):
+        """Prepare for deletion."""
+        if self.device is not None:
+            self.device.shutdown()
diff --git a/homeassistant/components/knx/services.yaml b/homeassistant/components/knx/services.yaml
index aa946459cfd..ef74acc49b1 100644
--- a/homeassistant/components/knx/services.yaml
+++ b/homeassistant/components/knx/services.yaml
@@ -2,25 +2,89 @@ send:
   description: "Send arbitrary data directly to the KNX bus."
   fields:
     address:
+      name: "Group address"
       description: "Group address(es) to write to."
+      required: true
       example: "1/1/0"
+      selector:
+        text:
     payload:
+      name: "Payload"
       description: "Payload to send to the bus. Integers are treated as DPT 1/2/3 payloads. For DPTs > 6 bits send a list. Each value represents 1 octet (0-255). Pad with 0 to DPT byte length."
+      required: true
       example: "[0, 4]"
     type:
+      name: "Value type"
       description: "Optional. If set, the payload will not be sent as raw bytes, but encoded as given DPT. Knx sensor types are valid values (see https://www.home-assistant.io/integrations/sensor.knx)."
+      required: false
       example: "temperature"
+      selector:
+        text:
 read:
   description: "Send GroupValueRead requests to the KNX bus. Response can be used from `knx_event` and will be processed in KNX entities."
   fields:
     address:
+      name: "Group address"
       description: "Group address(es) to send read request to. Lists will read multiple group addresses."
+      required: true
       example: "1/1/0"
+      selector:
+        text:
 event_register:
   description: "Add or remove single group address to knx_event filter for triggering `knx_event`s. Only addresses added with this service can be removed."
   fields:
     address:
+      name: "Group address"
       description: "Group address that shall be added or removed."
+      required: true
       example: "1/1/0"
     remove:
-      description: "Optional. If `True` the group address will be removed. Defaults to `False`."
+      name: "Remove event registration"
+      description: "Optional. If `True` the group address will be removed."
+      required: false
+      default: false
+      selector:
+        boolean:
+exposure_register:
+  description: "Add or remove exposures to KNX bus. Only exposures added with this service can be removed."
+  fields:
+    address:
+      name: "Group address"
+      description: "Group address state or attribute updates will be sent to. GroupValueRead requests will be answered. Per address only one exposure can be registered."
+      required: true
+      example: "1/1/0"
+      selector:
+        text:
+    type:
+      name: "Value type"
+      description: "Telegrams will be encoded as given DPT. 'binary' and all Knx sensor types are valid values (see https://www.home-assistant.io/integrations/sensor.knx)"
+      required: true
+      example: "percentU8"
+      selector:
+        text:
+    entity_id:
+      name: "Entity"
+      description: "Entity id whose state or attribute shall be exposed."
+      required: true
+      example: "light.kitchen"
+      selector:
+        entity:
+    attribute:
+      name: "Entity attribute"
+      description: "Optional. Attribute of the entity that shall be sent to the KNX bus. If not set the state will be sent. Eg. for a light the state is eigther “on” or “off” - with attribute you can expose its “brightness”."
+      required: false
+      example: "brightness"
+    default:
+      name: "Default value"
+      description: "Optional. Default value to send to the bus if the state or attribute value is None. Eg. a light with state “off” has no brightness attribute so a default value of 0 could be used. If not set (or None) no value would be sent to the bus and a GroupReadRequest to the address would return the last known value."
+      required: false
+      example: "0"
+    remove:
+      name: "Remove exposure"
+      description: "Optional. If `True` the exposure will be removed. Only `address` is required for removal."
+      required: false
+      default: false
+      selector:
+        boolean:
+reload:
+  description: "Reload KNX configuration."