diff --git a/homeassistant/components/esphome/config_flow.py b/homeassistant/components/esphome/config_flow.py
index 67e94121e1d..d1948df0690 100644
--- a/homeassistant/components/esphome/config_flow.py
+++ b/homeassistant/components/esphome/config_flow.py
@@ -6,7 +6,7 @@ from collections import OrderedDict
 from collections.abc import Mapping
 import json
 import logging
-from typing import Any
+from typing import Any, cast
 
 from aioesphomeapi import (
     APIClient,
@@ -31,6 +31,8 @@ from homeassistant.config_entries import (
 from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_PORT
 from homeassistant.core import callback
 from homeassistant.helpers.device_registry import format_mac
+from homeassistant.helpers.service_info.mqtt import MqttServiceInfo
+from homeassistant.util.json import json_loads_object
 
 from .const import (
     CONF_ALLOW_SERVICE_CALLS,
@@ -250,6 +252,42 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
 
         return await self.async_step_discovery_confirm()
 
+    async def async_step_mqtt(
+        self, discovery_info: MqttServiceInfo
+    ) -> ConfigFlowResult:
+        """Handle MQTT discovery."""
+        device_info = json_loads_object(discovery_info.payload)
+        if "mac" not in device_info:
+            return self.async_abort(reason="mqtt_missing_mac")
+
+        # there will be no port if the API is not enabled
+        if "port" not in device_info:
+            return self.async_abort(reason="mqtt_missing_api")
+
+        if "ip" not in device_info:
+            return self.async_abort(reason="mqtt_missing_ip")
+
+        # mac address is lowercase and without :, normalize it
+        unformatted_mac = cast(str, device_info["mac"])
+        mac_address = format_mac(unformatted_mac)
+
+        device_name = cast(str, device_info["name"])
+
+        self._device_name = device_name
+        self._name = cast(str, device_info.get("friendly_name", device_name))
+        self._host = cast(str, device_info["ip"])
+        self._port = cast(int, device_info["port"])
+
+        self._noise_required = "api_encryption" in device_info
+
+        # Check if already configured
+        await self.async_set_unique_id(mac_address)
+        self._abort_if_unique_id_configured(
+            updates={CONF_HOST: self._host, CONF_PORT: self._port}
+        )
+
+        return await self.async_step_discovery_confirm()
+
     async def async_step_dhcp(
         self, discovery_info: dhcp.DhcpServiceInfo
     ) -> ConfigFlowResult:
diff --git a/homeassistant/components/esphome/manifest.json b/homeassistant/components/esphome/manifest.json
index cde44fa3231..e41c61a40d5 100644
--- a/homeassistant/components/esphome/manifest.json
+++ b/homeassistant/components/esphome/manifest.json
@@ -14,6 +14,7 @@
   "integration_type": "device",
   "iot_class": "local_push",
   "loggers": ["aioesphomeapi", "noiseprotocol", "bleak_esphome"],
+  "mqtt": ["esphome/discover/#"],
   "requirements": [
     "aioesphomeapi==24.3.0",
     "esphome-dashboard-api==1.2.3",
diff --git a/homeassistant/components/esphome/strings.json b/homeassistant/components/esphome/strings.json
index e38e8e1a2c4..205b0b10744 100644
--- a/homeassistant/components/esphome/strings.json
+++ b/homeassistant/components/esphome/strings.json
@@ -5,7 +5,10 @@
       "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
       "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
       "mdns_missing_mac": "Missing MAC address in MDNS properties.",
-      "service_received": "Service received"
+      "service_received": "Service received",
+      "mqtt_missing_mac": "Missing MAC address in MQTT properties.",
+      "mqtt_missing_api": "Missing API port in MQTT properties.",
+      "mqtt_missing_ip": "Missing IP address in MQTT properties."
     },
     "error": {
       "resolve_error": "Can't resolve address of the ESP. If this error persists, please set a static IP address",
diff --git a/homeassistant/generated/mqtt.py b/homeassistant/generated/mqtt.py
index 0c456774e4d..f73388b203c 100644
--- a/homeassistant/generated/mqtt.py
+++ b/homeassistant/generated/mqtt.py
@@ -10,6 +10,9 @@ MQTT = {
     "dsmr_reader": [
         "dsmr/#",
     ],
+    "esphome": [
+        "esphome/discover/#",
+    ],
     "fully_kiosk": [
         "fully/deviceInfo/+",
     ],
diff --git a/tests/components/esphome/test_config_flow.py b/tests/components/esphome/test_config_flow.py
index 439092d9fb1..1142d2b0411 100644
--- a/tests/components/esphome/test_config_flow.py
+++ b/tests/components/esphome/test_config_flow.py
@@ -30,6 +30,7 @@ from homeassistant.components.hassio import HassioServiceInfo
 from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
+from homeassistant.helpers.service_info.mqtt import MqttServiceInfo
 
 from . import VALID_NOISE_PSK
 
@@ -1414,3 +1415,72 @@ async def test_user_discovers_name_no_dashboard(
         CONF_DEVICE_NAME: "test",
     }
     assert mock_client.noise_psk == VALID_NOISE_PSK
+
+
+async def mqtt_discovery_test_abort(hass: HomeAssistant, payload: str, reason: str):
+    """Test discovery aborted."""
+    service_info = MqttServiceInfo(
+        topic="esphome/discover/test",
+        payload=payload,
+        qos=0,
+        retain=False,
+        subscribed_topic="esphome/discover/#",
+        timestamp=None,
+    )
+    flow = await hass.config_entries.flow.async_init(
+        "esphome", context={"source": config_entries.SOURCE_MQTT}, data=service_info
+    )
+    assert flow["type"] is FlowResultType.ABORT
+    assert flow["reason"] == reason
+
+
+async def test_discovery_mqtt_no_mac(
+    hass: HomeAssistant, mock_client, mock_zeroconf: None, mock_setup_entry: None
+) -> None:
+    """Test discovery aborted if mac is missing in MQTT payload."""
+    await mqtt_discovery_test_abort(hass, "{}", "mqtt_missing_mac")
+
+
+async def test_discovery_mqtt_no_api(
+    hass: HomeAssistant, mock_client, mock_zeroconf: None, mock_setup_entry: None
+) -> None:
+    """Test discovery aborted if api/port is missing in MQTT payload."""
+    await mqtt_discovery_test_abort(hass, '{"mac":"abcdef123456"}', "mqtt_missing_api")
+
+
+async def test_discovery_mqtt_no_ip(
+    hass: HomeAssistant, mock_client, mock_zeroconf: None, mock_setup_entry: None
+) -> None:
+    """Test discovery aborted if ip is missing in MQTT payload."""
+    await mqtt_discovery_test_abort(
+        hass, '{"mac":"abcdef123456","port":6053}', "mqtt_missing_ip"
+    )
+
+
+async def test_discovery_mqtt_initiation(
+    hass: HomeAssistant, mock_client, mock_zeroconf: None, mock_setup_entry: None
+) -> None:
+    """Test discovery importing works."""
+    service_info = MqttServiceInfo(
+        topic="esphome/discover/test",
+        payload='{"name":"mock_name","mac":"1122334455aa","port":6053,"ip":"192.168.43.183"}',
+        qos=0,
+        retain=False,
+        subscribed_topic="esphome/discover/#",
+        timestamp=None,
+    )
+    flow = await hass.config_entries.flow.async_init(
+        "esphome", context={"source": config_entries.SOURCE_MQTT}, data=service_info
+    )
+
+    result = await hass.config_entries.flow.async_configure(
+        flow["flow_id"], user_input={}
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "test"
+    assert result["data"][CONF_HOST] == "192.168.43.183"
+    assert result["data"][CONF_PORT] == 6053
+
+    assert result["result"]
+    assert result["result"].unique_id == "11:22:33:44:55:aa"