diff --git a/homeassistant/components/google_assistant_sdk/__init__.py b/homeassistant/components/google_assistant_sdk/__init__.py
index 3b394ed571c3..31b609812338 100644
--- a/homeassistant/components/google_assistant_sdk/__init__.py
+++ b/homeassistant/components/google_assistant_sdk/__init__.py
@@ -1,9 +1,10 @@
 """Support for Google Assistant SDK."""
 
+import asyncio
 from typing import override
 
 from aiohttp import ClientError
-from gassist_text import TextAssistant
+from gassist_text import TextAssistantAsync
 from google.oauth2.credentials import Credentials
 
 from homeassistant.components import conversation
@@ -70,6 +71,7 @@ async def async_setup_entry(
         session=session, mem_storage=mem_storage
     )
     agent = GoogleAssistantConversationAgent(hass, entry)
+    entry.async_on_unload(agent.async_close)
     conversation.async_set_agent(hass, entry, agent)
 
     return True
@@ -93,9 +95,13 @@ class GoogleAssistantConversationAgent(conversation.AbstractConversationAgent):
         """Initialize the agent."""
         self.hass = hass
         self.entry = entry
-        self.assistant: TextAssistant | None = None
+        self.assistant: TextAssistantAsync | None = None
         self.session: OAuth2Session | None = None
         self.language: str | None = None
+        # The assistant holds the state of a single conversation and is shared
+        # by every request, so the requests have to be serialized. This also
+        # keeps a replacement from closing a channel that is still in use.
+        self._lock = asyncio.Lock()
 
     @property
     @override
@@ -103,34 +109,45 @@ class GoogleAssistantConversationAgent(conversation.AbstractConversationAgent):
         """Return a list of supported languages."""
         return SUPPORTED_LANGUAGE_CODES
 
+    async def async_close(self) -> None:
+        """Close the assistant, releasing its gRPC channel."""
+        async with self._lock:
+            await self._async_close_assistant()
+
+    async def _async_close_assistant(self) -> None:
+        """Close the assistant. The caller must hold the lock."""
+        if self.assistant:
+            await self.assistant.close()
+            self.assistant = None
+
     @override
     async def async_process(
         self, user_input: conversation.ConversationInput
     ) -> conversation.ConversationResult:
         """Process a sentence."""
-        if self.session:
-            session = self.session
-        else:
-            session = self.entry.runtime_data.session
-            self.session = session
-        if not session.valid_token:
-            await session.async_ensure_token_valid()
-            self.assistant = None
+        async with self._lock:
+            if self.session:
+                session = self.session
+            else:
+                session = self.entry.runtime_data.session
+                self.session = session
+            if not session.valid_token:
+                await session.async_ensure_token_valid()
+                await self._async_close_assistant()
 
-        language = best_matching_language_code(
-            self.hass,
-            user_input.language,
-            self.entry.options.get(CONF_LANGUAGE_CODE),
-        )
+            language = best_matching_language_code(
+                self.hass,
+                user_input.language,
+                self.entry.options.get(CONF_LANGUAGE_CODE),
+            )
 
-        if not self.assistant or language != self.language:
-            credentials = Credentials(session.token[CONF_ACCESS_TOKEN])  # type: ignore[no-untyped-call]
-            self.language = language
-            self.assistant = TextAssistant(credentials, self.language)
+            if not self.assistant or language != self.language:
+                await self._async_close_assistant()
+                credentials = Credentials(session.token[CONF_ACCESS_TOKEN])  # type: ignore[no-untyped-call]
+                self.language = language
+                self.assistant = TextAssistantAsync(credentials, self.language)
 
-        resp = await self.hass.async_add_executor_job(
-            self.assistant.assist, user_input.text
-        )
+            resp = await self.assistant.assist(user_input.text)
         text_response = resp[0] or "<empty response>"
 
         intent_response = intent.IntentResponse(language=language)
diff --git a/homeassistant/components/google_assistant_sdk/helpers.py b/homeassistant/components/google_assistant_sdk/helpers.py
index 9aecb871a907..d565e41138fe 100644
--- a/homeassistant/components/google_assistant_sdk/helpers.py
+++ b/homeassistant/components/google_assistant_sdk/helpers.py
@@ -7,7 +7,7 @@ from typing import Any
 import uuid
 
 from aiohttp import web
-from gassist_text import TextAssistant
+from gassist_text import TextAssistantAsync
 from google.oauth2.credentials import Credentials
 from grpc import RpcError
 
@@ -87,12 +87,12 @@ async def async_send_text_commands(
     credentials = Credentials(session.token[CONF_ACCESS_TOKEN])  # type: ignore[no-untyped-call]
     language_code = entry.options.get(CONF_LANGUAGE_CODE, default_language_code(hass))
     command_response_list = []
-    with TextAssistant(
+    async with TextAssistantAsync(
         credentials, language_code, audio_out=bool(media_players)
     ) as assistant:
         for command in commands:
             try:
-                resp = await hass.async_add_executor_job(assistant.assist, command)
+                resp = await assistant.assist(command)
             except RpcError as err:
                 _LOGGER.error(
                     "Failed to send command '%s' to Google Assistant: %s",
diff --git a/homeassistant/components/google_assistant_sdk/manifest.json b/homeassistant/components/google_assistant_sdk/manifest.json
index f1bd91aa82b6..e041515e5b8c 100644
--- a/homeassistant/components/google_assistant_sdk/manifest.json
+++ b/homeassistant/components/google_assistant_sdk/manifest.json
@@ -7,7 +7,7 @@
   "documentation": "https://www.home-assistant.io/integrations/google_assistant_sdk",
   "integration_type": "service",
   "iot_class": "cloud_polling",
-  "quality_scale": "gold",
-  "requirements": ["gassist-text==0.0.14"],
+  "quality_scale": "platinum",
+  "requirements": ["gassist-text==0.1.0"],
   "single_config_entry": true
 }
diff --git a/homeassistant/components/google_assistant_sdk/quality_scale.yaml b/homeassistant/components/google_assistant_sdk/quality_scale.yaml
index 32d308975633..4d840293fa10 100644
--- a/homeassistant/components/google_assistant_sdk/quality_scale.yaml
+++ b/homeassistant/components/google_assistant_sdk/quality_scale.yaml
@@ -97,7 +97,7 @@ rules:
     comment: No devices.
 
   # Platinum
-  async-dependency: todo
+  async-dependency: done
   inject-websession:
     status: exempt
     comment: The underlying library uses gRPC, not aiohttp/httpx, for communication.
diff --git a/requirements_all.txt b/requirements_all.txt
index 67c9b22a8f46..852f6d61b9f2 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -1100,7 +1100,7 @@ gTTS==2.5.4
 gardena-bluetooth==2.10.1
 
 # homeassistant.components.google_assistant_sdk
-gassist-text==0.0.14
+gassist-text==0.1.0
 
 # homeassistant.components.gatus
 gatus-api==1.2.0
diff --git a/tests/components/google_assistant_sdk/test_init.py b/tests/components/google_assistant_sdk/test_init.py
index 2d799f5e21fc..0473fc819837 100644
--- a/tests/components/google_assistant_sdk/test_init.py
+++ b/tests/components/google_assistant_sdk/test_init.py
@@ -1,5 +1,6 @@
 """Tests for Google Assistant SDK."""
 
+import asyncio
 from datetime import timedelta
 import http
 import time
@@ -164,7 +165,7 @@ async def test_send_text_command(
     expected_language_code: str,
     config_entry: MockConfigEntry,
 ) -> None:
-    """Test service call send_text_command calls TextAssistant."""
+    """Test service call send_text_command calls TextAssistantAsync."""
     await setup_integration()
 
     assert config_entry.state is ConfigEntryState.LOADED
@@ -174,7 +175,7 @@ async def test_send_text_command(
 
     command = "turn on home assistant unsupported device"
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant"
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync"
     ) as mock_text_assistant:
         await hass.services.async_call(
             DOMAIN,
@@ -186,7 +187,7 @@ async def test_send_text_command(
         ExpectedCredentials(), expected_language_code, audio_out=False
     )
     # pylint:disable-next=unnecessary-dunder-call
-    mock_text_assistant.assert_has_calls([call().__enter__().assist(command)])
+    mock_text_assistant.assert_has_calls([call().__aenter__().assist(command)])
 
 
 async def test_send_text_commands(
@@ -194,7 +195,7 @@ async def test_send_text_commands(
     setup_integration: ComponentSetup,
     config_entry: MockConfigEntry,
 ) -> None:
-    """Test service call send_text_command calls TextAssistant."""
+    """Test service call send_text_command calls TextAssistantAsync."""
     await setup_integration()
 
     assert config_entry.state is ConfigEntryState.LOADED
@@ -204,7 +205,7 @@ async def test_send_text_commands(
     command1_response = "what's the PIN?"
     command2_response = "opened the garage door"
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
         side_effect=[
             (command1_response, None, None),
             (command2_response, None, None),
@@ -278,7 +279,7 @@ async def test_send_text_command_grpc_error(
     command = "turn on home assistant unsupported device"
     with (
         patch(
-            "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+            "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
             side_effect=RpcError(),
         ) as mock_assist_call,
         pytest.raises(HomeAssistantError),
@@ -308,7 +309,7 @@ async def test_send_text_command_media_player(
     audio_response1 = b"joke1 audio response bytes"
     audio_response2 = b"joke2 audio response bytes"
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
         side_effect=[
             ("joke1 text", None, audio_response1),
             ("joke2 text", None, audio_response2),
@@ -399,7 +400,8 @@ async def test_conversation_agent(
     text1 = "tell me a joke"
     text2 = "tell me another one"
     with patch(
-        "homeassistant.components.google_assistant_sdk.TextAssistant"
+        "homeassistant.components.google_assistant_sdk.TextAssistantAsync",
+        autospec=True,
     ) as mock_text_assistant:
         await conversation.async_converse(
             hass, text1, None, Context(), "en-US", config_entry.entry_id
@@ -432,7 +434,8 @@ async def test_conversation_agent_refresh_token(
     text1 = "tell me a joke"
     text2 = "tell me another one"
     with patch(
-        "homeassistant.components.google_assistant_sdk.TextAssistant"
+        "homeassistant.components.google_assistant_sdk.TextAssistantAsync",
+        autospec=True,
     ) as mock_text_assistant:
         await conversation.async_converse(
             hass, text1, None, Context(), "en-US", config_entry.entry_id
@@ -463,6 +466,8 @@ async def test_conversation_agent_refresh_token(
     )
     mock_text_assistant.assert_has_calls([call().assist(text1)])
     mock_text_assistant.assert_has_calls([call().assist(text2)])
+    # The replaced assistant is closed rather than left holding its gRPC channel
+    mock_text_assistant.return_value.close.assert_awaited_once()
 
 
 async def test_conversation_agent_language_changed(
@@ -481,7 +486,8 @@ async def test_conversation_agent_language_changed(
     text1 = "tell me a joke"
     text2 = "cuéntame un chiste"
     with patch(
-        "homeassistant.components.google_assistant_sdk.TextAssistant"
+        "homeassistant.components.google_assistant_sdk.TextAssistantAsync",
+        autospec=True,
     ) as mock_text_assistant:
         await conversation.async_converse(
             hass, text1, None, Context(), "en-US", config_entry.entry_id
@@ -496,6 +502,92 @@ async def test_conversation_agent_language_changed(
     mock_text_assistant.assert_has_calls([call(ExpectedCredentials(), "es-ES")])
     mock_text_assistant.assert_has_calls([call().assist(text1)])
     mock_text_assistant.assert_has_calls([call().assist(text2)])
+    # The replaced assistant is closed rather than left holding its gRPC channel
+    mock_text_assistant.return_value.close.assert_awaited_once()
+
+
+async def test_conversation_agent_serializes_requests(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    setup_integration: ComponentSetup,
+) -> None:
+    """Test concurrent conversations do not overlap on the shared assistant.
+
+    The assistant holds the state of a single conversation, and it is closed
+    when it is replaced, so a request must not start while another one is still
+    waiting for its response.
+    """
+    await setup_integration()
+
+    assert await async_setup_component(hass, "homeassistant", {})
+    assert await async_setup_component(hass, "conversation", {})
+
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    events: list[str] = []
+
+    async def assist(text: str) -> tuple[str, None, None]:
+        events.append(f"start {text}")
+        await asyncio.sleep(0)
+        events.append(f"end {text}")
+        return (text, None, None)
+
+    async def close() -> None:
+        events.append("close")
+
+    with patch(
+        "homeassistant.components.google_assistant_sdk.TextAssistantAsync",
+        autospec=True,
+    ) as mock_text_assistant:
+        mock_text_assistant.return_value.assist.side_effect = assist
+        mock_text_assistant.return_value.close.side_effect = close
+        # Different languages, so the second request replaces the assistant
+        await asyncio.gather(
+            conversation.async_converse(
+                hass, "one", None, Context(), "en-US", config_entry.entry_id
+            ),
+            conversation.async_converse(
+                hass, "two", None, Context(), "es-ES", config_entry.entry_id
+            ),
+        )
+
+    # Whichever request runs first creates the assistant, and the other one
+    # changes the language, so it replaces and closes that one. Either order is
+    # valid, but a request has to finish before the next one starts, and the
+    # close has to land between them rather than during a request.
+    assert events in (
+        ["start one", "end one", "close", "start two", "end two"],
+        ["start two", "end two", "close", "start one", "end one"],
+    )
+
+
+async def test_conversation_agent_closed_on_unload(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    setup_integration: ComponentSetup,
+) -> None:
+    """Test unloading the entry closes the assistant the agent was holding."""
+    await setup_integration()
+
+    assert await async_setup_component(hass, "homeassistant", {})
+    assert await async_setup_component(hass, "conversation", {})
+
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    with patch(
+        "homeassistant.components.google_assistant_sdk.TextAssistantAsync",
+        autospec=True,
+    ) as mock_text_assistant:
+        await conversation.async_converse(
+            hass, "tell me a joke", None, Context(), "en-US", config_entry.entry_id
+        )
+        mock_text_assistant.return_value.close.assert_not_awaited()
+
+        await hass.config_entries.async_unload(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert config_entry.state is ConfigEntryState.NOT_LOADED
+    mock_text_assistant.return_value.close.assert_awaited_once()
 
 
 async def test_oauth_implementation_not_available(
diff --git a/tests/components/google_assistant_sdk/test_notify.py b/tests/components/google_assistant_sdk/test_notify.py
index ca4162c9e7aa..04db913fa29b 100644
--- a/tests/components/google_assistant_sdk/test_notify.py
+++ b/tests/components/google_assistant_sdk/test_notify.py
@@ -41,7 +41,7 @@ async def test_broadcast_no_targets(
     )
 
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant"
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync"
     ) as mock_text_assistant:
         await hass.services.async_call(
             notify.DOMAIN,
@@ -53,7 +53,7 @@ async def test_broadcast_no_targets(
         ExpectedCredentials(), language_code, audio_out=False
     )
     # pylint:disable-next=unnecessary-dunder-call
-    mock_text_assistant.assert_has_calls([call().__enter__().assist(expected_command)])
+    mock_text_assistant.assert_has_calls([call().__aenter__().assist(expected_command)])
 
 
 async def test_broadcast_grpc_error(
@@ -65,7 +65,7 @@ async def test_broadcast_grpc_error(
 
     with (
         patch(
-            "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+            "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
             side_effect=RpcError(),
         ) as mock_assist_call,
         pytest.raises(HomeAssistantError),
@@ -122,7 +122,7 @@ async def test_broadcast_one_target(
     )
 
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
         return_value=("text_response", None, b""),
     ) as mock_assist_call:
         await hass.services.async_call(
@@ -146,7 +146,7 @@ async def test_broadcast_two_targets(
     expected_command1 = "broadcast to basement time for dinner"
     expected_command2 = "broadcast to master bedroom time for dinner"
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
         return_value=("text_response", None, b""),
     ) as mock_assist_call:
         await hass.services.async_call(
@@ -167,7 +167,7 @@ async def test_broadcast_empty_message(
     await setup_integration()
 
     with patch(
-        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist",
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistantAsync.assist",
         return_value=("text_response", None, b""),
     ) as mock_assist_call:
         await hass.services.async_call(