diff --git a/homeassistant/data_entry_flow.py b/homeassistant/data_entry_flow.py
index 36c68008a8e..d08e76edbd2 100644
--- a/homeassistant/data_entry_flow.py
+++ b/homeassistant/data_entry_flow.py
@@ -104,6 +104,23 @@ class UnknownStep(FlowError):
     """Unknown step specified."""
 
 
+# ignore misc is required as vol.Invalid is not typed
+# mypy error: Class cannot subclass "Invalid" (has type "Any")
+class InvalidData(vol.Invalid):  # type: ignore[misc]
+    """Invalid data provided."""
+
+    def __init__(
+        self,
+        message: str,
+        path: list[str | vol.Marker] | None,
+        error_message: str | None,
+        schema_errors: dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(message, path, error_message, **kwargs)
+        self.schema_errors = schema_errors
+
+
 class AbortFlow(FlowError):
     """Exception to indicate a flow needs to be aborted."""
 
@@ -165,6 +182,29 @@ def _async_flow_handler_to_flow_result(
     return results
 
 
+def _map_error_to_schema_errors(
+    schema_errors: dict[str, Any],
+    error: vol.Invalid,
+    data_schema: vol.Schema,
+) -> None:
+    """Map an error to the correct position in the schema_errors.
+
+    Raises ValueError if the error path could not be found in the schema.
+    Limitation: Nested schemas are not supported and a ValueError will be raised.
+    """
+    schema = data_schema.schema
+    error_path = error.path
+    if not error_path or (path_part := error_path[0]) not in schema:
+        raise ValueError("Could not find path in schema")
+
+    if len(error_path) > 1:
+        raise ValueError("Nested schemas are not supported")
+
+    # path_part can also be vol.Marker, but we need a string key
+    path_part_str = str(path_part)
+    schema_errors[path_part_str] = error.error_message
+
+
 class FlowManager(abc.ABC):
     """Manage all the flows that are in progress."""
 
@@ -334,7 +374,26 @@ class FlowManager(abc.ABC):
         if (
             data_schema := cur_step.get("data_schema")
         ) is not None and user_input is not None:
-            user_input = data_schema(user_input)
+            try:
+                user_input = data_schema(user_input)
+            except vol.Invalid as ex:
+                raised_errors = [ex]
+                if isinstance(ex, vol.MultipleInvalid):
+                    raised_errors = ex.errors
+
+                schema_errors: dict[str, Any] = {}
+                for error in raised_errors:
+                    try:
+                        _map_error_to_schema_errors(schema_errors, error, data_schema)
+                    except ValueError:
+                        # If we get here, the path in the exception does not exist in the schema.
+                        schema_errors.setdefault("base", []).append(str(error))
+                raise InvalidData(
+                    "Schema validation failed",
+                    path=ex.path,
+                    error_message=ex.error_message,
+                    schema_errors=schema_errors,
+                ) from ex
 
         # Handle a menu navigation choice
         if cur_step["type"] == FlowResultType.MENU and user_input:
diff --git a/homeassistant/helpers/data_entry_flow.py b/homeassistant/helpers/data_entry_flow.py
index aa4ef36b251..9fdd48b59f0 100644
--- a/homeassistant/helpers/data_entry_flow.py
+++ b/homeassistant/helpers/data_entry_flow.py
@@ -110,10 +110,8 @@ class FlowManagerResourceView(_BaseFlowManagerView):
             result = await self._flow_mgr.async_configure(flow_id, data)
         except data_entry_flow.UnknownFlow:
             return self.json_message("Invalid flow specified", HTTPStatus.NOT_FOUND)
-        except vol.Invalid as ex:
-            return self.json_message(
-                f"User input malformed: {ex}", HTTPStatus.BAD_REQUEST
-            )
+        except data_entry_flow.InvalidData as ex:
+            return self.json({"errors": ex.schema_errors}, HTTPStatus.BAD_REQUEST)
 
         result = self._prepare_result_json(result)
 
diff --git a/tests/components/config/test_config_entries.py b/tests/components/config/test_config_entries.py
index 414f4eb39f2..84afee245a6 100644
--- a/tests/components/config/test_config_entries.py
+++ b/tests/components/config/test_config_entries.py
@@ -9,6 +9,7 @@ import voluptuous as vol
 from homeassistant import config_entries as core_ce, data_entry_flow
 from homeassistant.components.config import config_entries
 from homeassistant.config_entries import HANDLERS, ConfigFlow
+from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_RADIUS
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.generated import config_flows
 from homeassistant.helpers import config_entry_flow, config_validation as cv
@@ -1019,12 +1020,7 @@ async def test_options_flow_with_invalid_data(hass: HomeAssistant, client) -> No
         )
         assert resp.status == HTTPStatus.BAD_REQUEST
         data = await resp.json()
-        assert data == {
-            "message": (
-                "User input malformed: invalid is not a valid option for "
-                "dictionary value @ data['choices']"
-            )
-        }
+        assert data == {"errors": {"choices": "invalid is not a valid option"}}
 
 
 async def test_get_single(
@@ -2027,3 +2023,89 @@ async def test_subscribe_entries_ws_filtered(
             "type": "added",
         }
     ]
+
+
+async def test_flow_with_multiple_schema_errors(hass: HomeAssistant, client) -> None:
+    """Test an config flow with multiple schema errors."""
+    mock_integration(
+        hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True))
+    )
+    mock_platform(hass, "test.config_flow", None)
+
+    class TestFlow(core_ce.ConfigFlow):
+        async def async_step_user(self, user_input=None):
+            return self.async_show_form(
+                step_id="user",
+                data_schema=vol.Schema(
+                    {
+                        vol.Required(CONF_LATITUDE): cv.latitude,
+                        vol.Required(CONF_LONGITUDE): cv.longitude,
+                        vol.Required(CONF_RADIUS): vol.All(int, vol.Range(min=5)),
+                    }
+                ),
+            )
+
+    with patch.dict(HANDLERS, {"test": TestFlow}):
+        resp = await client.post(
+            "/api/config/config_entries/flow", json={"handler": "test"}
+        )
+        assert resp.status == HTTPStatus.OK
+        flow_id = (await resp.json())["flow_id"]
+
+        resp = await client.post(
+            f"/api/config/config_entries/flow/{flow_id}",
+            json={"latitude": 30000, "longitude": 30000, "radius": 1},
+        )
+        assert resp.status == HTTPStatus.BAD_REQUEST
+        data = await resp.json()
+        assert data == {
+            "errors": {
+                "latitude": "invalid latitude",
+                "longitude": "invalid longitude",
+                "radius": "value must be at least 5",
+            }
+        }
+
+
+async def test_flow_with_multiple_schema_errors_base(
+    hass: HomeAssistant, client
+) -> None:
+    """Test an config flow with multiple schema errors where fields are not in the schema."""
+    mock_integration(
+        hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True))
+    )
+    mock_platform(hass, "test.config_flow", None)
+
+    class TestFlow(core_ce.ConfigFlow):
+        async def async_step_user(self, user_input=None):
+            return self.async_show_form(
+                step_id="user",
+                data_schema=vol.Schema(
+                    {
+                        vol.Required(CONF_LATITUDE): cv.latitude,
+                    }
+                ),
+            )
+
+    with patch.dict(HANDLERS, {"test": TestFlow}):
+        resp = await client.post(
+            "/api/config/config_entries/flow", json={"handler": "test"}
+        )
+        assert resp.status == HTTPStatus.OK
+        flow_id = (await resp.json())["flow_id"]
+
+        resp = await client.post(
+            f"/api/config/config_entries/flow/{flow_id}",
+            json={"invalid": 30000, "invalid_2": 30000},
+        )
+        assert resp.status == HTTPStatus.BAD_REQUEST
+        data = await resp.json()
+        assert data == {
+            "errors": {
+                "base": [
+                    "extra keys not allowed @ data['invalid']",
+                    "extra keys not allowed @ data['invalid_2']",
+                ],
+                "latitude": "required key not provided",
+            }
+        }
diff --git a/tests/components/eafm/test_config_flow.py b/tests/components/eafm/test_config_flow.py
index 5addc80e5ae..208f406d8b9 100644
--- a/tests/components/eafm/test_config_flow.py
+++ b/tests/components/eafm/test_config_flow.py
@@ -2,7 +2,7 @@
 from unittest.mock import patch
 
 import pytest
-from voluptuous.error import MultipleInvalid
+from voluptuous.error import Invalid
 
 from homeassistant import config_entries
 from homeassistant.components.eafm import const
@@ -32,7 +32,7 @@ async def test_flow_invalid_station(hass: HomeAssistant, mock_get_stations) -> N
     )
     assert result["type"] == "form"
 
-    with pytest.raises(MultipleInvalid):
+    with pytest.raises(Invalid):
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"], user_input={"station": "My other station"}
         )
diff --git a/tests/components/homekit/test_config_flow.py b/tests/components/homekit/test_config_flow.py
index b925fcb341c..838f72be3c6 100644
--- a/tests/components/homekit/test_config_flow.py
+++ b/tests/components/homekit/test_config_flow.py
@@ -1444,7 +1444,7 @@ async def test_options_flow_exclude_mode_skips_category_entities(
 
     # sonos_config_switch.entity_id is a config category entity
     # so it should not be selectable since it will always be excluded
-    with pytest.raises(vol.error.MultipleInvalid):
+    with pytest.raises(vol.error.Invalid):
         await hass.config_entries.options.async_configure(
             result2["flow_id"],
             user_input={"entities": [sonos_config_switch.entity_id]},
@@ -1539,7 +1539,7 @@ async def test_options_flow_exclude_mode_skips_hidden_entities(
 
     # sonos_hidden_switch.entity_id is a hidden entity
     # so it should not be selectable since it will always be excluded
-    with pytest.raises(vol.error.MultipleInvalid):
+    with pytest.raises(vol.error.Invalid):
         await hass.config_entries.options.async_configure(
             result2["flow_id"],
             user_input={"entities": [sonos_hidden_switch.entity_id]},
diff --git a/tests/components/imap/test_config_flow.py b/tests/components/imap/test_config_flow.py
index 8c91797ae92..0561085823d 100644
--- a/tests/components/imap/test_config_flow.py
+++ b/tests/components/imap/test_config_flow.py
@@ -465,7 +465,7 @@ async def test_advanced_options_form(
                 # Check if entry was updated
                 for key, value in new_config.items():
                     assert entry.data[key] == value
-    except vol.MultipleInvalid:
+    except vol.Invalid:
         # Check if form was expected with these options
         assert assert_result == data_entry_flow.FlowResultType.FORM
 
diff --git a/tests/components/melnor/test_config_flow.py b/tests/components/melnor/test_config_flow.py
index 95a67644606..bb0a017611f 100644
--- a/tests/components/melnor/test_config_flow.py
+++ b/tests/components/melnor/test_config_flow.py
@@ -48,7 +48,7 @@ async def test_user_step_discovered_devices(
         assert result["type"] == FlowResultType.FORM
         assert result["step_id"] == "pick_device"
 
-        with pytest.raises(vol.MultipleInvalid):
+        with pytest.raises(vol.Invalid):
             await hass.config_entries.flow.async_configure(
                 result["flow_id"], user_input={CONF_ADDRESS: "wrong_address"}
             )
@@ -95,7 +95,7 @@ async def test_user_step_with_existing_device(
 
         assert result["type"] == FlowResultType.FORM
 
-        with pytest.raises(vol.MultipleInvalid):
+        with pytest.raises(vol.Invalid):
             await hass.config_entries.flow.async_configure(
                 result["flow_id"], user_input={CONF_ADDRESS: FAKE_ADDRESS_1}
             )
diff --git a/tests/components/mqtt/test_config_flow.py b/tests/components/mqtt/test_config_flow.py
index c2a7e0065ce..9e86a3554d6 100644
--- a/tests/components/mqtt/test_config_flow.py
+++ b/tests/components/mqtt/test_config_flow.py
@@ -672,7 +672,7 @@ async def test_keepalive_validation(
     assert result["step_id"] == "broker"
 
     if error:
-        with pytest.raises(vol.MultipleInvalid):
+        with pytest.raises(vol.Invalid):
             result = await hass.config_entries.options.async_configure(
                 result["flow_id"],
                 user_input=test_input,
diff --git a/tests/components/peco/test_config_flow.py b/tests/components/peco/test_config_flow.py
index 9ce87d707ff..833c66ab37a 100644
--- a/tests/components/peco/test_config_flow.py
+++ b/tests/components/peco/test_config_flow.py
@@ -3,7 +3,7 @@ from unittest.mock import patch
 
 from peco import HttpError, IncompatibleMeterError, UnresponsiveMeterError
 import pytest
-from voluptuous.error import MultipleInvalid
+from voluptuous.error import Invalid
 
 from homeassistant import config_entries
 from homeassistant.components.peco.const import DOMAIN
@@ -51,7 +51,7 @@ async def test_invalid_county(hass: HomeAssistant) -> None:
     with patch(
         "homeassistant.components.peco.async_setup_entry",
         return_value=True,
-    ), pytest.raises(MultipleInvalid):
+    ), pytest.raises(Invalid):
         await hass.config_entries.flow.async_configure(
             result["flow_id"],
             {
diff --git a/tests/components/risco/test_config_flow.py b/tests/components/risco/test_config_flow.py
index 8207ad819b7..cc6cefc1325 100644
--- a/tests/components/risco/test_config_flow.py
+++ b/tests/components/risco/test_config_flow.py
@@ -383,14 +383,14 @@ async def test_ha_to_risco_schema(hass: HomeAssistant) -> None:
     )
 
     # Test an HA state that isn't used
-    with pytest.raises(vol.error.MultipleInvalid):
+    with pytest.raises(vol.error.Invalid):
         await hass.config_entries.options.async_configure(
             result["flow_id"],
             user_input={**TEST_HA_TO_RISCO, "armed_custom_bypass": "D"},
         )
 
     # Test a combo that can't be selected
-    with pytest.raises(vol.error.MultipleInvalid):
+    with pytest.raises(vol.error.Invalid):
         await hass.config_entries.options.async_configure(
             result["flow_id"],
             user_input={**TEST_HA_TO_RISCO, "armed_night": "A"},