diff --git a/homeassistant/components/manual_mqtt/alarm_control_panel.py b/homeassistant/components/manual_mqtt/alarm_control_panel.py
index a94b1013782..4eb30028e8b 100644
--- a/homeassistant/components/manual_mqtt/alarm_control_panel.py
+++ b/homeassistant/components/manual_mqtt/alarm_control_panel.py
@@ -452,6 +452,6 @@ class ManualMQTTAlarm(alarm.AlarmControlPanelEntity):
         """Publish state change to MQTT."""
         if (new_state := event.data.get("new_state")) is None:
             return
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass, self._state_topic, new_state.state, self._qos, True
         )
diff --git a/homeassistant/components/mqtt/__init__.py b/homeassistant/components/mqtt/__init__.py
index 994ae2d108a..f42663ac4a8 100644
--- a/homeassistant/components/mqtt/__init__.py
+++ b/homeassistant/components/mqtt/__init__.py
@@ -245,39 +245,16 @@ def _build_publish_data(topic: Any, qos: int, retain: bool) -> ServiceDataType:
     return data
 
 
-@bind_hass
-def publish(hass: HomeAssistant, topic, payload, qos=None, retain=None) -> None:
+def publish(hass: HomeAssistant, topic, payload, qos=0, retain=False) -> None:
     """Publish message to an MQTT topic."""
     hass.add_job(async_publish, hass, topic, payload, qos, retain)
 
 
-@callback
-@bind_hass
-def async_publish(
-    hass: HomeAssistant, topic: Any, payload, qos=None, retain=None
+async def async_publish(
+    hass: HomeAssistant, topic: Any, payload, qos=0, retain=False
 ) -> None:
     """Publish message to an MQTT topic."""
-    data = _build_publish_data(topic, qos, retain)
-    data[ATTR_PAYLOAD] = payload
-    hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_PUBLISH, data))
-
-
-@bind_hass
-def publish_template(
-    hass: HomeAssistant, topic, payload_template, qos=None, retain=None
-) -> None:
-    """Publish message to an MQTT topic."""
-    hass.add_job(async_publish_template, hass, topic, payload_template, qos, retain)
-
-
-@bind_hass
-def async_publish_template(
-    hass: HomeAssistant, topic, payload_template, qos=None, retain=None
-) -> None:
-    """Publish message to an MQTT topic using a template payload."""
-    data = _build_publish_data(topic, qos, retain)
-    data[ATTR_PAYLOAD_TEMPLATE] = payload_template
-    hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_PUBLISH, data))
+    await hass.data[DATA_MQTT].async_publish(topic, str(payload), qos, retain)
 
 
 AsyncDeprecatedMessageCallbackType = Callable[
diff --git a/homeassistant/components/mqtt/alarm_control_panel.py b/homeassistant/components/mqtt/alarm_control_panel.py
index dfdf2dbfb26..5ac6e5ea89a 100644
--- a/homeassistant/components/mqtt/alarm_control_panel.py
+++ b/homeassistant/components/mqtt/alarm_control_panel.py
@@ -228,7 +228,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
         if code_required and not self._validate_code(code, "disarming"):
             return
         payload = self._config[CONF_PAYLOAD_DISARM]
-        self._publish(code, payload)
+        await self._publish(code, payload)
 
     async def async_alarm_arm_home(self, code=None):
         """Send arm home command.
@@ -239,7 +239,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
         if code_required and not self._validate_code(code, "arming home"):
             return
         action = self._config[CONF_PAYLOAD_ARM_HOME]
-        self._publish(code, action)
+        await self._publish(code, action)
 
     async def async_alarm_arm_away(self, code=None):
         """Send arm away command.
@@ -250,7 +250,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
         if code_required and not self._validate_code(code, "arming away"):
             return
         action = self._config[CONF_PAYLOAD_ARM_AWAY]
-        self._publish(code, action)
+        await self._publish(code, action)
 
     async def async_alarm_arm_night(self, code=None):
         """Send arm night command.
@@ -261,7 +261,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
         if code_required and not self._validate_code(code, "arming night"):
             return
         action = self._config[CONF_PAYLOAD_ARM_NIGHT]
-        self._publish(code, action)
+        await self._publish(code, action)
 
     async def async_alarm_arm_vacation(self, code=None):
         """Send arm vacation command.
@@ -272,7 +272,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
         if code_required and not self._validate_code(code, "arming vacation"):
             return
         action = self._config[CONF_PAYLOAD_ARM_VACATION]
-        self._publish(code, action)
+        await self._publish(code, action)
 
     async def async_alarm_arm_custom_bypass(self, code=None):
         """Send arm custom bypass command.
@@ -283,14 +283,14 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
         if code_required and not self._validate_code(code, "arming custom bypass"):
             return
         action = self._config[CONF_PAYLOAD_ARM_CUSTOM_BYPASS]
-        self._publish(code, action)
+        await self._publish(code, action)
 
-    def _publish(self, code, action):
+    async def _publish(self, code, action):
         """Publish via mqtt."""
         command_template = self._config[CONF_COMMAND_TEMPLATE]
         values = {"action": action, "code": code}
         payload = command_template.async_render(**values, parse_result=False)
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             payload,
diff --git a/homeassistant/components/mqtt/climate.py b/homeassistant/components/mqtt/climate.py
index 16d4ae695c1..5b186ff9126 100644
--- a/homeassistant/components/mqtt/climate.py
+++ b/homeassistant/components/mqtt/climate.py
@@ -659,9 +659,9 @@ class MqttClimate(MqttEntity, ClimateEntity):
         """Return the list of available fan modes."""
         return self._config[CONF_FAN_MODE_LIST]
 
-    def _publish(self, topic, payload):
+    async def _publish(self, topic, payload):
         if self._topic[topic] is not None:
-            mqtt.async_publish(
+            await mqtt.async_publish(
                 self.hass,
                 self._topic[topic],
                 payload,
@@ -669,7 +669,9 @@ class MqttClimate(MqttEntity, ClimateEntity):
                 self._config[CONF_RETAIN],
             )
 
-    def _set_temperature(self, temp, cmnd_topic, cmnd_template, state_topic, attr):
+    async def _set_temperature(
+        self, temp, cmnd_topic, cmnd_template, state_topic, attr
+    ):
         if temp is not None:
             if self._topic[state_topic] is None:
                 # optimistic mode
@@ -680,7 +682,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
                 or self._current_operation != HVAC_MODE_OFF
             ):
                 payload = self._command_templates[cmnd_template](temp)
-                self._publish(cmnd_topic, payload)
+                await self._publish(cmnd_topic, payload)
 
     async def async_set_temperature(self, **kwargs):
         """Set new target temperatures."""
@@ -688,7 +690,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
             operation_mode = kwargs.get(ATTR_HVAC_MODE)
             await self.async_set_hvac_mode(operation_mode)
 
-        self._set_temperature(
+        await self._set_temperature(
             kwargs.get(ATTR_TEMPERATURE),
             CONF_TEMP_COMMAND_TOPIC,
             CONF_TEMP_COMMAND_TEMPLATE,
@@ -696,7 +698,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
             "_target_temp",
         )
 
-        self._set_temperature(
+        await self._set_temperature(
             kwargs.get(ATTR_TARGET_TEMP_LOW),
             CONF_TEMP_LOW_COMMAND_TOPIC,
             CONF_TEMP_LOW_COMMAND_TEMPLATE,
@@ -704,7 +706,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
             "_target_temp_low",
         )
 
-        self._set_temperature(
+        await self._set_temperature(
             kwargs.get(ATTR_TARGET_TEMP_HIGH),
             CONF_TEMP_HIGH_COMMAND_TOPIC,
             CONF_TEMP_HIGH_COMMAND_TEMPLATE,
@@ -721,7 +723,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
             payload = self._command_templates[CONF_SWING_MODE_COMMAND_TEMPLATE](
                 swing_mode
             )
-            self._publish(CONF_SWING_MODE_COMMAND_TOPIC, payload)
+            await self._publish(CONF_SWING_MODE_COMMAND_TOPIC, payload)
 
         if self._topic[CONF_SWING_MODE_STATE_TOPIC] is None:
             self._current_swing_mode = swing_mode
@@ -731,7 +733,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
         """Set new target temperature."""
         if self._config[CONF_SEND_IF_OFF] or self._current_operation != HVAC_MODE_OFF:
             payload = self._command_templates[CONF_FAN_MODE_COMMAND_TEMPLATE](fan_mode)
-            self._publish(CONF_FAN_MODE_COMMAND_TOPIC, payload)
+            await self._publish(CONF_FAN_MODE_COMMAND_TOPIC, payload)
 
         if self._topic[CONF_FAN_MODE_STATE_TOPIC] is None:
             self._current_fan_mode = fan_mode
@@ -740,12 +742,14 @@ class MqttClimate(MqttEntity, ClimateEntity):
     async def async_set_hvac_mode(self, hvac_mode) -> None:
         """Set new operation mode."""
         if self._current_operation == HVAC_MODE_OFF and hvac_mode != HVAC_MODE_OFF:
-            self._publish(CONF_POWER_COMMAND_TOPIC, self._config[CONF_PAYLOAD_ON])
+            await self._publish(CONF_POWER_COMMAND_TOPIC, self._config[CONF_PAYLOAD_ON])
         elif self._current_operation != HVAC_MODE_OFF and hvac_mode == HVAC_MODE_OFF:
-            self._publish(CONF_POWER_COMMAND_TOPIC, self._config[CONF_PAYLOAD_OFF])
+            await self._publish(
+                CONF_POWER_COMMAND_TOPIC, self._config[CONF_PAYLOAD_OFF]
+            )
 
         payload = self._command_templates[CONF_MODE_COMMAND_TEMPLATE](hvac_mode)
-        self._publish(CONF_MODE_COMMAND_TOPIC, payload)
+        await self._publish(CONF_MODE_COMMAND_TOPIC, payload)
 
         if self._topic[CONF_MODE_STATE_TOPIC] is None:
             self._current_operation = hvac_mode
@@ -770,26 +774,28 @@ class MqttClimate(MqttEntity, ClimateEntity):
         optimistic_update = False
 
         if self._away:
-            optimistic_update = optimistic_update or self._set_away_mode(False)
+            optimistic_update = optimistic_update or await self._set_away_mode(False)
         elif preset_mode == PRESET_AWAY:
             if self._hold:
-                self._set_hold_mode(None)
-            optimistic_update = optimistic_update or self._set_away_mode(True)
+                await self._set_hold_mode(None)
+            optimistic_update = optimistic_update or await self._set_away_mode(True)
         else:
             hold_mode = preset_mode
             if preset_mode == PRESET_NONE:
                 hold_mode = None
-            optimistic_update = optimistic_update or self._set_hold_mode(hold_mode)
+            optimistic_update = optimistic_update or await self._set_hold_mode(
+                hold_mode
+            )
 
         if optimistic_update:
             self.async_write_ha_state()
 
-    def _set_away_mode(self, state):
+    async def _set_away_mode(self, state):
         """Set away mode.
 
         Returns if we should optimistically write the state.
         """
-        self._publish(
+        await self._publish(
             CONF_AWAY_MODE_COMMAND_TOPIC,
             self._config[CONF_PAYLOAD_ON] if state else self._config[CONF_PAYLOAD_OFF],
         )
@@ -800,7 +806,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
         self._away = state
         return True
 
-    def _set_hold_mode(self, hold_mode):
+    async def _set_hold_mode(self, hold_mode):
         """Set hold mode.
 
         Returns if we should optimistically write the state.
@@ -808,7 +814,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
         payload = self._command_templates[CONF_HOLD_COMMAND_TEMPLATE](
             hold_mode or "off"
         )
-        self._publish(CONF_HOLD_COMMAND_TOPIC, payload)
+        await self._publish(CONF_HOLD_COMMAND_TOPIC, payload)
 
         if self._topic[CONF_HOLD_STATE_TOPIC] is not None:
             return False
@@ -816,8 +822,8 @@ class MqttClimate(MqttEntity, ClimateEntity):
         self._hold = hold_mode
         return True
 
-    def _set_aux_heat(self, state):
-        self._publish(
+    async def _set_aux_heat(self, state):
+        await self._publish(
             CONF_AUX_COMMAND_TOPIC,
             self._config[CONF_PAYLOAD_ON] if state else self._config[CONF_PAYLOAD_OFF],
         )
@@ -828,11 +834,11 @@ class MqttClimate(MqttEntity, ClimateEntity):
 
     async def async_turn_aux_heat_on(self):
         """Turn auxiliary heater on."""
-        self._set_aux_heat(True)
+        await self._set_aux_heat(True)
 
     async def async_turn_aux_heat_off(self):
         """Turn auxiliary heater off."""
-        self._set_aux_heat(False)
+        await self._set_aux_heat(False)
 
     @property
     def supported_features(self):
diff --git a/homeassistant/components/mqtt/cover.py b/homeassistant/components/mqtt/cover.py
index 0af9d7d3739..859d9811617 100644
--- a/homeassistant/components/mqtt/cover.py
+++ b/homeassistant/components/mqtt/cover.py
@@ -528,7 +528,7 @@ class MqttCover(MqttEntity, CoverEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_COMMAND_TOPIC),
             self._config[CONF_PAYLOAD_OPEN],
@@ -549,7 +549,7 @@ class MqttCover(MqttEntity, CoverEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_COMMAND_TOPIC),
             self._config[CONF_PAYLOAD_CLOSE],
@@ -570,7 +570,7 @@ class MqttCover(MqttEntity, CoverEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_COMMAND_TOPIC),
             self._config[CONF_PAYLOAD_STOP],
@@ -580,7 +580,7 @@ class MqttCover(MqttEntity, CoverEntity):
 
     async def async_open_cover_tilt(self, **kwargs):
         """Tilt the cover open."""
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_TILT_COMMAND_TOPIC),
             self._config[CONF_TILT_OPEN_POSITION],
@@ -595,7 +595,7 @@ class MqttCover(MqttEntity, CoverEntity):
 
     async def async_close_cover_tilt(self, **kwargs):
         """Tilt the cover closed."""
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_TILT_COMMAND_TOPIC),
             self._config[CONF_TILT_CLOSED_POSITION],
@@ -626,7 +626,7 @@ class MqttCover(MqttEntity, CoverEntity):
             }
             tilt = template.async_render(parse_result=False, variables=variables)
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_TILT_COMMAND_TOPIC),
             tilt,
@@ -655,7 +655,7 @@ class MqttCover(MqttEntity, CoverEntity):
             }
             position = template.async_render(parse_result=False, variables=variables)
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config.get(CONF_SET_POSITION_TOPIC),
             position,
diff --git a/homeassistant/components/mqtt/fan.py b/homeassistant/components/mqtt/fan.py
index f950a4d2c60..34e5d66a5e7 100644
--- a/homeassistant/components/mqtt/fan.py
+++ b/homeassistant/components/mqtt/fan.py
@@ -520,7 +520,7 @@ class MqttFan(MqttEntity, FanEntity):
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_ON"])
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
@@ -541,7 +541,7 @@ class MqttFan(MqttEntity, FanEntity):
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_OFF"])
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
@@ -561,7 +561,7 @@ class MqttFan(MqttEntity, FanEntity):
             percentage_to_ranged_value(self._speed_range, percentage)
         )
         mqtt_payload = self._command_templates[ATTR_PERCENTAGE](percentage_payload)
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_PERCENTAGE_COMMAND_TOPIC],
             mqtt_payload,
@@ -584,7 +584,7 @@ class MqttFan(MqttEntity, FanEntity):
 
         mqtt_payload = self._command_templates[ATTR_PRESET_MODE](preset_mode)
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_PRESET_MODE_COMMAND_TOPIC],
             mqtt_payload,
@@ -610,7 +610,7 @@ class MqttFan(MqttEntity, FanEntity):
                 self._payload["OSCILLATE_OFF_PAYLOAD"]
             )
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_OSCILLATION_COMMAND_TOPIC],
             mqtt_payload,
diff --git a/homeassistant/components/mqtt/humidifier.py b/homeassistant/components/mqtt/humidifier.py
index e8bbbc7fd4b..e4f578ef94c 100644
--- a/homeassistant/components/mqtt/humidifier.py
+++ b/homeassistant/components/mqtt/humidifier.py
@@ -385,7 +385,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_ON"])
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
@@ -402,7 +402,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_OFF"])
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
@@ -419,7 +419,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[ATTR_HUMIDITY](humidity)
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_TARGET_HUMIDITY_COMMAND_TOPIC],
             mqtt_payload,
@@ -442,7 +442,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
 
         mqtt_payload = self._command_templates[ATTR_MODE](mode)
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_MODE_COMMAND_TOPIC],
             mqtt_payload,
diff --git a/homeassistant/components/mqtt/light/schema_basic.py b/homeassistant/components/mqtt/light/schema_basic.py
index 5dd3f13af25..1909d7e136b 100644
--- a/homeassistant/components/mqtt/light/schema_basic.py
+++ b/homeassistant/components/mqtt/light/schema_basic.py
@@ -811,9 +811,9 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
         should_update = False
         on_command_type = self._config[CONF_ON_COMMAND_TYPE]
 
-        def publish(topic, payload):
+        async def publish(topic, payload):
             """Publish an MQTT message."""
-            mqtt.async_publish(
+            await mqtt.async_publish(
                 self.hass,
                 self._topic[topic],
                 payload,
@@ -859,7 +859,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             return True
 
         if on_command_type == "first":
-            publish(CONF_COMMAND_TOPIC, self._payload["on"])
+            await publish(CONF_COMMAND_TOPIC, self._payload["on"])
             should_update = True
 
         # If brightness is being used instead of an on command, make sure
@@ -881,13 +881,13 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             # Legacy mode: Convert HS to RGB
             rgb = scale_rgbx(color_util.color_hsv_to_RGB(*hs_color, 100))
             rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
-            publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
+            await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
             should_update |= set_optimistic(
                 ATTR_HS_COLOR, hs_color, condition_attribute=ATTR_RGB_COLOR
             )
 
         if hs_color and self._topic[CONF_HS_COMMAND_TOPIC] is not None:
-            publish(CONF_HS_COMMAND_TOPIC, f"{hs_color[0]},{hs_color[1]}")
+            await publish(CONF_HS_COMMAND_TOPIC, f"{hs_color[0]},{hs_color[1]}")
             should_update |= set_optimistic(ATTR_HS_COLOR, hs_color, COLOR_MODE_HS)
 
         if (
@@ -897,7 +897,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
         ):
             # Legacy mode: Convert HS to XY
             xy_color = color_util.color_hs_to_xy(*hs_color)
-            publish(CONF_XY_COMMAND_TOPIC, f"{xy_color[0]},{xy_color[1]}")
+            await publish(CONF_XY_COMMAND_TOPIC, f"{xy_color[0]},{xy_color[1]}")
             should_update |= set_optimistic(
                 ATTR_HS_COLOR, hs_color, condition_attribute=ATTR_XY_COLOR
             )
@@ -909,7 +909,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
         ):
             scaled = scale_rgbx(rgb)
             rgb_s = render_rgbx(scaled, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
-            publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
+            await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
             should_update |= set_optimistic(ATTR_RGB_COLOR, rgb, COLOR_MODE_RGB)
 
         if (
@@ -919,7 +919,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
         ):
             scaled = scale_rgbx(rgbw)
             rgbw_s = render_rgbx(scaled, CONF_RGBW_COMMAND_TEMPLATE, COLOR_MODE_RGBW)
-            publish(CONF_RGBW_COMMAND_TOPIC, rgbw_s)
+            await publish(CONF_RGBW_COMMAND_TOPIC, rgbw_s)
             should_update |= set_optimistic(ATTR_RGBW_COLOR, rgbw, COLOR_MODE_RGBW)
 
         if (
@@ -929,7 +929,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
         ):
             scaled = scale_rgbx(rgbww)
             rgbww_s = render_rgbx(scaled, CONF_RGBWW_COMMAND_TEMPLATE, COLOR_MODE_RGBWW)
-            publish(CONF_RGBWW_COMMAND_TOPIC, rgbww_s)
+            await publish(CONF_RGBWW_COMMAND_TOPIC, rgbww_s)
             should_update |= set_optimistic(ATTR_RGBWW_COLOR, rgbww, COLOR_MODE_RGBWW)
 
         if (
@@ -937,7 +937,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             and self._topic[CONF_XY_COMMAND_TOPIC] is not None
             and not self._legacy_mode
         ):
-            publish(CONF_XY_COMMAND_TOPIC, f"{xy_color[0]},{xy_color[1]}")
+            await publish(CONF_XY_COMMAND_TOPIC, f"{xy_color[0]},{xy_color[1]}")
             should_update |= set_optimistic(ATTR_XY_COLOR, xy_color, COLOR_MODE_XY)
 
         if (
@@ -951,7 +951,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             )
             # Make sure the brightness is not rounded down to 0
             device_brightness = max(device_brightness, 1)
-            publish(CONF_BRIGHTNESS_COMMAND_TOPIC, device_brightness)
+            await publish(CONF_BRIGHTNESS_COMMAND_TOPIC, device_brightness)
             should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
         elif (
             ATTR_BRIGHTNESS in kwargs
@@ -964,7 +964,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             brightness = kwargs[ATTR_BRIGHTNESS]
             rgb = scale_rgbx(color_util.color_hsv_to_RGB(*hs_color, 100), brightness)
             rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
-            publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
+            await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
             should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
         elif (
             ATTR_BRIGHTNESS in kwargs
@@ -975,7 +975,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             rgb_color = self._rgb_color if self._rgb_color is not None else (255,) * 3
             rgb = scale_rgbx(rgb_color, kwargs[ATTR_BRIGHTNESS])
             rgb_s = render_rgbx(rgb, CONF_RGB_COMMAND_TEMPLATE, COLOR_MODE_RGB)
-            publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
+            await publish(CONF_RGB_COMMAND_TOPIC, rgb_s)
             should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
         elif (
             ATTR_BRIGHTNESS in kwargs
@@ -988,7 +988,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             )
             rgbw = scale_rgbx(rgbw_color, kwargs[ATTR_BRIGHTNESS])
             rgbw_s = render_rgbx(rgbw, CONF_RGBW_COMMAND_TEMPLATE, COLOR_MODE_RGBW)
-            publish(CONF_RGBW_COMMAND_TOPIC, rgbw_s)
+            await publish(CONF_RGBW_COMMAND_TOPIC, rgbw_s)
             should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
         elif (
             ATTR_BRIGHTNESS in kwargs
@@ -1001,7 +1001,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             )
             rgbww = scale_rgbx(rgbww_color, kwargs[ATTR_BRIGHTNESS])
             rgbww_s = render_rgbx(rgbww, CONF_RGBWW_COMMAND_TEMPLATE, COLOR_MODE_RGBWW)
-            publish(CONF_RGBWW_COMMAND_TOPIC, rgbww_s)
+            await publish(CONF_RGBWW_COMMAND_TOPIC, rgbww_s)
             should_update |= set_optimistic(ATTR_BRIGHTNESS, kwargs[ATTR_BRIGHTNESS])
         if (
             ATTR_COLOR_TEMP in kwargs
@@ -1012,7 +1012,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             if tpl:
                 color_temp = tpl({"value": color_temp})
 
-            publish(CONF_COLOR_TEMP_COMMAND_TOPIC, color_temp)
+            await publish(CONF_COLOR_TEMP_COMMAND_TOPIC, color_temp)
             should_update |= set_optimistic(
                 ATTR_COLOR_TEMP, kwargs[ATTR_COLOR_TEMP], COLOR_MODE_COLOR_TEMP
             )
@@ -1020,14 +1020,14 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
         if ATTR_EFFECT in kwargs and self._topic[CONF_EFFECT_COMMAND_TOPIC] is not None:
             effect = kwargs[ATTR_EFFECT]
             if effect in self._config.get(CONF_EFFECT_LIST):
-                publish(CONF_EFFECT_COMMAND_TOPIC, effect)
+                await publish(CONF_EFFECT_COMMAND_TOPIC, effect)
                 should_update |= set_optimistic(ATTR_EFFECT, effect)
 
         if ATTR_WHITE in kwargs and self._topic[CONF_WHITE_COMMAND_TOPIC] is not None:
             percent_white = float(kwargs[ATTR_WHITE]) / 255
             white_scale = self._config[CONF_WHITE_SCALE]
             device_white_value = min(round(percent_white * white_scale), white_scale)
-            publish(CONF_WHITE_COMMAND_TOPIC, device_white_value)
+            await publish(CONF_WHITE_COMMAND_TOPIC, device_white_value)
             should_update |= set_optimistic(
                 ATTR_BRIGHTNESS,
                 kwargs[ATTR_WHITE],
@@ -1041,11 +1041,11 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
             percent_white = float(kwargs[ATTR_WHITE_VALUE]) / 255
             white_scale = self._config[CONF_WHITE_VALUE_SCALE]
             device_white_value = min(round(percent_white * white_scale), white_scale)
-            publish(CONF_WHITE_VALUE_COMMAND_TOPIC, device_white_value)
+            await publish(CONF_WHITE_VALUE_COMMAND_TOPIC, device_white_value)
             should_update |= set_optimistic(ATTR_WHITE_VALUE, kwargs[ATTR_WHITE_VALUE])
 
         if on_command_type == "last":
-            publish(CONF_COMMAND_TOPIC, self._payload["on"])
+            await publish(CONF_COMMAND_TOPIC, self._payload["on"])
             should_update = True
 
         if self._optimistic:
@@ -1061,7 +1061,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             self._payload["off"],
diff --git a/homeassistant/components/mqtt/light/schema_json.py b/homeassistant/components/mqtt/light/schema_json.py
index 9915c2455df..412205ea9cf 100644
--- a/homeassistant/components/mqtt/light/schema_json.py
+++ b/homeassistant/components/mqtt/light/schema_json.py
@@ -621,7 +621,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
                 self._white_value = kwargs[ATTR_WHITE_VALUE]
                 should_update = True
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             json.dumps(message),
@@ -646,7 +646,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
 
         self._set_flash_and_transition(message, **kwargs)
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topic[CONF_COMMAND_TOPIC],
             json.dumps(message),
diff --git a/homeassistant/components/mqtt/light/schema_template.py b/homeassistant/components/mqtt/light/schema_template.py
index 6b7396846e1..0a19fbb9836 100644
--- a/homeassistant/components/mqtt/light/schema_template.py
+++ b/homeassistant/components/mqtt/light/schema_template.py
@@ -374,7 +374,7 @@ class MqttLightTemplate(MqttEntity, LightEntity, RestoreEntity):
         if ATTR_TRANSITION in kwargs:
             values["transition"] = kwargs[ATTR_TRANSITION]
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topics[CONF_COMMAND_TOPIC],
             self._templates[CONF_COMMAND_ON_TEMPLATE].async_render(
@@ -399,7 +399,7 @@ class MqttLightTemplate(MqttEntity, LightEntity, RestoreEntity):
         if ATTR_TRANSITION in kwargs:
             values["transition"] = kwargs[ATTR_TRANSITION]
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._topics[CONF_COMMAND_TOPIC],
             self._templates[CONF_COMMAND_OFF_TEMPLATE].async_render(
diff --git a/homeassistant/components/mqtt/lock.py b/homeassistant/components/mqtt/lock.py
index fdcd4294b8c..8de4c4431b9 100644
--- a/homeassistant/components/mqtt/lock.py
+++ b/homeassistant/components/mqtt/lock.py
@@ -149,7 +149,7 @@ class MqttLock(MqttEntity, LockEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             self._config[CONF_PAYLOAD_LOCK],
@@ -166,7 +166,7 @@ class MqttLock(MqttEntity, LockEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             self._config[CONF_PAYLOAD_UNLOCK],
diff --git a/homeassistant/components/mqtt/number.py b/homeassistant/components/mqtt/number.py
index 902d828d911..e3ff03b2fbe 100644
--- a/homeassistant/components/mqtt/number.py
+++ b/homeassistant/components/mqtt/number.py
@@ -231,7 +231,7 @@ class MqttNumber(MqttEntity, NumberEntity, RestoreEntity):
             self._current_number = current_number
             self.async_write_ha_state()
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             current_number,
diff --git a/homeassistant/components/mqtt/scene.py b/homeassistant/components/mqtt/scene.py
index 6cf953ccf44..c2b201e20e6 100644
--- a/homeassistant/components/mqtt/scene.py
+++ b/homeassistant/components/mqtt/scene.py
@@ -121,7 +121,7 @@ class MqttScene(
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             self._config[CONF_PAYLOAD_ON],
diff --git a/homeassistant/components/mqtt/select.py b/homeassistant/components/mqtt/select.py
index b289bd53f0d..6ef0dbc3776 100644
--- a/homeassistant/components/mqtt/select.py
+++ b/homeassistant/components/mqtt/select.py
@@ -168,7 +168,7 @@ class MqttSelect(MqttEntity, SelectEntity, RestoreEntity):
             self._attr_current_option = option
             self.async_write_ha_state()
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             option,
diff --git a/homeassistant/components/mqtt/switch.py b/homeassistant/components/mqtt/switch.py
index 3a593adf1e3..13a241f7ab1 100644
--- a/homeassistant/components/mqtt/switch.py
+++ b/homeassistant/components/mqtt/switch.py
@@ -165,7 +165,7 @@ class MqttSwitch(MqttEntity, SwitchEntity, RestoreEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             self._config[CONF_PAYLOAD_ON],
@@ -182,7 +182,7 @@ class MqttSwitch(MqttEntity, SwitchEntity, RestoreEntity):
 
         This method is a coroutine.
         """
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._config[CONF_COMMAND_TOPIC],
             self._config[CONF_PAYLOAD_OFF],
diff --git a/homeassistant/components/mqtt/vacuum/schema_legacy.py b/homeassistant/components/mqtt/vacuum/schema_legacy.py
index d827853d603..63bb47ce21b 100644
--- a/homeassistant/components/mqtt/vacuum/schema_legacy.py
+++ b/homeassistant/components/mqtt/vacuum/schema_legacy.py
@@ -380,7 +380,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_TURN_ON == 0:
             return
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_TURN_ON],
@@ -395,7 +395,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_TURN_OFF == 0:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_TURN_OFF],
@@ -410,7 +410,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_STOP == 0:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_STOP],
@@ -425,7 +425,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_CLEAN_SPOT == 0:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_CLEAN_SPOT],
@@ -440,7 +440,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_LOCATE == 0:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_LOCATE],
@@ -455,7 +455,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_PAUSE == 0:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_START_PAUSE],
@@ -470,7 +470,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         if self.supported_features & SUPPORT_RETURN_HOME == 0:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._payloads[CONF_PAYLOAD_RETURN_TO_BASE],
@@ -487,7 +487,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         ) or fan_speed not in self._fan_speed_list:
             return None
 
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass, self._set_fan_speed_topic, fan_speed, self._qos, self._retain
         )
         self._status = f"Setting fan to {fan_speed}..."
@@ -503,7 +503,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
             message = json.dumps(message)
         else:
             message = command
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass, self._send_command_topic, message, self._qos, self._retain
         )
         self._status = f"Sending command {message}..."
diff --git a/homeassistant/components/mqtt/vacuum/schema_state.py b/homeassistant/components/mqtt/vacuum/schema_state.py
index 80f566ff5de..8c654a526e4 100644
--- a/homeassistant/components/mqtt/vacuum/schema_state.py
+++ b/homeassistant/components/mqtt/vacuum/schema_state.py
@@ -240,7 +240,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """Start the vacuum."""
         if self.supported_features & SUPPORT_START == 0:
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._config[CONF_PAYLOAD_START],
@@ -252,7 +252,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """Pause the vacuum."""
         if self.supported_features & SUPPORT_PAUSE == 0:
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._config[CONF_PAYLOAD_PAUSE],
@@ -264,7 +264,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """Stop the vacuum."""
         if self.supported_features & SUPPORT_STOP == 0:
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._config[CONF_PAYLOAD_STOP],
@@ -278,7 +278,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
             fan_speed not in self._fan_speed_list
         ):
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._set_fan_speed_topic,
             fan_speed,
@@ -290,7 +290,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """Tell the vacuum to return to its dock."""
         if self.supported_features & SUPPORT_RETURN_HOME == 0:
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._config[CONF_PAYLOAD_RETURN_TO_BASE],
@@ -302,7 +302,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """Perform a spot clean-up."""
         if self.supported_features & SUPPORT_CLEAN_SPOT == 0:
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._config[CONF_PAYLOAD_CLEAN_SPOT],
@@ -314,7 +314,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """Locate the vacuum (usually by playing a song)."""
         if self.supported_features & SUPPORT_LOCATE == 0:
             return None
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._command_topic,
             self._config[CONF_PAYLOAD_LOCATE],
@@ -332,7 +332,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
             message = json.dumps(message)
         else:
             message = command
-        mqtt.async_publish(
+        await mqtt.async_publish(
             self.hass,
             self._send_command_topic,
             message,
diff --git a/homeassistant/components/mqtt_eventstream/__init__.py b/homeassistant/components/mqtt_eventstream/__init__.py
index d31d6d1cd53..6e6596b3425 100644
--- a/homeassistant/components/mqtt_eventstream/__init__.py
+++ b/homeassistant/components/mqtt_eventstream/__init__.py
@@ -3,6 +3,7 @@ import json
 
 import voluptuous as vol
 
+from homeassistant.components import mqtt
 from homeassistant.components.mqtt import valid_publish_topic, valid_subscribe_topic
 from homeassistant.const import (
     ATTR_SERVICE_DATA,
@@ -53,15 +54,13 @@ BLOCKED_EVENTS = [
 
 async def async_setup(hass, config):
     """Set up the MQTT eventstream component."""
-    mqtt = hass.components.mqtt
     conf = config.get(DOMAIN, {})
     pub_topic = conf.get(CONF_PUBLISH_TOPIC)
     sub_topic = conf.get(CONF_SUBSCRIBE_TOPIC)
     ignore_event = conf.get(CONF_IGNORE_EVENT)
     ignore_event.append(EVENT_TIME_CHANGED)
 
-    @callback
-    def _event_publisher(event):
+    async def _event_publisher(event):
         """Handle events by publishing them on the MQTT queue."""
         if event.origin != EventOrigin.local:
             return
@@ -82,7 +81,7 @@ async def async_setup(hass, config):
 
         event_info = {"event_type": event.event_type, "event_data": event.data}
         msg = json.dumps(event_info, cls=JSONEncoder)
-        mqtt.async_publish(pub_topic, msg)
+        await mqtt.async_publish(hass, pub_topic, msg)
 
     # Only listen for local events if you are going to publish them.
     if pub_topic:
@@ -117,6 +116,6 @@ async def async_setup(hass, config):
 
     # Only subscribe if you specified a topic.
     if sub_topic:
-        await mqtt.async_subscribe(sub_topic, _event_receiver)
+        await mqtt.async_subscribe(hass, sub_topic, _event_receiver)
 
     return True
diff --git a/homeassistant/components/mqtt_statestream/__init__.py b/homeassistant/components/mqtt_statestream/__init__.py
index d7c971b7d35..d0a13a7384b 100644
--- a/homeassistant/components/mqtt_statestream/__init__.py
+++ b/homeassistant/components/mqtt_statestream/__init__.py
@@ -3,9 +3,9 @@ import json
 
 import voluptuous as vol
 
+from homeassistant.components import mqtt
 from homeassistant.components.mqtt import valid_publish_topic
 from homeassistant.const import MATCH_ALL
-from homeassistant.core import callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entityfilter import (
     INCLUDE_EXCLUDE_BASE_FILTER_SCHEMA,
@@ -44,8 +44,7 @@ async def async_setup(hass, config):
     if not base_topic.endswith("/"):
         base_topic = f"{base_topic}/"
 
-    @callback
-    def _state_publisher(entity_id, old_state, new_state):
+    async def _state_publisher(entity_id, old_state, new_state):
         if new_state is None:
             return
 
@@ -55,22 +54,30 @@ async def async_setup(hass, config):
         payload = new_state.state
 
         mybase = f"{base_topic}{entity_id.replace('.', '/')}/"
-        hass.components.mqtt.async_publish(f"{mybase}state", payload, 1, True)
+        await mqtt.async_publish(hass, f"{mybase}state", payload, 1, True)
 
         if publish_timestamps:
             if new_state.last_updated:
-                hass.components.mqtt.async_publish(
-                    f"{mybase}last_updated", new_state.last_updated.isoformat(), 1, True
+                await mqtt.async_publish(
+                    hass,
+                    f"{mybase}last_updated",
+                    new_state.last_updated.isoformat(),
+                    1,
+                    True,
                 )
             if new_state.last_changed:
-                hass.components.mqtt.async_publish(
-                    f"{mybase}last_changed", new_state.last_changed.isoformat(), 1, True
+                await mqtt.async_publish(
+                    hass,
+                    f"{mybase}last_changed",
+                    new_state.last_changed.isoformat(),
+                    1,
+                    True,
                 )
 
         if publish_attributes:
             for key, val in new_state.attributes.items():
                 encoded_val = json.dumps(val, cls=JSONEncoder)
-                hass.components.mqtt.async_publish(mybase + key, encoded_val, 1, True)
+                await mqtt.async_publish(hass, mybase + key, encoded_val, 1, True)
 
     async_track_state_change(hass, MATCH_ALL, _state_publisher)
     return True
diff --git a/homeassistant/components/mysensors/gateway.py b/homeassistant/components/mysensors/gateway.py
index e7f97792493..41311f45b03 100644
--- a/homeassistant/components/mysensors/gateway.py
+++ b/homeassistant/components/mysensors/gateway.py
@@ -185,7 +185,9 @@ async def _get_gateway(
 
         def pub_callback(topic: str, payload: str, qos: int, retain: bool) -> None:
             """Call MQTT publish function."""
-            mqtt.async_publish(topic, payload, qos, retain)
+            hass.async_create_task(
+                mqtt.async_publish(hass, topic, payload, qos, retain)
+            )
 
         def sub_callback(
             topic: str, sub_cb: Callable[[str, ReceivePayloadType, int], None], qos: int
diff --git a/homeassistant/components/ozw/__init__.py b/homeassistant/components/ozw/__init__.py
index 238e7dcd8cd..04c7c3854bc 100644
--- a/homeassistant/components/ozw/__init__.py
+++ b/homeassistant/components/ozw/__init__.py
@@ -106,7 +106,7 @@ async def async_setup_entry(  # noqa: C901
                 _LOGGER.error("MQTT integration is not set up")
                 return
 
-            mqtt.async_publish(hass, topic, json.dumps(payload))
+            hass.async_create_task(mqtt.async_publish(hass, topic, json.dumps(payload)))
 
         manager_options["send_message"] = send_message
 
diff --git a/homeassistant/components/snips/__init__.py b/homeassistant/components/snips/__init__.py
index 256a4ae8719..4c65bd30a77 100644
--- a/homeassistant/components/snips/__init__.py
+++ b/homeassistant/components/snips/__init__.py
@@ -6,7 +6,6 @@ import logging
 import voluptuous as vol
 
 from homeassistant.components import mqtt
-from homeassistant.core import callback
 from homeassistant.helpers import config_validation as cv, intent
 
 DOMAIN = "snips"
@@ -90,22 +89,17 @@ SERVICE_SCHEMA_FEEDBACK = vol.Schema(
 async def async_setup(hass, config):
     """Activate Snips component."""
 
-    @callback
-    def async_set_feedback(site_ids, state):
+    async def async_set_feedback(site_ids, state):
         """Set Feedback sound state."""
         site_ids = site_ids if site_ids else config[DOMAIN].get(CONF_SITE_IDS)
         topic = FEEDBACK_ON_TOPIC if state else FEEDBACK_OFF_TOPIC
         for site_id in site_ids:
             payload = json.dumps({"siteId": site_id})
-            hass.components.mqtt.async_publish(
-                FEEDBACK_ON_TOPIC, "", qos=0, retain=False
-            )
-            hass.components.mqtt.async_publish(
-                topic, payload, qos=int(state), retain=state
-            )
+            await mqtt.async_publish(hass, FEEDBACK_ON_TOPIC, "", qos=0, retain=False)
+            await mqtt.async_publish(hass, topic, payload, qos=int(state), retain=state)
 
     if CONF_FEEDBACK in config[DOMAIN]:
-        async_set_feedback(None, config[DOMAIN][CONF_FEEDBACK])
+        await async_set_feedback(None, config[DOMAIN][CONF_FEEDBACK])
 
     async def message_received(msg):
         """Handle new messages on MQTT."""
@@ -153,7 +147,7 @@ async def async_setup(hass, config):
                 notification["text"] = intent_response.speech["plain"]["speech"]
 
             _LOGGER.debug("send_response %s", json.dumps(notification))
-            mqtt.async_publish(
+            await mqtt.async_publish(
                 hass, "hermes/dialogueManager/endSession", json.dumps(notification)
             )
         except intent.UnknownIntent:
@@ -163,7 +157,7 @@ async def async_setup(hass, config):
         except intent.IntentError:
             _LOGGER.exception("Error while handling intent: %s", intent_type)
 
-    await hass.components.mqtt.async_subscribe(INTENT_TOPIC, message_received)
+    await mqtt.async_subscribe(hass, INTENT_TOPIC, message_received)
 
     async def snips_say(call):
         """Send a Snips notification message."""
@@ -172,7 +166,7 @@ async def async_setup(hass, config):
             "customData": call.data.get(ATTR_CUSTOM_DATA, ""),
             "init": {"type": "notification", "text": call.data.get(ATTR_TEXT)},
         }
-        mqtt.async_publish(
+        await mqtt.async_publish(
             hass, "hermes/dialogueManager/startSession", json.dumps(notification)
         )
         return
@@ -189,18 +183,18 @@ async def async_setup(hass, config):
                 "intentFilter": call.data.get(ATTR_INTENT_FILTER, []),
             },
         }
-        mqtt.async_publish(
+        await mqtt.async_publish(
             hass, "hermes/dialogueManager/startSession", json.dumps(notification)
         )
         return
 
     async def feedback_on(call):
         """Turn feedback sounds on."""
-        async_set_feedback(call.data.get(ATTR_SITE_ID), True)
+        await async_set_feedback(call.data.get(ATTR_SITE_ID), True)
 
     async def feedback_off(call):
         """Turn feedback sounds off."""
-        async_set_feedback(call.data.get(ATTR_SITE_ID), False)
+        await async_set_feedback(call.data.get(ATTR_SITE_ID), False)
 
     hass.services.async_register(
         DOMAIN, SERVICE_SAY, snips_say, schema=SERVICE_SCHEMA_SAY
diff --git a/homeassistant/components/tasmota/__init__.py b/homeassistant/components/tasmota/__init__.py
index fd156e20c3c..f33359347b8 100644
--- a/homeassistant/components/tasmota/__init__.py
+++ b/homeassistant/components/tasmota/__init__.py
@@ -55,7 +55,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         qos: int | None = None,
         retain: bool | None = None,
     ) -> None:
-        mqtt.async_publish(hass, topic, payload, qos, retain)
+        if qos is None:
+            qos = 0
+        if retain is None:
+            retain = False
+        hass.async_create_task(mqtt.async_publish(hass, topic, payload, qos, retain))
 
     async def _subscribe_topics(sub_state: dict | None, topics: dict) -> dict:
         # Optionally mark message handlers as callback
diff --git a/tests/components/mqtt/test_init.py b/tests/components/mqtt/test_init.py
index 7c6d482c7ec..26ceb583818 100644
--- a/tests/components/mqtt/test_init.py
+++ b/tests/components/mqtt/test_init.py
@@ -12,7 +12,6 @@ from homeassistant.components import mqtt, websocket_api
 from homeassistant.components.mqtt import debug_info
 from homeassistant.components.mqtt.mixins import MQTT_ENTITY_DEVICE_INFO_SCHEMA
 from homeassistant.const import (
-    EVENT_CALL_SERVICE,
     EVENT_HOMEASSISTANT_STARTED,
     EVENT_HOMEASSISTANT_STOP,
     TEMP_CELSIUS,
@@ -92,29 +91,51 @@ async def test_mqtt_disconnects_on_home_assistant_stop(hass, mqtt_mock):
     assert mqtt_mock.async_disconnect.called
 
 
-async def test_publish_calls_service(hass, mqtt_mock, calls, record_calls):
-    """Test the publishing of call to services."""
-    hass.bus.async_listen_once(EVENT_CALL_SERVICE, record_calls)
-
-    mqtt.async_publish(hass, "test-topic", "test-payload")
+async def test_publish_(hass, mqtt_mock):
+    """Test the publish function."""
+    await mqtt.async_publish(hass, "test-topic", "test-payload")
     await hass.async_block_till_done()
+    assert mqtt_mock.async_publish.called
+    assert mqtt_mock.async_publish.call_args[0] == (
+        "test-topic",
+        "test-payload",
+        0,
+        False,
+    )
+    mqtt_mock.reset_mock()
 
-    assert len(calls) == 1
-    assert calls[0][0].data["service_data"][mqtt.ATTR_TOPIC] == "test-topic"
-    assert calls[0][0].data["service_data"][mqtt.ATTR_PAYLOAD] == "test-payload"
-    assert mqtt.ATTR_QOS not in calls[0][0].data["service_data"]
-    assert mqtt.ATTR_RETAIN not in calls[0][0].data["service_data"]
-
-    hass.bus.async_listen_once(EVENT_CALL_SERVICE, record_calls)
-
-    mqtt.async_publish(hass, "test-topic", "test-payload", 2, True)
+    await mqtt.async_publish(hass, "test-topic", "test-payload", 2, True)
     await hass.async_block_till_done()
+    assert mqtt_mock.async_publish.called
+    assert mqtt_mock.async_publish.call_args[0] == (
+        "test-topic",
+        "test-payload",
+        2,
+        True,
+    )
+    mqtt_mock.reset_mock()
 
-    assert len(calls) == 2
-    assert calls[1][0].data["service_data"][mqtt.ATTR_TOPIC] == "test-topic"
-    assert calls[1][0].data["service_data"][mqtt.ATTR_PAYLOAD] == "test-payload"
-    assert calls[1][0].data["service_data"][mqtt.ATTR_QOS] == 2
-    assert calls[1][0].data["service_data"][mqtt.ATTR_RETAIN] is True
+    mqtt.publish(hass, "test-topic2", "test-payload2")
+    await hass.async_block_till_done()
+    assert mqtt_mock.async_publish.called
+    assert mqtt_mock.async_publish.call_args[0] == (
+        "test-topic2",
+        "test-payload2",
+        0,
+        False,
+    )
+    mqtt_mock.reset_mock()
+
+    mqtt.publish(hass, "test-topic2", "test-payload2", 2, True)
+    await hass.async_block_till_done()
+    assert mqtt_mock.async_publish.called
+    assert mqtt_mock.async_publish.call_args[0] == (
+        "test-topic2",
+        "test-payload2",
+        2,
+        True,
+    )
+    mqtt_mock.reset_mock()
 
 
 async def test_service_call_without_topic_does_not_publish(hass, mqtt_mock):
@@ -134,18 +155,6 @@ async def test_service_call_with_template_payload_renders_template(hass, mqtt_mo
 
     If 'payload_template' is provided and 'payload' is not, then render it.
     """
-    mqtt.publish_template(hass, "test/topic", "{{ 1+1 }}")
-    await hass.async_block_till_done()
-    assert mqtt_mock.async_publish.called
-    assert mqtt_mock.async_publish.call_args[0][1] == "2"
-    mqtt_mock.reset_mock()
-
-    mqtt.async_publish_template(hass, "test/topic", "{{ 2+2 }}")
-    await hass.async_block_till_done()
-    assert mqtt_mock.async_publish.called
-    assert mqtt_mock.async_publish.call_args[0][1] == "4"
-    mqtt_mock.reset_mock()
-
     await hass.services.async_call(
         mqtt.DOMAIN,
         mqtt.SERVICE_PUBLISH,
diff --git a/tests/components/ozw/test_websocket_api.py b/tests/components/ozw/test_websocket_api.py
index ad3b568b62a..2cbe69f1c98 100644
--- a/tests/components/ozw/test_websocket_api.py
+++ b/tests/components/ozw/test_websocket_api.py
@@ -43,7 +43,7 @@ from homeassistant.components.websocket_api.const import (
 from .common import MQTTMessage, setup_ozw
 
 
-async def test_websocket_api(hass, generic_data, hass_ws_client):
+async def test_websocket_api(hass, generic_data, hass_ws_client, mqtt_mock):
     """Test the ozw websocket api."""
     await setup_ozw(hass, fixture=generic_data)
     client = await hass_ws_client(hass)
@@ -280,7 +280,7 @@ async def test_websocket_api(hass, generic_data, hass_ws_client):
     assert result["code"] == ERR_NOT_FOUND
 
 
-async def test_ws_locks(hass, lock_data, hass_ws_client):
+async def test_ws_locks(hass, lock_data, hass_ws_client, mqtt_mock):
     """Test lock websocket apis."""
     await setup_ozw(hass, fixture=lock_data)
     client = await hass_ws_client(hass)
@@ -319,7 +319,9 @@ async def test_ws_locks(hass, lock_data, hass_ws_client):
     assert msg["success"]
 
 
-async def test_refresh_node(hass, generic_data, sent_messages, hass_ws_client):
+async def test_refresh_node(
+    hass, generic_data, sent_messages, hass_ws_client, mqtt_mock
+):
     """Test the ozw refresh node api."""
     receive_message = await setup_ozw(hass, fixture=generic_data)
     client = await hass_ws_client(hass)
@@ -368,7 +370,7 @@ async def test_refresh_node(hass, generic_data, sent_messages, hass_ws_client):
     assert result["node_query_stage"] == "versions"
 
 
-async def test_refresh_node_unsubscribe(hass, generic_data, hass_ws_client):
+async def test_refresh_node_unsubscribe(hass, generic_data, hass_ws_client, mqtt_mock):
     """Test unsubscribing the ozw refresh node api."""
     await setup_ozw(hass, fixture=generic_data)
     client = await hass_ws_client(hass)
diff --git a/tests/components/snips/test_init.py b/tests/components/snips/test_init.py
index 82811b61925..14e58d54ebe 100644
--- a/tests/components/snips/test_init.py
+++ b/tests/components/snips/test_init.py
@@ -6,7 +6,6 @@ import pytest
 import voluptuous as vol
 
 from homeassistant.bootstrap import async_setup_component
-from homeassistant.components.mqtt import MQTT_PUBLISH_SCHEMA
 import homeassistant.components.snips as snips
 from homeassistant.helpers.intent import ServiceIntentHandler, async_register
 
@@ -47,38 +46,36 @@ async def test_snips_bad_config(hass, mqtt_mock):
 
 async def test_snips_config_feedback_on(hass, mqtt_mock):
     """Test Snips Config."""
-    calls = async_mock_service(hass, "mqtt", "publish", MQTT_PUBLISH_SCHEMA)
     result = await async_setup_component(
         hass, "snips", {"snips": {"feedback_sounds": True}}
     )
     assert result
     await hass.async_block_till_done()
 
-    assert len(calls) == 2
-    topic = calls[0].data["topic"]
+    assert mqtt_mock.async_publish.call_count == 2
+    topic = mqtt_mock.async_publish.call_args_list[0][0][0]
     assert topic == "hermes/feedback/sound/toggleOn"
-    topic = calls[1].data["topic"]
+    topic = mqtt_mock.async_publish.call_args_list[1][0][0]
     assert topic == "hermes/feedback/sound/toggleOn"
-    assert calls[1].data["qos"] == 1
-    assert calls[1].data["retain"]
+    assert mqtt_mock.async_publish.call_args_list[1][0][2] == 1
+    assert mqtt_mock.async_publish.call_args_list[1][0][3]
 
 
 async def test_snips_config_feedback_off(hass, mqtt_mock):
     """Test Snips Config."""
-    calls = async_mock_service(hass, "mqtt", "publish", MQTT_PUBLISH_SCHEMA)
     result = await async_setup_component(
         hass, "snips", {"snips": {"feedback_sounds": False}}
     )
     assert result
     await hass.async_block_till_done()
 
-    assert len(calls) == 2
-    topic = calls[0].data["topic"]
+    assert mqtt_mock.async_publish.call_count == 2
+    topic = mqtt_mock.async_publish.call_args_list[0][0][0]
     assert topic == "hermes/feedback/sound/toggleOn"
-    topic = calls[1].data["topic"]
+    topic = mqtt_mock.async_publish.call_args_list[1][0][0]
     assert topic == "hermes/feedback/sound/toggleOff"
-    assert calls[1].data["qos"] == 0
-    assert not calls[1].data["retain"]
+    assert mqtt_mock.async_publish.call_args_list[1][0][2] == 0
+    assert not mqtt_mock.async_publish.call_args_list[1][0][3]
 
 
 async def test_snips_config_no_feedback(hass, mqtt_mock):
@@ -232,7 +229,6 @@ async def test_snips_intent_with_duration(hass, mqtt_mock):
 
 async def test_intent_speech_response(hass, mqtt_mock):
     """Test intent speech response via Snips."""
-    calls = async_mock_service(hass, "mqtt", "publish", MQTT_PUBLISH_SCHEMA)
     result = await async_setup_component(hass, "snips", {"snips": {}})
     assert result
     result = await async_setup_component(
@@ -261,9 +257,9 @@ async def test_intent_speech_response(hass, mqtt_mock):
     async_fire_mqtt_message(hass, "hermes/intent/spokenIntent", payload)
     await hass.async_block_till_done()
 
-    assert len(calls) == 1
-    payload = json.loads(calls[0].data["payload"])
-    topic = calls[0].data["topic"]
+    assert mqtt_mock.async_publish.call_count == 1
+    payload = json.loads(mqtt_mock.async_publish.call_args[0][1])
+    topic = mqtt_mock.async_publish.call_args[0][0]
     assert payload["sessionId"] == "abcdef0123456789"
     assert payload["text"] == "I am speaking to you"
     assert topic == "hermes/dialogueManager/endSession"