diff --git a/homeassistant/components/mqtt/abbreviations.py b/homeassistant/components/mqtt/abbreviations.py
index 4cc391e0ca7..6bd2bb47923 100644
--- a/homeassistant/components/mqtt/abbreviations.py
+++ b/homeassistant/components/mqtt/abbreviations.py
@@ -18,6 +18,8 @@ ABBREVIATIONS = {
     "bri_stat_t": "brightness_state_topic",
     "bri_tpl": "brightness_template",
     "bri_val_tpl": "brightness_value_template",
+    "cln_segmnts_cmd_t": "clean_segments_command_topic",
+    "cln_segmnts_cmd_tpl": "clean_segments_command_template",
     "clr_temp_cmd_tpl": "color_temp_command_template",
     "clrm_stat_t": "color_mode_state_topic",
     "clrm_val_tpl": "color_mode_value_template",
diff --git a/homeassistant/components/mqtt/vacuum.py b/homeassistant/components/mqtt/vacuum.py
index 6896d51ef93..fb1166250f1 100644
--- a/homeassistant/components/mqtt/vacuum.py
+++ b/homeassistant/components/mqtt/vacuum.py
@@ -10,12 +10,13 @@ import voluptuous as vol
 from homeassistant.components import vacuum
 from homeassistant.components.vacuum import (
     ENTITY_ID_FORMAT,
+    Segment,
     StateVacuumEntity,
     VacuumActivity,
     VacuumEntityFeature,
 )
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import ATTR_SUPPORTED_FEATURES, CONF_NAME
+from homeassistant.const import ATTR_SUPPORTED_FEATURES, CONF_NAME, CONF_UNIQUE_ID
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -27,13 +28,14 @@ from . import subscription
 from .config import MQTT_BASE_SCHEMA
 from .const import CONF_COMMAND_TOPIC, CONF_RETAIN, CONF_STATE_TOPIC
 from .entity import MqttEntity, async_setup_entity_entry_helper
-from .models import ReceiveMessage
+from .models import MqttCommandTemplate, ReceiveMessage
 from .schemas import MQTT_ENTITY_COMMON_SCHEMA
 from .util import valid_publish_topic
 
 PARALLEL_UPDATES = 0
 
 FAN_SPEED = "fan_speed"
+SEGMENTS = "segments"
 STATE = "state"
 
 STATE_IDLE = "idle"
@@ -52,6 +54,8 @@ POSSIBLE_STATES: dict[str, VacuumActivity] = {
     STATE_CLEANING: VacuumActivity.CLEANING,
 }
 
+CONF_CLEAN_SEGMENTS_COMMAND_TOPIC = "clean_segments_command_topic"
+CONF_CLEAN_SEGMENTS_COMMAND_TEMPLATE = "clean_segments_command_template"
 CONF_SUPPORTED_FEATURES = ATTR_SUPPORTED_FEATURES
 CONF_PAYLOAD_TURN_ON = "payload_turn_on"
 CONF_PAYLOAD_TURN_OFF = "payload_turn_off"
@@ -137,8 +141,22 @@ MQTT_VACUUM_ATTRIBUTES_BLOCKED = frozenset(
 MQTT_VACUUM_DOCS_URL = "https://www.home-assistant.io/integrations/vacuum.mqtt/"
 
 
-PLATFORM_SCHEMA_MODERN = MQTT_BASE_SCHEMA.extend(
+def validate_clean_area_config(config: ConfigType) -> ConfigType:
+    """Validate clean area configuration."""
+    if CONF_CLEAN_SEGMENTS_COMMAND_TOPIC not in config:
+        return config
+    if not config.get(CONF_UNIQUE_ID):
+        raise vol.Invalid(
+            f"Option `{CONF_CLEAN_SEGMENTS_COMMAND_TOPIC}` requires `{CONF_UNIQUE_ID}` to be configured"
+        )
+
+    return config
+
+
+_BASE_SCHEMA = MQTT_BASE_SCHEMA.extend(
     {
+        vol.Optional(CONF_CLEAN_SEGMENTS_COMMAND_TOPIC): valid_publish_topic,
+        vol.Optional(CONF_CLEAN_SEGMENTS_COMMAND_TEMPLATE): cv.template,
         vol.Optional(CONF_FAN_SPEED_LIST, default=[]): vol.All(
             cv.ensure_list, [cv.string]
         ),
@@ -164,7 +182,10 @@ PLATFORM_SCHEMA_MODERN = MQTT_BASE_SCHEMA.extend(
     }
 ).extend(MQTT_ENTITY_COMMON_SCHEMA.schema)
 
-DISCOVERY_SCHEMA = PLATFORM_SCHEMA_MODERN.extend({}, extra=vol.ALLOW_EXTRA)
+PLATFORM_SCHEMA_MODERN = vol.All(_BASE_SCHEMA, validate_clean_area_config)
+DISCOVERY_SCHEMA = vol.All(
+    _BASE_SCHEMA.extend({}, extra=vol.ALLOW_EXTRA), validate_clean_area_config
+)
 
 
 async def async_setup_entry(
@@ -191,9 +212,11 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
     _entity_id_format = ENTITY_ID_FORMAT
     _attributes_extra_blocked = MQTT_VACUUM_ATTRIBUTES_BLOCKED
 
+    _segments: list[Segment]
     _command_topic: str | None
     _set_fan_speed_topic: str | None
     _send_command_topic: str | None
+    _clean_segments_command_topic: str | None = None
     _payloads: dict[str, str | None]
 
     def __init__(
@@ -229,6 +252,14 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         self._attr_supported_features = _strings_to_services(
             supported_feature_strings, STRING_TO_SERVICE
         )
+        self._clean_segments_command_topic = config.get(
+            CONF_CLEAN_SEGMENTS_COMMAND_TOPIC
+        )
+        self._clean_segments_command_template = MqttCommandTemplate(
+            config.get(CONF_CLEAN_SEGMENTS_COMMAND_TEMPLATE),
+            entity=self,
+        ).async_render
+
         self._attr_fan_speed_list = config[CONF_FAN_SPEED_LIST]
         self._command_topic = config.get(CONF_COMMAND_TOPIC)
         self._set_fan_speed_topic = config.get(CONF_SET_FAN_SPEED_TOPIC)
@@ -262,6 +293,24 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
                 POSSIBLE_STATES[cast(str, state)] if payload[STATE] else None
             )
             del payload[STATE]
+        if (
+            (segments_payload := payload.pop(SEGMENTS, None))
+            and self._clean_segments_command_topic is not None
+            and isinstance(segments_payload, dict)
+            and (
+                segments := [
+                    Segment(id=segment_id, name=str(segment_name))
+                    for segment_id, segment_name in segments_payload.items()
+                ]
+            )
+        ):
+            self._segments = segments
+            self._attr_supported_features |= VacuumEntityFeature.CLEAN_AREA
+            if (last_seen := self.last_seen_segments) is not None and {
+                s.id: s for s in last_seen
+            } != {s.id: s for s in self._segments}:
+                self.async_create_segments_issue()
+
         self._update_state_attributes(payload)
 
     @callback
@@ -277,6 +326,20 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
         """(Re)Subscribe to topics."""
         subscription.async_subscribe_topics_internal(self.hass, self._sub_state)
 
+    async def async_clean_segments(self, segment_ids: list[str], **kwargs: Any) -> None:
+        """Perform an area clean."""
+        assert self._clean_segments_command_topic is not None
+        await self.async_publish_with_config(
+            self._clean_segments_command_topic,
+            self._clean_segments_command_template(
+                json_dumps(segment_ids), {"value": segment_ids}
+            ),
+        )
+
+    async def async_get_segments(self) -> list[Segment]:
+        """Return the available segments."""
+        return self._segments
+
     async def _async_publish_command(self, feature: VacuumEntityFeature) -> None:
         """Publish a command."""
         if self._command_topic is None:
diff --git a/tests/components/mqtt/test_vacuum.py b/tests/components/mqtt/test_vacuum.py
index ea5d9f8f8e7..52ac9ad64c2 100644
--- a/tests/components/mqtt/test_vacuum.py
+++ b/tests/components/mqtt/test_vacuum.py
@@ -3,7 +3,7 @@
 from copy import deepcopy
 import json
 from typing import Any
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import pytest
 
@@ -27,9 +27,15 @@ from homeassistant.components.vacuum import (
     SERVICE_STOP,
     VacuumActivity,
 )
-from homeassistant.const import CONF_NAME, ENTITY_MATCH_ALL, STATE_UNKNOWN
+from homeassistant.const import (
+    ATTR_SUPPORTED_FEATURES,
+    CONF_NAME,
+    ENTITY_MATCH_ALL,
+    STATE_UNKNOWN,
+)
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er, issue_registry as ir
 
 from .common import (
     help_custom_config,
@@ -63,7 +69,11 @@ from .common import (
 
 from tests.common import async_fire_mqtt_message
 from tests.components.vacuum import common
-from tests.typing import MqttMockHAClientGenerator, MqttMockPahoClient
+from tests.typing import (
+    MqttMockHAClientGenerator,
+    MqttMockPahoClient,
+    WebSocketGenerator,
+)
 
 COMMAND_TOPIC = "vacuum/command"
 SEND_COMMAND_TOPIC = "vacuum/send_command"
@@ -82,6 +92,17 @@ DEFAULT_CONFIG = {
     }
 }
 
+CONFIG_CLEAN_SEGMENTS = {
+    mqtt.DOMAIN: {
+        vacuum.DOMAIN: {
+            CONF_NAME: "test",
+            CONF_STATE_TOPIC: STATE_TOPIC,
+            "unique_id": "veryunique",
+            mqttvacuum.CONF_CLEAN_SEGMENTS_COMMAND_TOPIC: "vacuum/clean_segment",
+        }
+    }
+}
+
 DEFAULT_CONFIG_2 = {mqtt.DOMAIN: {vacuum.DOMAIN: {"name": "test"}}}
 
 CONFIG_ALL_SERVICES = help_custom_config(
@@ -294,6 +315,283 @@ async def test_command_without_command_topic(
     mqtt_mock.async_publish.reset_mock()
 
 
+@pytest.mark.parametrize("hass_config", [CONFIG_CLEAN_SEGMENTS])
+async def test_clean_segments_initial_setup_without_repair_issue(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+) -> None:
+    """Test initial setup does not fire repair flow after cleanable segments are received."""
+    await mqtt_mock_entry()
+    # Receive a valid state
+    state = hass.states.get("vacuum.test")
+    assert state.state == STATE_UNKNOWN
+
+    message = """{
+        "battery_level": 54,
+        "state": "cleaning",
+        "segments":{
+            "1":"Livingroom",
+            "2":"Kitchen",
+            "3":"Diningroom"
+        }
+    }"""
+    async_fire_mqtt_message(hass, "vacuum/state", message)
+    await hass.async_block_till_done()
+    state = hass.states.get("vacuum.test")
+    assert state.state == VacuumActivity.CLEANING
+    assert (
+        state.attributes.get(ATTR_SUPPORTED_FEATURES)
+        & vacuum.VacuumEntityFeature.CLEAN_AREA
+    )
+    issue_registry = ir.async_get(hass)
+    assert len(issue_registry.issues) == 0
+
+
+@pytest.mark.parametrize("hass_config", [CONFIG_CLEAN_SEGMENTS])
+async def test_clean_segments_command(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    entity_registry: er.EntityRegistry,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+) -> None:
+    """Test cleaning segments and repair flow."""
+    config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    entity_registry.async_get_or_create(
+        vacuum.DOMAIN,
+        mqtt.DOMAIN,
+        "veryunique",
+        config_entry=config_entry,
+        suggested_object_id="test",
+    )
+    entity_registry.async_update_entity_options(
+        "vacuum.test",
+        vacuum.DOMAIN,
+        {
+            "area_mapping": {"Nabu Casa": ["1", "2"]},
+            "last_seen_segments": [
+                {"id": "1", "name": "Livingroom"},
+                {"id": "2", "name": "Kitchen"},
+            ],
+        },
+    )
+    mqtt_mock = await mqtt_mock_entry()
+    await hass.async_block_till_done()
+    message = """{
+        "battery_level": 54,
+        "state": "idle",
+        "segments":{
+            "1":"Livingroom",
+            "2":"Kitchen"
+        }
+    }"""
+    async_fire_mqtt_message(hass, "vacuum/state", message)
+    await hass.async_block_till_done()
+    state = hass.states.get("vacuum.test")
+    assert state.state == VacuumActivity.IDLE
+    assert (
+        state.attributes.get(ATTR_SUPPORTED_FEATURES)
+        & vacuum.VacuumEntityFeature.CLEAN_AREA
+    )
+
+    issue_registry = ir.async_get(hass)
+    # We do not expect a repair flow as the segments did not change
+    assert len(issue_registry.issues) == 0
+
+    await common.async_clean_area(hass, ["Nabu Casa"], entity_id="vacuum.test")
+    assert (
+        call("vacuum/clean_segment", '["1","2"]', 0, False)
+        in mqtt_mock.async_publish.mock_calls
+    )
+    await hass.async_block_till_done()
+    message = """{
+        "battery_level": 54,
+        "state": "cleaning",
+        "segments":{
+            "1":"Livingroom",
+            "2":"Kitchen",
+            "3": "Diningroom"
+        }
+    }"""
+    async_fire_mqtt_message(hass, "vacuum/state", message)
+    await hass.async_block_till_done()
+    # We expect a repair issue now as the available segments have changed
+    assert len(issue_registry.issues) == 1
+
+    client = await hass_ws_client(hass)
+    await client.send_json_auto_id(
+        {"type": "vacuum/get_segments", "entity_id": "vacuum.test"}
+    )
+    msg = await client.receive_json()
+    assert msg["success"]
+    assert msg["result"]["segments"] == [
+        {"id": "1", "name": "Livingroom", "group": None},
+        {"id": "2", "name": "Kitchen", "group": None},
+        {"id": "3", "name": "Diningroom", "group": None},
+    ]
+
+
+@pytest.mark.parametrize(
+    "hass_config",
+    [
+        help_custom_config(
+            vacuum.DOMAIN,
+            CONFIG_CLEAN_SEGMENTS,
+            ({"clean_segments_command_template": "{{ ';'.join(value) }}"},),
+        )
+    ],
+)
+async def test_clean_segments_command_template(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    entity_registry: er.EntityRegistry,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+) -> None:
+    """Test clean segments with command template."""
+    mqtt_mock = await mqtt_mock_entry()
+    entity_registry.async_update_entity_options(
+        "vacuum.test",
+        vacuum.DOMAIN,
+        {
+            "area_mapping": {"Livingroom": ["1"], "Kitchen": ["2"]},
+            "last_seen_segments": [
+                {"id": "1", "name": "Livingroom"},
+                {"id": "2", "name": "Kitchen"},
+            ],
+        },
+    )
+    await hass.async_block_till_done()
+    message = """{
+        "battery_level": 54,
+        "state": "idle",
+        "segments":{
+            "1":"Livingroom",
+            "2":"Kitchen"
+        }
+    }"""
+    async_fire_mqtt_message(hass, "vacuum/state", message)
+    await hass.async_block_till_done()
+    state = hass.states.get("vacuum.test")
+    assert state.state == VacuumActivity.IDLE
+    assert (
+        state.attributes.get(ATTR_SUPPORTED_FEATURES)
+        & vacuum.VacuumEntityFeature.CLEAN_AREA
+    )
+
+    await common.async_clean_area(
+        hass, ["Livingroom", "Kitchen"], entity_id="vacuum.test"
+    )
+    assert (
+        call("vacuum/clean_segment", "1;2", 0, False)
+        in mqtt_mock.async_publish.mock_calls
+    )
+
+    client = await hass_ws_client(hass)
+    await client.send_json_auto_id(
+        {"type": "vacuum/get_segments", "entity_id": "vacuum.test"}
+    )
+    msg = await client.receive_json()
+    assert msg["success"]
+    assert msg["result"]["segments"] == [
+        {"id": "1", "name": "Livingroom", "group": None},
+        {"id": "2", "name": "Kitchen", "group": None},
+    ]
+
+
+@pytest.mark.usefixtures("hass")
+@pytest.mark.parametrize(
+    ("hass_config", "error_message"),
+    [
+        (
+            help_custom_config(
+                vacuum.DOMAIN,
+                DEFAULT_CONFIG,
+                (
+                    {
+                        "clean_segments_command_topic": "test-topic",
+                    },
+                ),
+            ),
+            "Option `clean_segments_command_topic` requires `unique_id` to be configured",
+        ),
+    ],
+)
+async def test_clean_segments_config_validation(
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    error_message: str,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test status clean segment config validation."""
+    await mqtt_mock_entry()
+    assert error_message in caplog.text
+
+
+async def test_removing_clean_segments_command_topic_resets_feature(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+) -> None:
+    """Test the clean area feature is reset if the vacuum is reconfigured.
+
+    The `clean_segments_command_topic` is required to support clean area support.
+    When this option is removed, the clean area feature should be reset.
+    """
+    await mqtt_mock_entry()
+
+    config_with_clean_segments_command_topic = CONFIG_CLEAN_SEGMENTS[mqtt.DOMAIN][
+        vacuum.DOMAIN
+    ]
+    async_fire_mqtt_message(
+        hass,
+        "homeassistant/vacuum/bla/config",
+        json.dumps(config_with_clean_segments_command_topic),
+    )
+    await hass.async_block_till_done()
+    message = """{
+        "battery_level": 54,
+        "state": "idle",
+        "segments":{
+            "1":"Livingroom",
+            "2":"Kitchen"
+        }
+    }"""
+    async_fire_mqtt_message(hass, "vacuum/state", message)
+    await hass.async_block_till_done()
+    state = hass.states.get("vacuum.test")
+    assert state.state == VacuumActivity.IDLE
+    assert (
+        state.attributes.get(ATTR_SUPPORTED_FEATURES)
+        & vacuum.VacuumEntityFeature.CLEAN_AREA
+    )
+
+    config_without_clean_segments_command_topic = (
+        config_with_clean_segments_command_topic.copy()
+    )
+    config_without_clean_segments_command_topic.pop(
+        mqttvacuum.CONF_CLEAN_SEGMENTS_COMMAND_TOPIC
+    )
+    async_fire_mqtt_message(
+        hass,
+        "homeassistant/vacuum/bla/config",
+        json.dumps(config_without_clean_segments_command_topic),
+    )
+    await hass.async_block_till_done()
+    message = """{
+        "battery_level": 30,
+        "state": "cleaning",
+        "segments":{
+            "1":"Livingroom",
+            "2":"Kitchen"
+        }
+    }"""
+    async_fire_mqtt_message(hass, "vacuum/state", message)
+    await hass.async_block_till_done()
+    state = hass.states.get("vacuum.test")
+    assert state.state == VacuumActivity.CLEANING
+    assert not (
+        state.attributes.get(ATTR_SUPPORTED_FEATURES)
+        & vacuum.VacuumEntityFeature.CLEAN_AREA
+    )
+
+
 @pytest.mark.parametrize("hass_config", [CONFIG_ALL_SERVICES])
 async def test_status(
     hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator