Home Assistant Core Skill
Core Home Assistant API integration patterns, authentication, and entity management.
Activation Triggers
Activate this skill when working with:
- Home Assistant REST API
- WebSocket API connections
- Entity state management
- Service calls
- Event bus subscriptions
Authentication
Long-Lived Access Token
import httpx
class HAClient:
def __init__(self, url: str, token: str):
self.url = url.rstrip('/')
self.headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
async def get_states(self) -> list:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.url}/api/states",
headers=self.headers
)
response.raise_for_status()
return response.json()
async def get_state(self, entity_id: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.url}/api/states/{entity_id}",
headers=self.headers
)
response.raise_for_status()
return response.json()
async def call_service(
self,
domain: str,
service: str,
data: dict = None,
target: dict = None
) -> dict:
payload = {}
if data:
payload.update(data)
if target:
payload["target"] = target
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.url}/api/services/{domain}/{service}",
headers=self.headers,
json=payload
)
response.raise_for_status()
return response.json()
TypeScript Client
interface HAState {
entity_id: string;
state: string;
attributes: Record<string, any>;
last_changed: string;
last_updated: string;
}
class HomeAssistantClient {
private url: string;
private token: string;
constructor(url: string, token: string) {
this.url = url.replace(/\/$/, '');
this.token = token;
}
private get headers(): HeadersInit {
return {
'Authorization': `Bearer ${this.token}`,
'Content-Type': 'application/json'
};
}
async getStates(): Promise<HAState[]> {
const response = await fetch(`${this.url}/api/states`, {
headers: this.headers
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
async callService(
domain: string,
service: string,
data?: Record<string, any>,
target?: { entity_id?: string | string[] }
): Promise<any> {
const response = await fetch(
`${this.url}/api/services/${domain}/${service}`,
{
method: 'POST',
headers: this.headers,
body: JSON.stringify({ ...data, target })
}
);
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
}
WebSocket API
Real-Time State Updates
import asyncio
import json
import websockets
class HAWebSocket:
def __init__(self, url: str, token: str):
self.url = url.replace('http', 'ws') + '/api/websocket'
self.token = token
self.message_id = 0
self.ws = None
async def connect(self):
self.ws = await websockets.connect(self.url)
# Wait for auth_required
await self.ws.recv()
# Send auth
await self.ws.send(json.dumps({
"type": "auth",
"access_token": self.token
}))
# Wait for auth_ok
result = json.loads(await self.ws.recv())
if result["type"] != "auth_ok":
raise Exception("Authentication failed")
async def subscribe_events(self, event_type: str = None):
self.message_id += 1
msg = {
"id": self.message_id,
"type": "subscribe_events"
}
if event_type:
msg["event_type"] = event_type
await self.ws.send(json.dumps(msg))
return self.message_id
async def subscribe_state_changes(self, entity_id: str = None):
self.message_id += 1
msg = {
"id": self.message_id,
"type": "subscribe_trigger",
"trigger": {
"platform": "state"
}
}
if entity_id:
msg["trigger"]["entity_id"] = entity_id
await self.ws.send(json.dumps(msg))
return self.message_id
async def listen(self):
async for message in self.ws:
yield json.loads(message)
# Usage
async def monitor_states():
ws = HAWebSocket("http://homeassistant.local:8123", "token")
await ws.connect()
await ws.subscribe_events("state_changed")
async for event in ws.listen():
if event.get("type") == "event":
data = event["event"]["data"]
print(f"{data['entity_id']}: {data['new_state']['state']}")
Entity Domains
| Domain | Description | Common Services |
|--------|-------------|-----------------|
| light | Lighting control | turn_on, turn_off, toggle |
| switch | On/off switches | turn_on, turn_off, toggle |
| climate | HVAC systems | set_temperature, set_hvac_mode |
| cover | Blinds, doors | open, close, set_position |
| lock | Smart locks | lock, unlock |
| media_player | Media devices | play, pause, volume_set |
| fan | Fans | turn_on, set_speed |
| vacuum | Robot vacuums | start, stop, return_to_base |
| camera | Security cameras | snapshot, record |
| sensor | Sensors | (read-only) |
| binary_sensor | Binary sensors | (read-only) |
Service Call Patterns
Light Control
# Turn on with brightness
await ha.call_service("light", "turn_on", {
"brightness_pct": 75,
"color_temp_kelvin": 4000,
"transition": 2
}, target={"entity_id": "light.living_room"})
# RGB color
await ha.call_service("light", "turn_on", {
"rgb_color": [255, 100, 50]
}, target={"entity_id": "light.accent"})
Climate Control
# Set temperature
await ha.call_service("climate", "set_temperature", {
"temperature": 72,
"hvac_mode": "heat"
}, target={"entity_id": "climate.thermostat"})
# Set preset
await ha.call_service("climate", "set_preset_mode", {
"preset_mode": "away"
}, target={"entity_id": "climate.thermostat"})
Media Player
# Play media
await ha.call_service("media_player", "play_media", {
"media_content_id": "spotify:playlist:abc123",
"media_content_type": "playlist"
}, target={"entity_id": "media_player.living_room"})
# Volume control
await ha.call_service("media_player", "volume_set", {
"volume_level": 0.5
}, target={"entity_id": "media_player.living_room"})
Error Handling
from httpx import HTTPStatusError
async def safe_service_call(ha, domain, service, data, target):
try:
result = await ha.call_service(domain, service, data, target)
return {"success": True, "result": result}
except HTTPStatusError as e:
if e.response.status_code == 401:
return {"success": False, "error": "Authentication failed"}
elif e.response.status_code == 404:
return {"success": False, "error": "Service or entity not found"}
else:
return {"success": False, "error": str(e)}
except Exception as e:
return {"success": False, "error": str(e)}
Best Practices
- Use area and label targeting for group operations
- Implement retry logic for network failures
- Cache entity states to reduce API calls
- Use WebSocket for real-time monitoring
- Validate entity_ids before service calls
- Handle unavailable states gracefully