diff --git a/homeassistant/components/cloud/__init__.py b/homeassistant/components/cloud/__init__.py
index d7d57835e3ade1..6e5cddd0f28764 100644
--- a/homeassistant/components/cloud/__init__.py
+++ b/homeassistant/components/cloud/__init__.py
@@ -5,6 +5,7 @@
 from collections.abc import Awaitable, Callable
 from datetime import datetime, timedelta
 from enum import Enum
+from typing import cast
 
 from hass_nabucasa import Cloud
 import voluptuous as vol
@@ -176,6 +177,22 @@ def async_active_subscription(hass: HomeAssistant) -> bool:
     return async_is_logged_in(hass) and not hass.data[DOMAIN].subscription_expired
 
 
+async def async_get_or_create_cloudhook(hass: HomeAssistant, webhook_id: str) -> str:
+    """Get or create a cloudhook."""
+    if not async_is_connected(hass):
+        raise CloudNotConnected
+
+    if not async_is_logged_in(hass):
+        raise CloudNotAvailable
+
+    cloud: Cloud[CloudClient] = hass.data[DOMAIN]
+    cloudhooks = cloud.client.cloudhooks
+    if hook := cloudhooks.get(webhook_id):
+        return cast(str, hook["cloudhook_url"])
+
+    return await async_create_cloudhook(hass, webhook_id)
+
+
 @bind_hass
 async def async_create_cloudhook(hass: HomeAssistant, webhook_id: str) -> str:
     """Create a cloudhook."""
diff --git a/homeassistant/components/mobile_app/__init__.py b/homeassistant/components/mobile_app/__init__.py
index cb5c0ae5c3ddd2..124ef750baa128 100644
--- a/homeassistant/components/mobile_app/__init__.py
+++ b/homeassistant/components/mobile_app/__init__.py
@@ -36,6 +36,7 @@
 )
 from .helpers import savable_state
 from .http_api import RegistrationsView
+from .util import async_create_cloud_hook
 from .webhook import handle_webhook
 
 PLATFORMS = [Platform.SENSOR, Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER]
@@ -103,26 +104,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     registration_name = f"Mobile App: {registration[ATTR_DEVICE_NAME]}"
     webhook_register(hass, DOMAIN, registration_name, webhook_id, handle_webhook)
 
-    async def create_cloud_hook() -> None:
-        """Create a cloud hook."""
-        hook = await cloud.async_create_cloudhook(hass, webhook_id)
-        hass.config_entries.async_update_entry(
-            entry, data={**entry.data, CONF_CLOUDHOOK_URL: hook}
-        )
-
     async def manage_cloudhook(state: cloud.CloudConnectionState) -> None:
         if (
             state is cloud.CloudConnectionState.CLOUD_CONNECTED
             and CONF_CLOUDHOOK_URL not in entry.data
         ):
-            await create_cloud_hook()
+            await async_create_cloud_hook(hass, webhook_id, entry)
 
     if (
         CONF_CLOUDHOOK_URL not in entry.data
         and cloud.async_active_subscription(hass)
         and cloud.async_is_connected(hass)
     ):
-        await create_cloud_hook()
+        await async_create_cloud_hook(hass, webhook_id, entry)
+
     entry.async_on_unload(cloud.async_listen_connection_change(hass, manage_cloudhook))
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
diff --git a/homeassistant/components/mobile_app/http_api.py b/homeassistant/components/mobile_app/http_api.py
index 3c34a291df15e9..92bb473d51acdb 100644
--- a/homeassistant/components/mobile_app/http_api.py
+++ b/homeassistant/components/mobile_app/http_api.py
@@ -35,6 +35,7 @@
     SCHEMA_APP_DATA,
 )
 from .helpers import supports_encryption
+from .util import async_create_cloud_hook
 
 
 class RegistrationsView(HomeAssistantView):
@@ -69,8 +70,8 @@ async def post(self, request: Request, data: dict) -> Response:
         webhook_id = secrets.token_hex()
 
         if cloud.async_active_subscription(hass):
-            data[CONF_CLOUDHOOK_URL] = await cloud.async_create_cloudhook(
-                hass, webhook_id
+            data[CONF_CLOUDHOOK_URL] = await async_create_cloud_hook(
+                hass, webhook_id, None
             )
 
         data[CONF_WEBHOOK_ID] = webhook_id
diff --git a/homeassistant/components/mobile_app/util.py b/homeassistant/components/mobile_app/util.py
index 45641861e5cd61..a7871d935edf37 100644
--- a/homeassistant/components/mobile_app/util.py
+++ b/homeassistant/components/mobile_app/util.py
@@ -1,8 +1,11 @@
 """Mobile app utility functions."""
 from __future__ import annotations
 
+import asyncio
 from typing import TYPE_CHECKING
 
+from homeassistant.components import cloud
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant, callback
 
 from .const import (
@@ -10,6 +13,7 @@
     ATTR_PUSH_TOKEN,
     ATTR_PUSH_URL,
     ATTR_PUSH_WEBSOCKET_CHANNEL,
+    CONF_CLOUDHOOK_URL,
     DATA_CONFIG_ENTRIES,
     DATA_DEVICES,
     DATA_NOTIFY,
@@ -53,3 +57,19 @@ def get_notify_service(hass: HomeAssistant, webhook_id: str) -> str | None:
             return target_service
 
     return None
+
+
+_CLOUD_HOOK_LOCK = asyncio.Lock()
+
+
+async def async_create_cloud_hook(
+    hass: HomeAssistant, webhook_id: str, entry: ConfigEntry | None
+) -> str:
+    """Create a cloud hook."""
+    async with _CLOUD_HOOK_LOCK:
+        hook = await cloud.async_get_or_create_cloudhook(hass, webhook_id)
+        if entry:
+            hass.config_entries.async_update_entry(
+                entry, data={**entry.data, CONF_CLOUDHOOK_URL: hook}
+            )
+        return hook
diff --git a/tests/components/cloud/conftest.py b/tests/components/cloud/conftest.py
index ef8cb037cdbf72..42852b15206110 100644
--- a/tests/components/cloud/conftest.py
+++ b/tests/components/cloud/conftest.py
@@ -109,6 +109,7 @@ def mock_is_connected() -> bool:
 
         is_connected = PropertyMock(side_effect=mock_is_connected)
         type(mock_cloud).is_connected = is_connected
+        type(mock_cloud.iot).connected = is_connected
 
         # Properties that we mock as attributes.
         mock_cloud.expiration_date = utcnow()
diff --git a/tests/components/cloud/test_init.py b/tests/components/cloud/test_init.py
index e12775d5a4a5f4..850f8e12e02085 100644
--- a/tests/components/cloud/test_init.py
+++ b/tests/components/cloud/test_init.py
@@ -1,12 +1,18 @@
 """Test the cloud component."""
+from collections.abc import Callable, Coroutine
 from typing import Any
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from hass_nabucasa import Cloud
 import pytest
 
 from homeassistant.components import cloud
-from homeassistant.components.cloud.const import DOMAIN
+from homeassistant.components.cloud import (
+    CloudNotAvailable,
+    CloudNotConnected,
+    async_get_or_create_cloudhook,
+)
+from homeassistant.components.cloud.const import DOMAIN, PREF_CLOUDHOOKS
 from homeassistant.components.cloud.prefs import STORAGE_KEY
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import Context, HomeAssistant
@@ -214,3 +220,57 @@ async def test_remote_ui_url(hass: HomeAssistant, mock_cloud_fixture) -> None:
         cl.client.prefs._prefs["remote_domain"] = "example.com"
 
         assert cloud.async_remote_ui_url(hass) == "https://example.com"
+
+
+async def test_async_get_or_create_cloudhook(
+    hass: HomeAssistant,
+    cloud: MagicMock,
+    set_cloud_prefs: Callable[[dict[str, Any]], Coroutine[Any, Any, None]],
+) -> None:
+    """Test async_get_or_create_cloudhook."""
+    assert await async_setup_component(hass, "cloud", {"cloud": {}})
+    await hass.async_block_till_done()
+
+    webhook_id = "mock-webhook-id"
+    cloudhook_url = "https://cloudhook.nabu.casa/abcdefg"
+
+    with patch(
+        "homeassistant.components.cloud.async_create_cloudhook",
+        return_value=cloudhook_url,
+    ) as async_create_cloudhook_mock:
+        # create cloudhook as it does not exist
+        assert (await async_get_or_create_cloudhook(hass, webhook_id)) == cloudhook_url
+        async_create_cloudhook_mock.assert_called_once_with(hass, webhook_id)
+
+        await set_cloud_prefs(
+            {
+                PREF_CLOUDHOOKS: {
+                    webhook_id: {
+                        "webhook_id": webhook_id,
+                        "cloudhook_id": "random-id",
+                        "cloudhook_url": cloudhook_url,
+                        "managed": True,
+                    }
+                }
+            }
+        )
+
+        async_create_cloudhook_mock.reset_mock()
+
+        # get cloudhook as it exists
+        assert await async_get_or_create_cloudhook(hass, webhook_id) == cloudhook_url
+        async_create_cloudhook_mock.assert_not_called()
+
+    # Simulate logged out
+    cloud.id_token = None
+
+    # Not logged in
+    with pytest.raises(CloudNotAvailable):
+        await async_get_or_create_cloudhook(hass, webhook_id)
+
+    # Simulate disconnected
+    cloud.iot.state = "disconnected"
+
+    # Not connected
+    with pytest.raises(CloudNotConnected):
+        await async_get_or_create_cloudhook(hass, webhook_id)
diff --git a/tests/components/mobile_app/test_init.py b/tests/components/mobile_app/test_init.py
index d504703c222339..6a365e84fb0f1f 100644
--- a/tests/components/mobile_app/test_init.py
+++ b/tests/components/mobile_app/test_init.py
@@ -88,15 +88,17 @@ async def _test_create_cloud_hook(
     ), patch(
         "homeassistant.components.cloud.async_is_connected", return_value=True
     ), patch(
-        "homeassistant.components.cloud.async_create_cloudhook", autospec=True
-    ) as mock_create_cloudhook:
+        "homeassistant.components.cloud.async_get_or_create_cloudhook", autospec=True
+    ) as mock_async_get_or_create_cloudhook:
         cloud_hook = "https://hook-url"
-        mock_create_cloudhook.return_value = cloud_hook
+        mock_async_get_or_create_cloudhook.return_value = cloud_hook
 
         assert await hass.config_entries.async_setup(config_entry.entry_id)
         await hass.async_block_till_done()
         assert config_entry.state is ConfigEntryState.LOADED
-        await additional_steps(config_entry, mock_create_cloudhook, cloud_hook)
+        await additional_steps(
+            config_entry, mock_async_get_or_create_cloudhook, cloud_hook
+        )
 
 
 async def test_create_cloud_hook_on_setup(