diff --git a/homeassistant/components/google_assistant/__init__.py b/homeassistant/components/google_assistant/__init__.py
index 644179858a36c5..feb5a138c08565 100644
--- a/homeassistant/components/google_assistant/__init__.py
+++ b/homeassistant/components/google_assistant/__init__.py
@@ -8,7 +8,11 @@
 from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
 from homeassistant.const import CONF_API_KEY, CONF_NAME, Platform
 from homeassistant.core import HomeAssistant, ServiceCall
-from homeassistant.helpers import config_validation as cv, device_registry as dr
+from homeassistant.helpers import (
+    config_validation as cv,
+    device_registry as dr,
+    discovery,
+)
 from homeassistant.helpers.typing import ConfigType
 
 from .const import (  # noqa: F401
@@ -32,6 +36,7 @@
     SOURCE_CLOUD,
 )
 from .const import EVENT_QUERY_RECEIVED  # noqa: F401
+from .helpers import async_send_text_command
 from .http import GoogleAssistantView, GoogleConfig
 
 from .const import EVENT_COMMAND_RECEIVED, EVENT_SYNC_RECEIVED  # noqa: F401, isort:skip
@@ -94,6 +99,16 @@ def _check_report_state(data):
     {vol.Optional(DOMAIN): GOOGLE_ASSISTANT_SCHEMA}, extra=vol.ALLOW_EXTRA
 )
 
+SERVICE_SEND_TEXT_COMMAND = "send_text_command"
+SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND = "command"
+SERVICE_SEND_TEXT_COMMAND_SCHEMA = vol.All(
+    {
+        vol.Required(SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND): vol.All(
+            str, vol.Length(min=1)
+        ),
+    },
+)
+
 
 async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool:
     """Activate Google Actions component."""
@@ -159,11 +174,33 @@ async def request_sync_service_handler(call: ServiceCall) -> None:
 
         await google_config.async_sync_entities(agent_user_id)
 
-    # Register service only if key is provided
+    async def send_text_command_service_handler(call: ServiceCall) -> None:
+        """Handle request send text command calls."""
+        command: str = call.data[SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND]
+        await async_send_text_command(config[CONF_SERVICE_ACCOUNT], command)
+
+    # Register services only if key is provided
     if CONF_SERVICE_ACCOUNT in config:
         hass.services.async_register(
             DOMAIN, SERVICE_REQUEST_SYNC, request_sync_service_handler
         )
+        hass.services.async_register(
+            DOMAIN,
+            SERVICE_SEND_TEXT_COMMAND,
+            send_text_command_service_handler,
+            schema=SERVICE_SEND_TEXT_COMMAND_SCHEMA,
+        )
+        # set up notify platform, no entry support for notify platform yet,
+        # have to use discovery to load platform.
+        hass.async_create_task(
+            discovery.async_load_platform(
+                hass,
+                Platform.NOTIFY,
+                DOMAIN,
+                {CONF_NAME: DOMAIN},
+                hass.data[DOMAIN][DATA_CONFIG],
+            )
+        )
 
     hass.config_entries.async_setup_platforms(entry, PLATFORMS)
 
diff --git a/homeassistant/components/google_assistant/helpers.py b/homeassistant/components/google_assistant/helpers.py
index 3351af94648b11..dce8fcab18228f 100644
--- a/homeassistant/components/google_assistant/helpers.py
+++ b/homeassistant/components/google_assistant/helpers.py
@@ -8,9 +8,12 @@
 from http import HTTPStatus
 import logging
 import pprint
+from typing import Any
 
 from aiohttp.web import json_response
 from awesomeversion import AwesomeVersion
+from gassist_text import TextAssistant
+from google.oauth2 import service_account
 from yarl import URL
 
 from homeassistant.components import webhook
@@ -45,6 +48,7 @@
 
 SYNC_DELAY = 15
 _LOGGER = logging.getLogger(__name__)
+ASSISTANT_SDK_SCOPE = "https://www.googleapis.com/auth/assistant-sdk-prototype"
 LOCAL_SDK_VERSION_HEADER = "HA-Cloud-Version"
 LOCAL_SDK_MIN_VERSION = AwesomeVersion("2.1.5")
 
@@ -725,3 +729,15 @@ def async_get_entities(
             entities.append(entity)
 
     return entities
+
+
+async def async_send_text_command(service_account_info: dict[str, Any], command: str):
+    """Send a command as a text query to Google Assistant."""
+    # Add token_uri if missing. Credentials requires it but config doesn't
+    if "token_uri" not in service_account_info:
+        service_account_info["token_uri"] = "https://oauth2.googleapis.com/token"
+    credentials = service_account.Credentials.from_service_account_info(
+        service_account_info, scopes=[ASSISTANT_SDK_SCOPE]
+    )
+    with TextAssistant(credentials) as assistant:
+        assistant.assist(command)
diff --git a/homeassistant/components/google_assistant/manifest.json b/homeassistant/components/google_assistant/manifest.json
index fcd7c983937eb0..348099245ba393 100644
--- a/homeassistant/components/google_assistant/manifest.json
+++ b/homeassistant/components/google_assistant/manifest.json
@@ -2,6 +2,7 @@
   "domain": "google_assistant",
   "name": "Google Assistant",
   "documentation": "https://www.home-assistant.io/integrations/google_assistant",
+  "requirements": ["gassist-text==0.0.3"],
   "dependencies": ["http"],
   "after_dependencies": ["camera"],
   "codeowners": ["@home-assistant/cloud"],
diff --git a/homeassistant/components/google_assistant/notify.py b/homeassistant/components/google_assistant/notify.py
new file mode 100644
index 00000000000000..82e15da0835c44
--- /dev/null
+++ b/homeassistant/components/google_assistant/notify.py
@@ -0,0 +1,42 @@
+"""Support for Google Assistant broadcast notifications."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from .const import CONF_SERVICE_ACCOUNT, DATA_CONFIG, DOMAIN
+from .helpers import async_send_text_command
+
+
+async def async_get_service(
+    hass: HomeAssistant,
+    config: ConfigType,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> BaseNotificationService:
+    """Get the broadcast notification service."""
+    return BroadcastNotificationService(hass)
+
+
+class BroadcastNotificationService(BaseNotificationService):
+    """Implement broadcast notification service."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the service."""
+        self.hass = hass
+
+    async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
+        """Send a message."""
+        if not message:
+            return
+        service_account_info = self.hass.data[DOMAIN][DATA_CONFIG][CONF_SERVICE_ACCOUNT]
+        targets = kwargs.get(ATTR_TARGET)
+        if not targets:
+            command = f"broadcast {message}"
+            await async_send_text_command(service_account_info, command)
+        else:
+            for target in targets:
+                command = f"broadcast to {target} {message}"
+                await async_send_text_command(service_account_info, command)
diff --git a/homeassistant/components/google_assistant/services.yaml b/homeassistant/components/google_assistant/services.yaml
index fe5ef51c2ce9ca..888a18385c8574 100644
--- a/homeassistant/components/google_assistant/services.yaml
+++ b/homeassistant/components/google_assistant/services.yaml
@@ -7,3 +7,13 @@ request_sync:
       description: "Only needed for automations. Specific Home Assistant user id (not username, ID in configuration > users > under username) to sync with Google Assistant. Do not need when you call this service through Home Assistant front end or API. Used in automation script or other place where context.user_id is missing."
       selector:
         text:
+send_text_command:
+  name: Send text command
+  description: Send a command as a text query to Google Assistant.
+  fields:
+    command:
+      name: Command
+      description: Command to send to Google Assistant.
+      example: turn off kitchen TV
+      selector:
+        text:
diff --git a/requirements_all.txt b/requirements_all.txt
index af6ec2135468ff..ef20e8b535e7a2 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -739,6 +739,9 @@ gTTS==2.2.4
 # homeassistant.components.garages_amsterdam
 garages-amsterdam==3.0.0
 
+# homeassistant.components.google_assistant
+gassist-text==0.0.3
+
 # homeassistant.components.google
 gcal-sync==4.0.2
 
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 9496175043d453..a89f468acfd535 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -555,6 +555,9 @@ gTTS==2.2.4
 # homeassistant.components.garages_amsterdam
 garages-amsterdam==3.0.0
 
+# homeassistant.components.google_assistant
+gassist-text==0.0.3
+
 # homeassistant.components.google
 gcal-sync==4.0.2
 
diff --git a/tests/components/google_assistant/test_init.py b/tests/components/google_assistant/test_init.py
index bdd6932c91d8b2..e56aa048387170 100644
--- a/tests/components/google_assistant/test_init.py
+++ b/tests/components/google_assistant/test_init.py
@@ -1,5 +1,6 @@
 """The tests for google-assistant init."""
 from http import HTTPStatus
+from unittest.mock import patch
 
 from homeassistant.components import google_assistant as ga
 from homeassistant.core import Context, HomeAssistant
@@ -69,3 +70,21 @@ async def test_request_sync_service(aioclient_mock, hass):
     )
 
     assert aioclient_mock.call_count == 2  # token + request
+
+
+async def test_send_text_command_service(hass):
+    """Test send_text_command calls TextAssistant."""
+    await async_setup_component(hass, ga.DOMAIN, {ga.DOMAIN: DUMMY_CONFIG})
+    await hass.async_block_till_done()
+
+    command = "turn on home assistant unsupported device"
+    with patch(
+        "homeassistant.components.google_assistant.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            ga.DOMAIN,
+            ga.SERVICE_SEND_TEXT_COMMAND,
+            {ga.SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND: command},
+            blocking=True,
+        )
+    mock_assist_call.assert_called_once_with(command)
diff --git a/tests/components/google_assistant/test_notify.py b/tests/components/google_assistant/test_notify.py
new file mode 100644
index 00000000000000..20a0ab0f844233
--- /dev/null
+++ b/tests/components/google_assistant/test_notify.py
@@ -0,0 +1,99 @@
+"""Tests for the Google Assistant notify."""
+from unittest.mock import call, patch
+
+from homeassistant.components import google_assistant as ga, notify
+from homeassistant.components.google_assistant import GOOGLE_ASSISTANT_SCHEMA
+from homeassistant.setup import async_setup_component
+
+from .test_http import DUMMY_CONFIG
+
+
+async def setup_notify(hass):
+    """Test setup."""
+    await async_setup_component(hass, ga.DOMAIN, {ga.DOMAIN: DUMMY_CONFIG})
+    await hass.async_block_till_done()
+    assert hass.services.has_service(notify.DOMAIN, ga.DOMAIN)
+
+
+async def test_broadcast_no_targets(hass):
+    """Test broadcast to all."""
+    await setup_notify(hass)
+
+    message = "time for dinner"
+    expected_command = "broadcast time for dinner"
+    with patch(
+        "homeassistant.components.google_assistant.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN, ga.DOMAIN, {notify.ATTR_MESSAGE: message}
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_called_once_with(expected_command)
+
+
+async def test_broadcast_one_target(hass):
+    """Test broadcast to one target."""
+    await setup_notify(hass)
+
+    message = "time for dinner"
+    target = "basement"
+    expected_command = "broadcast to basement time for dinner"
+    with patch(
+        "homeassistant.components.google_assistant.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            ga.DOMAIN,
+            {notify.ATTR_MESSAGE: message, notify.ATTR_TARGET: [target]},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_called_once_with(expected_command)
+
+
+async def test_broadcast_two_targets(hass):
+    """Test broadcast to two targets."""
+    await setup_notify(hass)
+
+    message = "time for dinner"
+    target1 = "basement"
+    target2 = "master bedroom"
+    expected_command1 = "broadcast to basement time for dinner"
+    expected_command2 = "broadcast to master bedroom time for dinner"
+    with patch(
+        "homeassistant.components.google_assistant.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            ga.DOMAIN,
+            {notify.ATTR_MESSAGE: message, notify.ATTR_TARGET: [target1, target2]},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_has_calls(
+        [call(expected_command1), call(expected_command2)]
+    )
+
+
+async def test_broadcast_empty_message(hass):
+    """Test broadcast empty message."""
+    await setup_notify(hass)
+
+    message = ""
+    with patch(
+        "homeassistant.components.google_assistant.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            ga.DOMAIN,
+            {notify.ATTR_MESSAGE: message},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_not_called()
+
+
+async def test_no_broadcast_service(hass):
+    """Test no broadcast service when there is no service_account in config."""
+    await async_setup_component(
+        hass, ga.DOMAIN, {ga.DOMAIN: GOOGLE_ASSISTANT_SCHEMA({"project_id": "1234"})}
+    )
+    await hass.async_block_till_done()
+    assert not hass.services.has_service(notify.DOMAIN, ga.DOMAIN)