17 KiB
17 KiB
Home Assistant Python API Reference
This document provides guidance on using the Home Assistant WebSocket and REST APIs with Python.
RECOMMENDED: Use the WebSocket API with automation engine commands for all interactions with Home Assistant. This is the most powerful and efficient approach.
Table of Contents
- Authentication
- WebSocket API (RECOMMENDED)
- REST API (Optional)
- PEP 723 Inline Script Metadata
- Official Documentation
Authentication
All API requests require a Long-Lived Access Token. For WebSocket connections, you'll authenticate after connecting. For REST API requests, include the token in the Authorization header.
WebSocket Authentication: Handled automatically by the connection setup (see below)
REST API Authentication:
import httpx
url = "http://homeassistant.local:8123"
token = "YOUR_LONG_LIVED_ACCESS_TOKEN"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
WebSocket API (RECOMMENDED)
This is the primary way to interact with Home Assistant. The WebSocket API provides access to the automation engine, allowing you to use native Home Assistant syntax for triggers, conditions, and actions.
Connection Setup
First, establish a WebSocket connection:
# /// script
# requires-python = ">=3.8"
# dependencies = [
# "websocket-client>=1.6.0",
# ]
# ///
import websocket
import json
import threading
import time
class HomeAssistantWebSocket:
def __init__(self, url, token):
self.url = url.replace("http://", "ws://").replace("https://", "wss://")
self.token = token
self.ws = None
self.msg_id = 1
self.callbacks = {}
self.authenticated = False
def connect(self):
"""Connect to Home Assistant WebSocket API."""
self.ws = websocket.WebSocketApp(
f"{self.url}/api/websocket",
on_message=self._on_message,
on_open=self._on_open,
on_error=self._on_error
)
# Run WebSocket in background thread
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
# Wait for authentication
timeout = 5
start = time.time()
while not self.authenticated and time.time() - start < timeout:
time.sleep(0.1)
def _on_open(self, ws):
"""Handle WebSocket connection open."""
print("Connected to Home Assistant")
def _on_error(self, ws, error):
"""Handle WebSocket errors."""
print(f"WebSocket error: {error}")
def _on_message(self, ws, message):
"""Handle incoming messages."""
data = json.loads(message)
if data.get("type") == "auth_required":
# Send authentication
ws.send(json.dumps({
"type": "auth",
"access_token": self.token
}))
elif data.get("type") == "auth_ok":
print("Authenticated successfully")
self.authenticated = True
elif data.get("type") == "auth_invalid":
print("Authentication failed - check your token")
elif data.get("id") in self.callbacks:
# Call the registered callback
self.callbacks[data["id"]](data)
def send_command(self, command, callback=None):
"""Send a command and optionally register a callback."""
msg_id = self.msg_id
self.msg_id += 1
command["id"] = msg_id
if callback:
self.callbacks[msg_id] = callback
self.ws.send(json.dumps(command))
return msg_id
# Usage
ha = HomeAssistantWebSocket("http://homeassistant.local:8123", "YOUR_TOKEN")
ha.connect()
subscribe_trigger - Listen for Events
PREFERRED METHOD for listening to specific state changes, time patterns, numeric thresholds, and more.
Why use this: Instead of filtering all state changes yourself, let Home Assistant's automation engine notify you only when your specific conditions are met.
# Subscribe to motion sensor state change
def on_motion_detected(message):
print(f"Motion detected! {message}")
# Your logic here
ha.send_command({
"type": "subscribe_trigger",
"trigger": {
"platform": "state",
"entity_id": "binary_sensor.motion_sensor",
"to": "on"
}
}, on_motion_detected)
More trigger examples:
# Time pattern - every 5 minutes
ha.send_command({
"type": "subscribe_trigger",
"trigger": {
"platform": "time_pattern",
"minutes": "/5"
}
}, lambda msg: print(f"5 minutes passed"))
# Numeric state - temperature above threshold
ha.send_command({
"type": "subscribe_trigger",
"trigger": {
"platform": "numeric_state",
"entity_id": "sensor.temperature",
"above": 25
}
}, lambda msg: print(f"Temperature above 25°C!"))
# State change with duration
ha.send_command({
"type": "subscribe_trigger",
"trigger": {
"platform": "state",
"entity_id": "binary_sensor.motion",
"to": "off",
"for": {"minutes": 5}
}
}, lambda msg: print(f"No motion for 5 minutes"))
# Template trigger
ha.send_command({
"type": "subscribe_trigger",
"trigger": {
"platform": "template",
"value_template": "{{ states('sun.sun') == 'above_horizon' }}"
}
}, lambda msg: print(f"Sun is up!"))
# Multiple triggers
ha.send_command({
"type": "subscribe_trigger",
"trigger": [
{
"platform": "state",
"entity_id": "binary_sensor.door",
"to": "on"
},
{
"platform": "state",
"entity_id": "binary_sensor.window",
"to": "on"
}
]
}, lambda msg: print(f"Door or window opened!"))
execute_script - Execute Actions
MOST POWERFUL METHOD: Execute sequences of actions using Home Assistant's native syntax.
Why use this:
- Execute complex automation logic
- Use
wait_for_triggerto wait for events - Chain multiple actions together
- Keep your script minimal - all logic is in HA syntax
- Get response data from service calls
def on_complete(message):
print(f"Script completed: {message}")
# Simple sequence
ha.send_command({
"type": "execute_script",
"sequence": [
{
"service": "light.turn_on",
"target": {"entity_id": "light.living_room"},
"data": {"brightness": 255}
},
{
"delay": {"seconds": 5}
},
{
"service": "light.turn_off",
"target": {"entity_id": "light.living_room"}
}
]
}, on_complete)
Advanced: Using wait_for_trigger
# Turn on light when motion detected, turn off after 5 minutes of no motion
ha.send_command({
"type": "execute_script",
"sequence": [
{
"service": "light.turn_on",
"target": {"entity_id": "light.living_room"}
},
{
"wait_for_trigger": [
{
"platform": "state",
"entity_id": "binary_sensor.motion",
"to": "off",
"for": {"minutes": 5}
}
],
"timeout": {"hours": 2}
},
{
"service": "light.turn_off",
"target": {"entity_id": "light.living_room"}
}
]
}, on_complete)
Getting response data from service calls:
def on_weather(message):
weather_data = message.get("result", {}).get("response_variable")
print(f"Weather forecast: {weather_data}")
ha.send_command({
"type": "execute_script",
"sequence": [
{
"service": "weather.get_forecasts",
"target": {"entity_id": "weather.home"},
"data": {"type": "daily"},
"response_variable": "weather_data"
},
{
"stop": "Done",
"response_variable": "weather_data"
}
]
}, on_weather)
Complex automation example:
# Full automation logic: turn on lights when dark, turn off after no motion
ha.send_command({
"type": "execute_script",
"sequence": [
# Check if it's dark
{
"condition": "numeric_state",
"entity_id": "sensor.light_level",
"below": 100
},
# Turn on lights at 50% brightness
{
"service": "light.turn_on",
"target": {"area_id": "living_room"},
"data": {"brightness_pct": 50}
},
# Wait for motion to stop for 10 minutes
{
"wait_for_trigger": [
{
"platform": "state",
"entity_id": "binary_sensor.motion",
"to": "off",
"for": {"minutes": 10}
}
],
"timeout": {"hours": 4}
},
# Turn off lights
{
"service": "light.turn_off",
"target": {"area_id": "living_room"}
}
]
}, on_complete)
Using conditions and choose:
# Different actions based on time of day
ha.send_command({
"type": "execute_script",
"sequence": [
{
"choose": [
{
"conditions": {
"condition": "time",
"after": "06:00:00",
"before": "22:00:00"
},
"sequence": [
{
"service": "light.turn_on",
"target": {"entity_id": "light.living_room"},
"data": {"brightness_pct": 100}
}
]
}
],
"default": [
{
"service": "light.turn_on",
"target": {"entity_id": "light.living_room"},
"data": {"brightness_pct": 20}
}
]
}
]
}, on_complete)
test_condition - Test Conditions
Test conditions server-side without implementing logic in your code.
Why use this: Offload condition logic to Home Assistant. Your script stays simple while using HA's powerful condition engine.
def check_result(message):
if message.get("result", {}).get("result"):
print("Condition is true")
else:
print("Condition is false")
# Numeric state condition
ha.send_command({
"type": "test_condition",
"condition": {
"condition": "numeric_state",
"entity_id": "sensor.temperature",
"above": 20
}
}, check_result)
More condition examples:
# State condition
ha.send_command({
"type": "test_condition",
"condition": {
"condition": "state",
"entity_id": "light.living_room",
"state": "on"
}
}, check_result)
# Time condition
ha.send_command({
"type": "test_condition",
"condition": {
"condition": "time",
"after": "18:00:00",
"before": "23:00:00"
}
}, check_result)
# Template condition
ha.send_command({
"type": "test_condition",
"condition": {
"condition": "template",
"value_template": "{{ is_state('sun.sun', 'above_horizon') }}"
}
}, check_result)
# And/Or conditions
ha.send_command({
"type": "test_condition",
"condition": {
"condition": "and",
"conditions": [
{
"condition": "state",
"entity_id": "binary_sensor.motion",
"state": "on"
},
{
"condition": "numeric_state",
"entity_id": "sensor.light_level",
"below": 100
}
]
}
}, check_result)
subscribe_entities - Monitor All Entities
Subscribe to get real-time updates for all entity states. Useful for dashboards or monitoring applications.
def on_entities_update(message):
# Get the event with updated entities
if message.get("type") == "event":
event = message.get("event", {})
entities = event.get("a", {}) # 'a' contains added/updated entities
for entity_id, entity_data in entities.items():
print(f"{entity_id}: {entity_data.get('s')} ({entity_data.get('a', {})})")
ha.send_command({
"type": "subscribe_entities"
}, on_entities_update)
Note: For Python, you'll need to manually track the entity state map. For Node.js, home-assistant-js-websocket provides a built-in helper that maintains this for you.
Registry Information
Get information about devices, areas, and floors:
def on_registry_response(message):
items = message.get("result", [])
for item in items:
print(item)
# Get entity registry
ha.send_command({
"type": "config/entity_registry/list"
}, on_registry_response)
# Get device registry
ha.send_command({
"type": "config/device_registry/list"
}, on_registry_response)
# Get area registry
ha.send_command({
"type": "config/area_registry/list"
}, on_registry_response)
# Get floor registry
ha.send_command({
"type": "config/floor_registry/list"
}, on_registry_response)
REST API (Optional)
Note: For most use cases, prefer the WebSocket API above. Use REST API only for simple queries or when WebSocket is not available.
Connection Validation
Validate connection and get Home Assistant version:
# /// script
# requires-python = ">=3.8"
# dependencies = [
# "httpx>=0.27.0",
# ]
# ///
import httpx
url = "http://homeassistant.local:8123"
token = "YOUR_LONG_LIVED_ACCESS_TOKEN"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# Get configuration and version
response = httpx.get(f"{url}/api/config", headers=headers)
config = response.json()
print(f"Home Assistant Version: {config['version']}")
print(f"Location: {config['location_name']}")
print(f"Time Zone: {config['time_zone']}")
Basic Queries
Simple REST queries for when you don't need real-time updates:
import httpx
# Get all entity states
response = httpx.get(f"{url}/api/states", headers=headers)
states = response.json()
for state in states:
print(f"{state['entity_id']}: {state['state']}")
# Get specific entity state
entity_id = "light.living_room"
response = httpx.get(f"{url}/api/states/{entity_id}", headers=headers)
state = response.json()
print(f"State: {state['state']}")
print(f"Attributes: {state['attributes']}")
# Call a service (prefer execute_script via WebSocket instead)
response = httpx.post(
f"{url}/api/services/light/turn_on",
headers=headers,
json={
"entity_id": "light.living_room",
"brightness": 255
}
)
print(f"Service called: {response.json()}")
# Get history
from datetime import datetime, timedelta
end_time = datetime.now()
start_time = end_time - timedelta(hours=1)
response = httpx.get(
f"{url}/api/history/period/{start_time.isoformat()}",
headers=headers,
params={"filter_entity_id": "sensor.temperature"}
)
history = response.json()
Error handling with httpx:
import httpx
try:
response = httpx.get(f"{url}/api/config", headers=headers, timeout=10.0)
response.raise_for_status()
config = response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
print("Authentication failed - check your token")
elif e.response.status_code == 404:
print("Endpoint not found")
else:
print(f"HTTP error: {e.response.status_code}")
except httpx.TimeoutException:
print("Request timed out")
except httpx.RequestError as e:
print(f"Connection failed: {e}")
PEP 723 Inline Script Metadata
When creating standalone Python scripts for users, always include inline script metadata at the top of the file using PEP 723 format. This allows tools like uv and pipx to automatically manage dependencies.
Format
# /// script
# requires-python = ">=3.8"
# dependencies = [
# "websocket-client>=1.6.0",
# "httpx>=0.27.0",
# ]
# ///
Running Scripts
Users can run scripts with PEP 723 metadata using:
# Using uv (recommended)
uv run script.py
# Using pipx
pipx run script.py
# Traditional approach
pip install websocket-client httpx
python script.py
Official Documentation
For complete API documentation, see: