diff --git a/subiquity/cloudinit.py b/subiquity/cloudinit.py
index 1d838d985..b9f2cc9af 100644
--- a/subiquity/cloudinit.py
+++ b/subiquity/cloudinit.py
@@ -87,6 +87,10 @@ def supports_recoverable_errors() -> bool:
     return cloud_init_version() >= "23.4"
 
 
+def supports_schema_subcommand() -> bool:
+    return cloud_init_version() >= "22.2"
+
+
 def read_json_extended_status(stream):
     try:
         status = json.loads(stream)
@@ -167,6 +171,13 @@ async def validate_cloud_init_top_level_keys() -> None:
     :raises CloudInitSchemaTopLevelKeyError: If cloud-init schema did not
             validate successfully.
     """
+    if not supports_schema_subcommand():
+        log.debug(
+            "Host cloud-config doesn't support 'schema' subcommand. "
+            "Skipping top-level key cloud-config validation."
+        )
+        return None
+
     causes: list[str] = await get_unknown_keys()
 
     if causes:
diff --git a/subiquity/tests/test_cloudinit.py b/subiquity/tests/test_cloudinit.py
index 2fdeb852e..a92b77b65 100644
--- a/subiquity/tests/test_cloudinit.py
+++ b/subiquity/tests/test_cloudinit.py
@@ -226,6 +226,36 @@ async def test_get_schema_warn_on_timeout(self, log_mock, wait_for_mock):
         log_mock.warning.assert_called()
         self.assertEqual([], sources)
 
+    @parameterized.expand(
+        (
+            ("20.2", True),
+            ("22.1", True),
+            ("22.2", False),
+            ("23.0", False),
+        )
+    )
+    async def test_version_check_and_skip(self, version, should_skip):
+        """Test that top-level key validation skipped on right versions.
+
+        The "schema" subcommand, which the top-level key validation relies
+        on, was added in cloud-init version 22.2. This test is to ensure
+        that it's skipped on older releases.
+        """
+        with (
+            patch("subiquity.cloudinit.get_unknown_keys") as keys_mock,
+            patch("subiquity.cloudinit.cloud_init_version") as version_mock,
+        ):
+            version_mock.return_value = version
+
+            if should_skip:
+                await validate_cloud_init_top_level_keys()
+                keys_mock.assert_not_called()
+
+            else:
+                keys_mock.return_value = []  # avoid raise condition
+                await validate_cloud_init_top_level_keys()
+                keys_mock.assert_called_once()
+
 
 class TestCloudInitRandomStrings(SubiTestCase):
     def test_passwd_constraints(self):