feat: add HomeAssistant components (#5803)
* Add Homeassistant Component https://www.home-assistant.io/ * Add files via upload * Add files via upload * refactor: rename Home Assistant control and state listing components Introduced two new components for Home Assistant integration: 1. HomeAssistantControl: A tool for controlling Home Assistant devices, allowing actions like turn_on, turn_off, and toggle with specified entity IDs. 2. ListHomeAssistantStates: A tool to retrieve the current states of Home Assistant entities, with optional filtering by domain. Both components include necessary input fields and documentation links for user guidance. * refactor: format Home Assistant SVG component and index file for consistency * fix: spread props in Home Assistant SVG component for better customization * refactor: update Home Assistant icon and improve SVG structure for better readability --------- Co-authored-by: Nadir J <31660040+NadirJ@users.noreply.github.com> Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
This commit is contained in:
YAMON.IO
GitHub
parent
413e855d22
commit
e2d0ec865a
7 changed files with 331 additions and 0 deletions
|
|
@ -0,0 +1,7 @@
|
|||
from .home_assistant_control import HomeAssistantControl
|
||||
from .list_home_assistant_states import ListHomeAssistantStates
|
||||
|
||||
__all__ = [
|
||||
"HomeAssistantControl",
|
||||
"ListHomeAssistantStates",
|
||||
]
|
||||
|
|
@ -0,0 +1,152 @@
|
|||
import json
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
from langchain.tools import StructuredTool
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from langflow.base.langchain_utilities.model import LCToolComponent
|
||||
from langflow.field_typing import Tool
|
||||
from langflow.inputs import SecretStrInput, StrInput
|
||||
from langflow.schema import Data
|
||||
|
||||
|
||||
class HomeAssistantControl(LCToolComponent):
|
||||
"""This tool is used to control Home Assistant devices.
|
||||
|
||||
A very simple tool to control Home Assistant devices.
|
||||
- The agent only needs to provide action (turn_on, turn_off, toggle) + entity_id (e.g., switch.xxx, light.xxx).
|
||||
- The domain (e.g., 'switch', 'light') is automatically extracted from entity_id.
|
||||
"""
|
||||
|
||||
display_name: str = "Home Assistant Control"
|
||||
description: str = (
|
||||
"A very simple tool to control Home Assistant devices. "
|
||||
"Only action (turn_on, turn_off, toggle) and entity_id need to be provided."
|
||||
)
|
||||
documentation: str = "https://developers.home-assistant.io/docs/api/rest/"
|
||||
icon: str = "HomeAssistant"
|
||||
|
||||
# --- Input fields for LangFlow UI (token, URL) ---
|
||||
inputs = [
|
||||
SecretStrInput(
|
||||
name="ha_token",
|
||||
display_name="Home Assistant Token",
|
||||
info="Home Assistant Long-Lived Access Token",
|
||||
required=True,
|
||||
),
|
||||
StrInput(
|
||||
name="base_url",
|
||||
display_name="Home Assistant URL",
|
||||
info="e.g., http://192.168.0.10:8123",
|
||||
required=True,
|
||||
),
|
||||
StrInput(
|
||||
name="default_action",
|
||||
display_name="Default Action (Optional)",
|
||||
info="One of turn_on, turn_off, toggle",
|
||||
required=False,
|
||||
),
|
||||
StrInput(
|
||||
name="default_entity_id",
|
||||
display_name="Default Entity ID (Optional)",
|
||||
info="Default entity ID to control (e.g., switch.unknown_switch_3)",
|
||||
required=False,
|
||||
),
|
||||
]
|
||||
|
||||
# --- Parameters exposed to the agent (Pydantic schema) ---
|
||||
class ToolSchema(BaseModel):
|
||||
"""Parameters to be passed by the agent: action, entity_id only."""
|
||||
|
||||
action: str = Field(..., description="Home Assistant service name. (One of turn_on, turn_off, toggle)")
|
||||
entity_id: str = Field(
|
||||
...,
|
||||
description="Entity ID to control (e.g., switch.xxx, light.xxx, cover.xxx, etc.)."
|
||||
"Do not infer; use the list_homeassistant_states tool to retrieve it.",
|
||||
)
|
||||
|
||||
def run_model(self) -> Data:
|
||||
"""Used when the 'Run' button is clicked in LangFlow.
|
||||
|
||||
- Uses default_action and default_entity_id entered in the UI.
|
||||
"""
|
||||
action = self.default_action or "turn_off"
|
||||
entity_id = self.default_entity_id or "switch.unknown_switch_3"
|
||||
|
||||
result = self._control_device(
|
||||
ha_token=self.ha_token,
|
||||
base_url=self.base_url,
|
||||
action=action,
|
||||
entity_id=entity_id,
|
||||
)
|
||||
return self._make_data_response(result)
|
||||
|
||||
def build_tool(self) -> Tool:
|
||||
"""Returns a tool to be used by the agent (LLM).
|
||||
|
||||
- The agent can only pass action and entity_id as arguments.
|
||||
"""
|
||||
return StructuredTool.from_function(
|
||||
name="home_assistant_control",
|
||||
description=(
|
||||
"A tool to control Home Assistant devices easily. "
|
||||
"Parameters: action ('turn_on'/'turn_off'/'toggle'), entity_id ('switch.xxx', etc.)."
|
||||
"Entity ID must be obtained using the list_homeassistant_states tool and not guessed."
|
||||
),
|
||||
func=self._control_device_for_tool, # Wrapper function below
|
||||
args_schema=self.ToolSchema,
|
||||
)
|
||||
|
||||
def _control_device_for_tool(self, action: str, entity_id: str) -> dict[str, Any] | str:
|
||||
"""Function called by the agent.
|
||||
|
||||
-> Internally calls _control_device.
|
||||
"""
|
||||
return self._control_device(
|
||||
ha_token=self.ha_token,
|
||||
base_url=self.base_url,
|
||||
action=action,
|
||||
entity_id=entity_id,
|
||||
)
|
||||
|
||||
def _control_device(
|
||||
self,
|
||||
ha_token: str,
|
||||
base_url: str,
|
||||
action: str,
|
||||
entity_id: str,
|
||||
) -> dict[str, Any] | str:
|
||||
"""Actual logic to call the Home Assistant service.
|
||||
|
||||
The domain is extracted from the beginning of the entity_id.
|
||||
Example: entity_id="switch.unknown_switch_3" -> domain="switch".
|
||||
"""
|
||||
try:
|
||||
domain = entity_id.split(".")[0] # switch, light, cover, etc.
|
||||
url = f"{base_url}/api/services/{domain}/{action}"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {ha_token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {"entity_id": entity_id}
|
||||
|
||||
response = requests.post(url, headers=headers, json=payload, timeout=10)
|
||||
response.raise_for_status()
|
||||
|
||||
return response.json() # HA response JSON on success
|
||||
except requests.exceptions.RequestException as e:
|
||||
return f"Error: Failed to call service. {e}"
|
||||
except Exception as e: # noqa: BLE001
|
||||
return f"An unexpected error occurred: {e}"
|
||||
|
||||
def _make_data_response(self, result: dict[str, Any] | str) -> Data:
|
||||
"""Returns a response in the LangFlow Data format."""
|
||||
if isinstance(result, str):
|
||||
# Handle error messages
|
||||
return Data(text=result)
|
||||
|
||||
# Convert dict to JSON string
|
||||
formatted_json = json.dumps(result, indent=2, ensure_ascii=False)
|
||||
return Data(data=result, text=formatted_json)
|
||||
|
|
@ -0,0 +1,137 @@
|
|||
import json
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
from langchain.tools import StructuredTool
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from langflow.base.langchain_utilities.model import LCToolComponent
|
||||
from langflow.field_typing import Tool
|
||||
from langflow.inputs import SecretStrInput, StrInput
|
||||
from langflow.schema import Data
|
||||
|
||||
|
||||
class ListHomeAssistantStates(LCToolComponent):
|
||||
display_name: str = "List HomeAssistant States"
|
||||
description: str = (
|
||||
"Retrieve states from Home Assistant. "
|
||||
"The agent only needs to specify 'filter_domain' (optional). "
|
||||
"Token and base_url are not exposed to the agent."
|
||||
)
|
||||
documentation: str = "https://developers.home-assistant.io/docs/api/rest/"
|
||||
icon = "HomeAssistant"
|
||||
|
||||
# 1) Define fields to be received in LangFlow UI
|
||||
inputs = [
|
||||
SecretStrInput(
|
||||
name="ha_token",
|
||||
display_name="Home Assistant Token",
|
||||
info="Home Assistant Long-Lived Access Token",
|
||||
required=True,
|
||||
),
|
||||
StrInput(
|
||||
name="base_url",
|
||||
display_name="Home Assistant URL",
|
||||
info="e.g., http://192.168.0.10:8123",
|
||||
required=True,
|
||||
),
|
||||
StrInput(
|
||||
name="filter_domain",
|
||||
display_name="Default Filter Domain (Optional)",
|
||||
info="light, switch, sensor, etc. (Leave empty to fetch all)",
|
||||
required=False,
|
||||
),
|
||||
]
|
||||
|
||||
# 2) Pydantic schema containing only parameters exposed to the agent
|
||||
class ToolSchema(BaseModel):
|
||||
"""Parameters to be passed by the agent: filter_domain only."""
|
||||
|
||||
filter_domain: str = Field("", description="Filter domain (e.g., 'light'). If empty, returns all.")
|
||||
|
||||
def run_model(self) -> Data:
|
||||
"""Execute the LangFlow component.
|
||||
|
||||
Uses self.ha_token, self.base_url, self.filter_domain as entered in the UI.
|
||||
Triggered when 'Run' is clicked directly without an agent.
|
||||
"""
|
||||
filter_domain = self.filter_domain or "" # Use "" for fetching all states
|
||||
result = self._list_states(
|
||||
ha_token=self.ha_token,
|
||||
base_url=self.base_url,
|
||||
filter_domain=filter_domain,
|
||||
)
|
||||
return self._make_data_response(result)
|
||||
|
||||
def build_tool(self) -> Tool:
|
||||
"""Build a tool object to be used by the agent.
|
||||
|
||||
The agent can only pass 'filter_domain' as a parameter.
|
||||
'ha_token' and 'base_url' are not exposed (stored as self attributes).
|
||||
"""
|
||||
return StructuredTool.from_function(
|
||||
name="list_homeassistant_states",
|
||||
description=(
|
||||
"Retrieve states from Home Assistant. "
|
||||
"You can provide filter_domain='light', 'switch', etc. to narrow results."
|
||||
),
|
||||
func=self._list_states_for_tool, # Wrapper function below
|
||||
args_schema=self.ToolSchema, # Requires only filter_domain
|
||||
)
|
||||
|
||||
def _list_states_for_tool(self, filter_domain: str = "") -> list[Any] | str:
|
||||
"""Execute the tool when called by the agent.
|
||||
|
||||
'ha_token' and 'base_url' are stored in self (not exposed).
|
||||
"""
|
||||
return self._list_states(
|
||||
ha_token=self.ha_token,
|
||||
base_url=self.base_url,
|
||||
filter_domain=filter_domain,
|
||||
)
|
||||
|
||||
def _list_states(
|
||||
self,
|
||||
ha_token: str,
|
||||
base_url: str,
|
||||
filter_domain: str = "",
|
||||
) -> list[Any] | str:
|
||||
"""Call the Home Assistant /api/states endpoint."""
|
||||
try:
|
||||
headers = {
|
||||
"Authorization": f"Bearer {ha_token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
url = f"{base_url}/api/states"
|
||||
response = requests.get(url, headers=headers, timeout=10)
|
||||
response.raise_for_status()
|
||||
|
||||
all_states = response.json()
|
||||
if filter_domain:
|
||||
return [st for st in all_states if st.get("entity_id", "").startswith(f"{filter_domain}.")]
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
return f"Error: Failed to fetch states. {e}"
|
||||
except (ValueError, TypeError) as e:
|
||||
return f"Error processing response: {e}"
|
||||
return all_states
|
||||
|
||||
def _make_data_response(self, result: list[Any] | str | dict) -> Data:
|
||||
"""Format the response into a Data object."""
|
||||
try:
|
||||
if isinstance(result, list):
|
||||
# Wrap list data into a dictionary and convert to text
|
||||
wrapped_result = {"result": result}
|
||||
return Data(data=wrapped_result, text=json.dumps(wrapped_result, indent=2, ensure_ascii=False))
|
||||
if isinstance(result, dict):
|
||||
# Return dictionary as-is
|
||||
return Data(data=result, text=json.dumps(result, indent=2, ensure_ascii=False))
|
||||
if isinstance(result, str):
|
||||
# Return error messages or strings
|
||||
return Data(data={}, text=result)
|
||||
|
||||
# Handle unexpected data types
|
||||
return Data(data={}, text="Error: Unexpected response format.")
|
||||
except (TypeError, ValueError) as e:
|
||||
# Handle specific exceptions during formatting
|
||||
return Data(data={}, text=f"Error: Failed to process response. Details: {e!s}")
|
||||
Loading…
Add table
Add a link
Reference in a new issue