Stream OpenClaw responses through native Assist

custom_components/openclaw/conversation.py

@@ -9,7 +9,7 @@
 from datetime import datetime, timezone
 import logging
 import re
-from typing import Any
+from typing import Any, AsyncIterator
 
 from homeassistant.components import conversation
 from homeassistant.config_entries import ConfigEntry
@@ -70,30 +70,46 @@ async def async_setup_entry(
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up the OpenClaw conversation agent."""
-    agent = OpenClawConversationAgent(hass, entry)
-    conversation.async_set_agent(hass, entry, agent)
+    async_add_entities([OpenClawConversationAgent(hass, entry)])
 
 
 async def async_unload_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
 ) -> bool:
     """Unload the conversation agent."""
-    conversation.async_unset_agent(hass, entry)
     return True
 
 
-class OpenClawConversationAgent(conversation.AbstractConversationAgent):
+class OpenClawConversationAgent(
+    conversation.ConversationEntity,
+    conversation.AbstractConversationAgent,
+):
     """Conversation agent that routes messages through OpenClaw.
 
     Enables OpenClaw to appear as a selectable agent in the Assist pipeline,
     allowing use with Voice PE, satellites, and the built-in HA Assist dialog.
     """
 
+    _attr_supports_streaming = True
+    _attr_has_entity_name = True
+
     def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
         """Initialize the conversation agent."""
         self.hass = hass
         self.entry = entry
+        self._attr_unique_id = entry.entry_id
+        self._attr_name = entry.title or "OpenClaw"
+
+    async def async_added_to_hass(self) -> None:
+        """Register this entity as the config entry's conversation agent."""
+        await super().async_added_to_hass()
+        conversation.async_set_agent(self.hass, self.entry, self)
+
+    async def async_will_remove_from_hass(self) -> None:
+        """Unregister the config entry's conversation agent."""
+        conversation.async_unset_agent(self.hass, self.entry)
+        await super().async_will_remove_from_hass()
 
     @property
     def attribution(self) -> dict[str, str]:
@@ -112,10 +128,30 @@ def supported_languages(self) -> list[str] | str:
     async def async_process(
         self, user_input: conversation.ConversationInput
     ) -> conversation.ConversationResult:
-        """Process a user message through OpenClaw.
+        """Process a message on legacy Home Assistant conversation APIs.
+
+        Home Assistant before 2025.8 cannot return a native streaming chat-log
+        result. Newer cores delegate through ``_async_handle_message`` and get
+        native delta updates from the same shared implementation.
+        """
+        if hasattr(conversation, "async_get_result_from_chat_log"):
+            return await super().async_process(user_input)
+        return await self._async_process_message(user_input)
+
+    async def _async_handle_message(
+        self,
+        user_input: conversation.ConversationInput,
+        chat_log: conversation.ChatLog,
+    ) -> conversation.ConversationResult:
+        """Process a message through the native Home Assistant chat log."""
+        return await self._async_process_message(user_input, chat_log)
 
-        Tries streaming first for lower latency (first-token fast).
-        Falls back to non-streaming if the stream yields nothing.
+    async def _async_process_message(
+        self,
+        user_input: conversation.ConversationInput,
+        chat_log: conversation.ChatLog | None = None,
+    ) -> conversation.ConversationResult:
+        """Process one Assist message, streaming when the chat log supports it.
 
         Args:
             user_input: The conversation input from HA Assist.
@@ -176,6 +212,7 @@ async def async_process(
                 resolved_agent_id,
                 system_prompt,
                 active_model,
+                chat_log,
             )
         except OpenClawApiError as err:
             _LOGGER.error("OpenClaw conversation error: %s", err)
@@ -193,6 +230,7 @@ async def async_process(
                             resolved_agent_id,
                             system_prompt,
                             active_model,
+                            chat_log,
                         )
                     except OpenClawApiError as retry_err:
                         return self._error_result(
@@ -222,6 +260,15 @@ async def async_process(
         )
         coordinator.update_last_activity()
 
+        if chat_log is not None:
+            result = conversation.async_get_result_from_chat_log(user_input, chat_log)
+            # Preserve the existing tool-code safety behavior for final TTS.
+            result.response.async_set_speech(_scrub_tool_code_fences(full_response))
+            result.continue_conversation = (
+                result.continue_conversation or self._should_continue(full_response)
+            )
+            return result
+
         intent_response = intent.IntentResponse(language=user_input.language)
         intent_response.async_set_speech(full_response)
 
@@ -279,8 +326,48 @@ async def _get_response(
         agent_id: str | None = None,
         system_prompt: str | None = None,
         model: str | None = None,
+        chat_log: conversation.ChatLog | None = None,
     ) -> str:
-        """Get a response from OpenClaw, trying streaming first."""
+        """Get a response, streaming into the HA chat log when available."""
+        if chat_log is not None:
+            try:
+                full_response = await self._async_stream_response_to_chat_log(
+                    chat_log,
+                    client,
+                    message,
+                    conversation_id,
+                    agent_id,
+                    system_prompt,
+                    model,
+                )
+            except OpenClawApiError as err:
+                _LOGGER.warning(
+                    "OpenClaw stream failed; using a non-streaming response: %s", err
+                )
+                full_response = ""
+
+            if full_response:
+                return full_response
+
+            response = await client.async_send_message(
+                message=message,
+                session_id=conversation_id,
+                model=model,
+                system_prompt=system_prompt,
+                agent_id=agent_id,
+                extra_headers=_VOICE_REQUEST_HEADERS,
+            )
+            full_response = _scrub_tool_code_fences(
+                extract_text_recursive(response) or ""
+            )
+            chat_log.async_add_assistant_content_without_tools(
+                conversation.AssistantContent(
+                    agent_id=self.entity_id or self.entry.entry_id,
+                    content=full_response or None,
+                )
+            )
+            return full_response
+
         full_response = ""
         async for chunk in client.async_stream_message(
             message=message,
@@ -305,6 +392,55 @@ async def _get_response(
         )
         return _scrub_tool_code_fences(extract_text_recursive(response) or "")
 
+    async def _async_stream_response_to_chat_log(
+        self,
+        chat_log: conversation.ChatLog,
+        client: OpenClawApiClient,
+        message: str,
+        conversation_id: str,
+        agent_id: str | None,
+        system_prompt: str | None,
+        model: str | None,
+    ) -> str:
+        """Forward OpenClaw SSE deltas to Home Assistant's native chat log."""
+        response_parts: list[str] = []
+        async for content in chat_log.async_add_delta_content_stream(
+            self.entity_id or self.entry.entry_id,
+            self._async_openclaw_delta_stream(
+                client,
+                message,
+                conversation_id,
+                agent_id,
+                system_prompt,
+                model,
+            ),
+        ):
+            if isinstance(content, conversation.AssistantContent) and content.content:
+                response_parts.append(content.content)
+        return _scrub_tool_code_fences("".join(response_parts))
+
+    async def _async_openclaw_delta_stream(
+        self,
+        client: OpenClawApiClient,
+        message: str,
+        conversation_id: str,
+        agent_id: str | None,
+        system_prompt: str | None,
+        model: str | None,
+    ) -> AsyncIterator[dict[str, str]]:
+        """Translate OpenClaw SSE chunks to Home Assistant chat-log deltas."""
+        yield {"role": "assistant"}
+        async for chunk in client.async_stream_message(
+            message=message,
+            session_id=conversation_id,
+            model=model,
+            system_prompt=system_prompt,
+            agent_id=agent_id,
+            extra_headers=_VOICE_REQUEST_HEADERS,
+        ):
+            if chunk:
+                yield {"content": chunk}
+
     @staticmethod
     def _should_continue(response: str) -> bool:
         """Determine if the conversation should continue after this response.

custom_components/openclaw/manifest.json

@@ -8,7 +8,7 @@
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/techartdev/OpenClawHomeAssistantIntegration/issues",
   "requirements": [],
-  "version": "0.1.64",
+  "version": "0.1.65-rc.1",
   "dependencies": ["conversation"],
   "after_dependencies": ["hassio", "lovelace"]
 }