Add tool/function call support to webhook conversation responses

custom_components/webhook_conversation/ai_task.py

@@ -4,6 +4,7 @@
 
 import base64
 import logging
+from typing import Any
 
 import anyio
 from voluptuous_openapi import convert
@@ -14,6 +15,7 @@
 from homeassistant.helpers import llm
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
+from .const import CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD
 from .entity import WebhookConversationLLMBaseEntity
 from .models import WebhookConversationBinaryObject
 
@@ -75,14 +77,19 @@ async def _async_generate_data(
                 task.structure.schema, custom_serializer=llm.selector_serializer
             )
 
+        reply: Any
         if self._streaming_enabled:
-            reply_parts = [
-                content_chunk
-                async for content_chunk in self._send_payload_streaming(payload)
-            ]
+            reply_parts = []
+            async for chunk_data in self._send_payload_streaming(payload):
+                if chunk_data.get("type") == "item" and "content" in chunk_data:
+                    reply_parts.append(chunk_data["content"])
             reply = "".join(reply_parts)
         else:
-            reply = await self._send_payload(payload)
+            output_field: str = self._subentry.data.get(
+                CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD
+            )
+            result = await self._send_payload(payload)
+            reply = result.get(output_field)
 
         if not task.structure:
             text = reply if isinstance(reply, str) else str(reply)

custom_components/webhook_conversation/conversation.py

@@ -15,15 +15,18 @@
     area_registry as ar,
     device_registry as dr,
     entity_registry as er,
+    llm,
 )
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from .const import DOMAIN
+from .const import CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD, DOMAIN
 from .entity import WebhookConversationLLMBaseEntity
 from .models import WebhookConversationPayload
 
 _LOGGER = logging.getLogger(__name__)
 
+MAX_TOOL_ITERATIONS = 10
+
 
 async def async_setup_entry(
     hass: HomeAssistant,
@@ -137,31 +140,71 @@ def set_default(obj: Any) -> Any:
         payload["user_id"] = user_input.context.user_id
         payload["user_name"] = user.name if user else None
 
-        if self._streaming_enabled:
-            async for _ in chat_log.async_add_delta_content_stream(
-                self.entity_id,
-                self._transform_webhook_stream(payload),
-            ):
-                pass
-        else:
-            reply = await self._send_payload(payload)
-            async for _ in chat_log.async_add_assistant_content(
-                conversation.AssistantContent(
+        output_field: str = self._subentry.data.get(
+            CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD
+        )
+
+        for _iteration in range(MAX_TOOL_ITERATIONS):
+            if self._streaming_enabled:
+                async for _ in chat_log.async_add_delta_content_stream(
                     self.entity_id,
-                    reply,
+                    self._transform_webhook_stream(payload),
+                ):
+                    pass
+            else:
+                result = await self._send_payload(payload, allow_tool_only=True)
+                reply = result.get(output_field)
+                tool_calls = _parse_tool_calls(result.get("tool_calls"))
+                async for _ in chat_log.async_add_assistant_content(
+                    conversation.AssistantContent(
+                        agent_id=self.entity_id,
+                        content=reply,
+                        tool_calls=tool_calls or None,
+                    )
+                ):
+                    pass
+
+            if not chat_log.unresponded_tool_results:
+                break
+
+            payload = self._build_payload(chat_log, include_last=True)
+            payload["query"] = user_messages[-1]["content"]
+            payload["agent_id"] = user_input.agent_id
+            payload["device_id"] = user_input.device_id
+            payload["device_info"] = (
+                (
+                    device.dict_repr
+                    if (device := device_registry.async_get(user_input.device_id))
+                    else None
                 )
-            ):
-                pass
+                if user_input.device_id
+                else None
+            )
+            payload["exposed_entities"] = json.dumps(
+                self._get_exposed_entities(), default=set_default
+            )
+            payload["language"] = user_input.language
+            payload["user_id"] = user_input.context.user_id
+            payload["user_name"] = user.name if user else None
+        else:
+            raise HomeAssistantError(
+                f"Webhook tool call loop exceeded {MAX_TOOL_ITERATIONS} iterations"
+            )
 
     async def _transform_webhook_stream(
         self, payload: WebhookConversationPayload
     ) -> AsyncIterator[conversation.AssistantContentDeltaDict]:
         """Transform webhook streaming content into HA format."""
         yield {"role": "assistant"}
 
-        async for content_delta in self._send_payload_streaming(payload):
-            _LOGGER.debug("Webhook streaming response: %s", content_delta)
-            yield {"content": content_delta}
+        async for chunk_data in self._send_payload_streaming(payload):
+            _LOGGER.debug("Webhook streaming response: %s", chunk_data)
+            if chunk_data.get("type") == "item" and "content" in chunk_data:
+                yield {"content": chunk_data["content"]}
+            elif chunk_data.get("type") == "tool_calls" and "tool_calls" in chunk_data:
+                tool_calls = _parse_tool_calls(chunk_data["tool_calls"])
+                if tool_calls:
+                    yield {"tool_calls": tool_calls}
 
     def _get_exposed_entities(self) -> list[dict[str, Any]]:
         states = [
@@ -212,3 +255,20 @@ def _get_exposed_entities(self) -> list[dict[str, Any]]:
                 }
             )
         return exposed_entities
+
+
+def _parse_tool_calls(
+    raw_tool_calls: list[dict[str, Any]] | None,
+) -> list[llm.ToolInput] | None:
+    """Parse tool calls from webhook response into ToolInput objects."""
+    if not raw_tool_calls:
+        return None
+    return [
+        llm.ToolInput(
+            tool_name=tool_call["name"],
+            tool_args=tool_call.get("arguments", {}),
+            id=tool_call.get("id", ""),
+        )
+        for tool_call in raw_tool_calls
+        if "name" in tool_call
+    ] or None

custom_components/webhook_conversation/entity.py

@@ -91,7 +91,11 @@ def __init__(self, config_entry: ConfigEntry, subentry: ConfigSubentry) -> None:
             CONF_ENABLE_STREAMING, DEFAULT_ENABLE_STREAMING
         )
 
-    async def _send_payload(self, payload: WebhookConversationPayload) -> Any:
+    async def _send_payload(
+        self,
+        payload: WebhookConversationPayload,
+        allow_tool_only: bool = False,
+    ) -> dict[str, Any]:
         """Send the payload to the webhook."""
         _LOGGER.debug(
             "Webhook request: %s",
@@ -118,15 +122,18 @@ async def _send_payload(self, payload: WebhookConversationPayload) -> Any:
         output_field: str = self._subentry.data.get(
             CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD
         )
-        if not isinstance(result, dict) or output_field not in result:
+        if not isinstance(result, dict) or (
+            output_field not in result
+            and not (allow_tool_only and "tool_calls" in result)
+        ):
             raise HomeAssistantError(f"Invalid webhook response: {result}")
 
         _LOGGER.debug("Webhook response: %s", result)
-        return result.get(output_field)
+        return result
 
     async def _send_payload_streaming(
         self, payload: WebhookConversationPayload
-    ) -> AsyncGenerator[str]:
+    ) -> AsyncGenerator[dict[str, Any]]:
         """Send the payload to the webhook and stream the response."""
         _LOGGER.debug("Webhook streaming request: %s", payload)
 
@@ -153,32 +160,32 @@ async def _send_payload_streaming(
                         try:
                             chunk_data = json.loads(line_str)
                             chunk_type = chunk_data.get("type")
-                            if chunk_type == "item" and "content" in chunk_data:
-                                yield chunk_data["content"]
-                            elif chunk_type == "error":
+                            if chunk_type == "error":
                                 raise HomeAssistantError(
                                     f"n8n error: {chunk_data.get('message', chunk_data)}"
                                 )
                             # We don't break on "end" because n8n can send multiple
                             # begin/end blocks when using tools or intermediate steps.
                             # We keep reading until the stream actually closes.
+                            yield chunk_data
                         except json.JSONDecodeError:
                             _LOGGER.warning(
                                 "Failed to parse streaming response chunk: %s", line_str
                             )
                             continue
 
     def _build_payload(
-        self, chat_log: conversation.ChatLog
+        self, chat_log: conversation.ChatLog, include_last: bool = False
     ) -> WebhookConversationPayload:
         """Create a base payload from the chat log for webhook calls."""
         system_message = chat_log.content[0]
         if not isinstance(system_message, conversation.SystemContent):
             raise TypeError("First message must be a system message")
 
+        end_idx = len(chat_log.content) if include_last else -1
         messages = [
             self._convert_content_to_param(content)
-            for content in chat_log.content[1:-1]
+            for content in chat_log.content[1:end_idx]
         ]
 
         return WebhookConversationPayload(
@@ -195,11 +202,35 @@ def _convert_content_to_param(
         self, content: conversation.Content
     ) -> WebhookConversationMessage:
         """Convert native chat content into a simple dict."""
+        if isinstance(content, conversation.ToolResultContent):
+            return WebhookConversationMessage(
+                {
+                    "role": content.role,
+                    "content": json.dumps(content.tool_result)
+                    if content.tool_result is not None
+                    else "",
+                    "tool_call_id": content.tool_call_id,
+                    "tool_name": content.tool_name,
+                }
+            )
+        if isinstance(content, conversation.AssistantContent) and content.tool_calls:
+            return WebhookConversationMessage(
+                {
+                    "role": content.role,
+                    "content": content.content or "",
+                    "tool_calls": [
+                        {
+                            "id": tool_call.id,
+                            "name": tool_call.tool_name,
+                            "arguments": tool_call.tool_args,
+                        }
+                        for tool_call in content.tool_calls
+                    ],
+                }
+            )
         return WebhookConversationMessage(
             {
                 "role": content.role,
-                "content": str(content.tool_result or "")
-                if isinstance(content, conversation.ToolResultContent)
-                else content.content or "",
+                "content": content.content or "",
             }
         )

custom_components/webhook_conversation/models.py

@@ -8,11 +8,22 @@
 MessageRole = Literal["assistant", "system", "tool_result", "user"]
 
 
+class WebhookConversationToolCall(TypedDict):
+    """A single tool call item."""
+
+    id: str
+    name: str
+    arguments: dict[str, Any]
+
+
 class WebhookConversationMessage(TypedDict):
     """A single message item."""
 
     role: MessageRole
     content: str
+    tool_calls: NotRequired[list[WebhookConversationToolCall]]
+    tool_call_id: NotRequired[str]
+    tool_name: NotRequired[str]
 
 
 class WebhookConversationBinaryObject(TypedDict):