ha-api

Home Assistant API 集成

掌握 Home Assistant 的 REST 和 WebSocket API，实现外部集成、状态管理和实时通信。

⚠️ 开始之前

此技能可避免 5 种常见的 API 集成错误，并节省约 30% 的令牌开销。

方面	详情
可避免的常见错误	5 种以上（身份验证、WebSocket 生命周期、状态格式、错误处理）
令牌节省	与手动发现 API 相比约节省 30%
设置时间	2-5 分钟；手动设置需 15-20 分钟

此技能可避免的已知问题

1. 身份验证标头不正确 - Authorization 标头中的 Bearer 令牌必须带有 "Bearer " 前缀
2. WebSocket 生命周期管理 - 未处理 auth_required，或状态订阅方式不正确
3. 状态格式不匹配 - 混淆 state 与 attributes；JSON 负载结构不正确
4. 错误响应处理 - 未区分 4xx 客户端错误和 5xx 服务端错误
5. 服务调用的域名/服务不匹配 - 未正确路由到服务端点

快速入门

步骤 1：设置身份验证

# In Home Assistant UI:
# Settings → My Home → Create Long-Lived Access Token
# Store token securely (never commit to git)

export HA_TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
export HA_URL="http://192.168.1.100:8123"

为何重要： 所有 API 请求都需要 Bearer 令牌身份验证。为外部应用创建专用令牌后，无需更改密码即可撤销访问权限。

步骤 2：测试 REST API 连通性

# Get all entity states
curl -X GET "${HA_URL}/api/states" \
  -H "Authorization: Bearer ${HA_TOKEN}" \
  -H "Content-Type: application/json"

为何重要： 在构建复杂集成之前，先验证 Home Assistant 实例可访问且令牌有效。

步骤 3：选择 API 类型

• REST API：适用于一次性请求、状态查询和服务调用（HTTP 轮询）
• WebSocket API：适用于实时事件、持续订阅和更低延迟（约 50 ms，而非数秒）

为何重要： 不同的使用场景需要不同的 API。WebSocket 擅长实时应用；REST 则更适合偶尔发起的请求，且更简单。

重要规则

✅ 务必做到

• ✅ 将访问令牌存储在环境变量或安全保管库中（切勿硬编码）
• ✅ 在 Authorization 标头中加入 "Bearer " 前缀（大小写和空格必须完全一致）
• ✅ 发送其他命令前，先验证 WebSocket 的 auth_required 消息
• ✅ 处理 HTTP 错误（401 = 令牌无效或已过期，404 = 未找到实体，502 = HA 不可用）
• ✅ 调用服务时指定完整的域名/服务（例如 "light/turn_on"，而不只是 "turn_on"）

❌ 切勿这样做

• ❌ 将访问令牌提交到 git 或记录在日志中
• ❌ 跳过初始 WebSocket 身份验证握手
• ❌ 混用 Bearer 令牌身份验证和用户名/密码
• ❌ 假设状态值始终是字符串（值可能为 "on"、123 或 null）
• ❌ 将 entity_id 用作路径参数调用服务端点（应改用 JSON 负载）

常见错误

❌ 错误 - 缺少 Bearer 前缀：

curl -X GET "http://ha:8123/api/states" \
  -H "Authorization: ${HA_TOKEN}"  # Missing "Bearer "

✅ 正确 - 必须带有 Bearer 前缀：

curl -X GET "http://ha:8123/api/states" \
  -H "Authorization: Bearer ${HA_TOKEN}"

原因： Home Assistant 的 API 使用标准 Bearer 令牌身份验证。"Bearer " 前缀用于告知服务端采用的是基于令牌的身份验证方案，而非用户名/密码。

REST API 端点参考

状态端点

获取所有状态：

GET /api/states
Authorization: Bearer {token}

响应（200 OK）：

[
  {
    "entity_id": "light.living_room",
    "state": "on",
    "attributes": {
      "brightness": 255,
      "color_mode": "color_temp",
      "friendly_name": "Living Room Light"
    },
    "last_changed": "2025-12-31T18:00:00+00:00",
    "last_updated": "2025-12-31T18:05:00+00:00"
  }
]

获取单个实体状态：

GET /api/states/{entity_id}
Authorization: Bearer {token}

创建/更新实体状态：

POST /api/states/{entity_id}
Authorization: Bearer {token}
Content-Type: application/json

{
  "state": "on",
  "attributes": {
    "friendly_name": "Custom Entity",
    "custom_attribute": "value"
  }
}

响应（201 Created 或 200 OK）：

{
  "entity_id": "sensor.custom_sensor",
  "state": "on",
  "attributes": { ... }
}

服务端点

获取所有可用服务：

GET /api/services
Authorization: Bearer {token}

响应（200 OK）：

[
  {
    "domain": "light",
    "services": {
      "turn_on": {
        "description": "Turn on light(s)",
        "fields": {
          "entity_id": {
            "description": "The entity_id of the light(s)",
            "example": ["light.living_room", "light.bedroom"]
          },
          "brightness": {
            "description": "Brightness 0-255",
            "example": 180
          }
        }
      },
      "turn_off": { ... }
    }
  }
]

调用服务：

POST /api/services/{domain}/{service}
Authorization: Bearer {token}
Content-Type: application/json

{
  "entity_id": "light.living_room",
  "brightness": 180,
  "transition": 2
}

响应（200 OK）：

[
  {
    "entity_id": "light.living_room",
    "state": "on",
    "attributes": { ... }
  }
]

事件端点

获取所有事件：

GET /api/events
Authorization: Bearer {token}

触发事件：

POST /api/events/{event_type}
Authorization: Bearer {token}
Content-Type: application/json

{
  "custom_data": "value"
}

历史记录端点

获取实体历史记录：

GET /api/history/period/{timestamp}?filter_entity_id={entity_id}
Authorization: Bearer {token}

响应（200 OK）：

[
  [
    {
      "entity_id": "sensor.temperature",
      "state": "22.5",
      "attributes": { ... },
      "last_changed": "2025-12-31T12:00:00+00:00"
    }
  ]
]

配置端点

获取 Home Assistant 配置：

GET /api/config
Authorization: Bearer {token}

响应（200 OK）：

{
  "latitude": 52.3,
  "longitude": 4.9,
  "elevation": 0,
  "unit_system": {
    "length": "km",
    "mass": "kg",
    "temperature": "°C",
    "volume": "L"
  },
  "time_zone": "Europe/Amsterdam",
  "components": ["light", "switch", "sensor", ...]
}

模板端点

渲染模板：

POST /api/template
Authorization: Bearer {token}
Content-Type: application/json

{
  "template": "{{ states('sensor.temperature') }}"
}

响应（200 OK）：

{
  "template": "{{ states('sensor.temperature') }}",
  "result": "22.5"
}

WebSocket API 参考

连接流程

1. 打开到 /api/websocket 的 WebSocket 连接
2. 接收 auth_required 消息（必须在 10 秒内响应）
3. 发送包含令牌的 auth 消息
4. 接收 auth_ok 确认
5. 发送订阅/命令
6. 实时接收响应和事件

WebSocket 命令

身份验证：

// Message 1: Server sends (automatically)
{
  "type": "auth_required",
  "ha_version": "2025.1.0"
}

// Message 2: Client responds
{
  "type": "auth",
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

// Message 3: Server confirms
{
  "type": "auth_ok",
  "ha_version": "2025.1.0"
}

订阅事件：

{
  "id": 1,
  "type": "subscribe_events",
  "event_type": "state_changed"
}

// Responses come as:
{
  "id": 1,
  "type": "event",
  "event": {
    "type": "state_changed",
    "data": {
      "entity_id": "light.living_room",
      "old_state": { "state": "off", ... },
      "new_state": { "state": "on", ... }
    }
  }
}

调用服务：

{
  "id": 2,
  "type": "call_service",
  "domain": "light",
  "service": "turn_on",
  "service_data": {
    "entity_id": "light.living_room",
    "brightness": 200
  }
}

// Response:
{
  "id": 2,
  "type": "result",
  "success": true,
  "result": [
    {
      "entity_id": "light.living_room",
      "state": "on",
      "attributes": { ... }
    }
  ]
}

获取当前状态：

{
  "id": 3,
  "type": "get_states"
}

// Response:
{
  "id": 3,
  "type": "result",
  "success": true,
  "result": [ ... ]  // Array of all entity states
}

订阅特定触发器：

{
  "id": 4,
  "type": "subscribe_trigger",
  "trigger": {
    "platform": "state",
    "entity_id": "light.living_room"
  }
}

// Trigger fires when light state changes:
{
  "id": 4,
  "type": "event",
  "event": { ... }
}

代码示例

Python - REST API

import requests
import os

HA_URL = os.getenv("HA_URL", "http://localhost:8123")
HA_TOKEN = os.getenv("HA_TOKEN")

headers = {
    "Authorization": f"Bearer {HA_TOKEN}",
    "Content-Type": "application/json"
}

# Get all states
response = requests.get(f"{HA_URL}/api/states", headers=headers)
response.raise_for_status()
states = response.json()

# Get single entity
response = requests.get(f"{HA_URL}/api/states/light.living_room", headers=headers)
light_state = response.json()
print(f"Light state: {light_state['state']}")
print(f"Brightness: {light_state['attributes'].get('brightness', 'N/A')}")

# Call service
service_data = {
    "entity_id": "light.living_room",
    "brightness": 180,
    "transition": 2
}
response = requests.post(
    f"{HA_URL}/api/services/light/turn_on",
    headers=headers,
    json=service_data
)
response.raise_for_status()
print(f"Service call successful: {response.json()}")

# Handle errors
try:
    response = requests.get(f"{HA_URL}/api/states/invalid.entity", headers=headers)
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print("ERROR: Token invalid or expired")
    elif e.response.status_code == 404:
        print("ERROR: Entity not found")
    else:
        print(f"ERROR: {e}")

Python - WebSocket API

import asyncio
import aiohttp
import json
import os

HA_URL = os.getenv("HA_URL", "http://localhost:8123").replace("http", "ws")
HA_TOKEN = os.getenv("HA_TOKEN")

async def monitor_light_changes():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(f"{HA_URL}/api/websocket") as ws:
            # Wait for auth_required
            msg = await ws.receive_json()
            print(f"Server: {msg}")

            # Send auth
            await ws.send_json({
                "type": "auth",
                "access_token": HA_TOKEN
            })

            # Wait for auth_ok
            msg = await ws.receive_json()
            print(f"Server: {msg}")

            # Subscribe to state changes
            await ws.send_json({
                "id": 1,
                "type": "subscribe_events",
                "event_type": "state_changed"
            })

            # Listen for events
            async for msg in ws:
                event = msg.json()
                if event.get("type") == "event":
                    data = event["event"]["data"]
                    print(f"Entity: {data['entity_id']}")
                    print(f"  Old state: {data['old_state']['state']}")
                    print(f"  New state: {data['new_state']['state']}")

# Run the monitoring loop
asyncio.run(monitor_light_changes())

JavaScript - REST API

const HA_URL = process.env.HA_URL || "http://localhost:8123";
const HA_TOKEN = process.env.HA_TOKEN;

const headers = {
  "Authorization": `Bearer ${HA_TOKEN}`,
  "Content-Type": "application/json"
};

// Get all states
async function getAllStates() {
  const response = await fetch(`${HA_URL}/api/states`, { headers });
  if (!response.ok) {
    throw new Error(`HTTP ${response.status}: ${response.statusText}`);
  }
  return response.json();
}

// Get single entity
async function getEntityState(entityId) {
  const response = await fetch(`${HA_URL}/api/states/${entityId}`, { headers });
  if (response.status === 404) {
    throw new Error(`Entity ${entityId} not found`);
  }
  if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
  }
  const state = await response.json();
  console.log(`${entityId}: ${state.state}`);
  console.log(`Attributes:`, state.attributes);
  return state;
}

// Call service
async function callService(domain, service, data) {
  const response = await fetch(
    `${HA_URL}/api/services/${domain}/${service}`,
    {
      method: "POST",
      headers,
      body: JSON.stringify(data)
    }
  );
  if (!response.ok) {
    const error = await response.text();
    throw new Error(`HTTP ${response.status}: ${error}`);
  }
  return response.json();
}

// Example usage
(async () => {
  try {
    const states = await getAllStates();
    console.log(`Found ${states.length} entities`);

    await getEntityState("light.living_room");

    const result = await callService("light", "turn_on", {
      entity_id: "light.living_room",
      brightness: 180
    });
    console.log("Service call successful");
  } catch (error) {
    console.error(error);
  }
})();

JavaScript - WebSocket API

const HA_URL = (process.env.HA_URL || "http://localhost:8123").replace(/^http/, "ws");
const HA_TOKEN = process.env.HA_TOKEN;

async function subscribeToStateChanges() {
  const ws = new WebSocket(`${HA_URL}/api/websocket`);

  return new Promise((resolve, reject) => {
    ws.onopen = () => console.log("WebSocket connected");

    ws.onmessage = (event) => {
      const msg = JSON.parse(event.data);
      console.log("Message:", msg);

      if (msg.type === "auth_required") {
        // Respond to auth challenge
        ws.send(JSON.stringify({
          type: "auth",
          access_token: HA_TOKEN
        }));
      } else if (msg.type === "auth_ok") {
        // Authentication successful, subscribe to events
        ws.send(JSON.stringify({
          id: 1,
          type: "subscribe_events",
          event_type: "state_changed"
        }));
      } else if (msg.type === "event" && msg.event?.type === "state_changed") {
        // Handle state change
        const data = msg.event.data;
        console.log(`${data.entity_id} changed:`);
        console.log(`  ${data.old_state.state} → ${data.new_state.state}`);
      }
    };

    ws.onerror = (error) => {
      console.error("WebSocket error:", error);
      reject(error);
    };

    ws.onclose = () => {
      console.log("WebSocket closed");
      resolve();
    };
  });
}

subscribeToStateChanges();

cURL - 常见操作

# Get all entities
curl -X GET "http://localhost:8123/api/states" \
  -H "Authorization: Bearer ${HA_TOKEN}"

# Get specific entity
curl -X GET "http://localhost:8123/api/states/light.living_room" \
  -H "Authorization: Bearer ${HA_TOKEN}"

# Turn on light with brightness
curl -X POST "http://localhost:8123/api/services/light/turn_on" \
  -H "Authorization: Bearer ${HA_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "entity_id": "light.living_room",
    "brightness": 200,
    "transition": 2
  }'

# Turn off light
curl -X POST "http://localhost:8123/api/services/light/turn_off" \
  -H "Authorization: Bearer ${HA_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"entity_id": "light.living_room"}'

# Set climate temperature
curl -X POST "http://localhost:8123/api/services/climate/set_temperature" \
  -H "Authorization: Bearer ${HA_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "entity_id": "climate.living_room",
    "temperature": 21
  }'

# Get service schema
curl -X GET "http://localhost:8123/api/services/light" \
  -H "Authorization: Bearer ${HA_TOKEN}"

# Call automation
curl -X POST "http://localhost:8123/api/services/automation/trigger" \
  -H "Authorization: Bearer ${HA_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"entity_id": "automation.my_automation"}'

# Render template
curl -X POST "http://localhost:8123/api/template" \
  -H "Authorization: Bearer ${HA_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"template": "{{ states(\"sensor.temperature\") }}"}'

预防已知问题

问题	根本原因	解决方案
401 未授权	令牌无效、已过期或格式错误	在 Settings → Create Long-Lived Access Token 中创建新令牌；确认包含 Bearer 前缀
404 未找到	实体不存在或域错误	使用 GET /api/states 检查 entity_id；确认 domain.service 格式
WebSocket 身份验证超时	未在 10 秒内响应 auth_required	收到 auth_required 后立即发送 auth 消息
属性混淆	将状态字符串与 attributes 对象混为一谈	状态始终是字符串；attributes 包含元数据（亮度、颜色等）
服务调用静默失败	域/服务错误或缺少必填字段	使用 GET /api/services 查找可用服务和必需参数

错误处理模式

HTTP 状态码

import requests

try:
    response = requests.get(f"{HA_URL}/api/states/{entity_id}", headers=headers)

    if response.status_code == 401:
        print("Token invalid/expired - create new token in HA UI")
    elif response.status_code == 403:
        print("Forbidden - token lacks necessary permissions")
    elif response.status_code == 404:
        print("Entity not found - check entity_id")
    elif response.status_code == 502:
        print("Home Assistant unavailable - check server status")
    elif response.status_code >= 500:
        print("Server error - try again later")
    else:
        response.raise_for_status()
        return response.json()
except requests.exceptions.ConnectionError:
    print("Cannot connect to Home Assistant - verify HA_URL and network")
except requests.exceptions.Timeout:
    print("Request timeout - Home Assistant is slow to respond")

WebSocket 错误处理

const ws = new WebSocket(wsUrl);

ws.onerror = (error) => {
  console.error("WebSocket error:", error);
  // Reconnect after delay
  setTimeout(connect, 5000);
};

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);

  if (msg.type === "result" && !msg.success) {
    console.error("Command failed:", msg.error);
  }
};

ws.onclose = () => {
  console.log("Connection closed");
  // Implement reconnection logic
  setTimeout(connect, 3000);
};

捆绑资源

参考资料

位于 references/：

• REST_API_REFERENCE.md - 完整的 REST 端点文档
• WEBSOCKET_API_REFERENCE.md - WebSocket 命令参考
• AUTHENTICATION.md - 身份验证令牌创建和安全最佳实践
• SERVICE_CATALOG.md - 按域分类的常见服务

**注意：**有关特定主题的深入说明，请参阅上述参考文件。

依赖项

必需

包	语言	用途
requests	Python	用于 REST API 调用的 HTTP 客户端
aiohttp	Python	异步 HTTP/WebSocket 客户端
fetch	JavaScript	原生 HTTP 客户端（浏览器/Node.js）
WebSocket	JavaScript	原生 WebSocket API（浏览器/Node.js）

可选

包	语言	用途
httpx	Python	支持流式传输的高级 HTTP 客户端
websockets	Python	纯 Python WebSocket 库
axios	JavaScript	基于 Promise 的 HTTP 客户端

官方文档

• Home Assistant REST API
• Home Assistant WebSocket API
• Home Assistant 身份验证
• Home Assistant 集成 API

故障排除

无法进行身份验证 - 401 未授权

**症状：**所有 API 请求均返回 401，即使令牌正确也是如此。

解决方案：

1. 在 Home Assistant 中验证令牌（Settings → Developer Tools → States/Services）
2. 检查 Authorization 标头是否包含 "Bearer " 前缀
3. 如果旧令牌已过期，请创建新令牌
4. 确保已设置 HA_URL 和 HA_TOKEN 环境变量

echo $HA_TOKEN  # Verify token is set
echo $HA_URL    # Verify URL is correct (http not https)

WebSocket 连接立即关闭

**症状：**WebSocket 连接后随即关闭，且没有响应。

解决方案：

1. 确认 Home Assistant 版本支持 WebSocket API（2021.1+）
2. 检查防火墙是否未阻止 WebSocket 升级
3. 确保在收到 auth_required 后 10 秒内发送身份验证响应
4. 使用正确的 /api/websocket 路径，而非 /api/websocket/

// Incorrect path
ws = new WebSocket("ws://ha:8123/api/websocket/");

// Correct path
ws = new WebSocket("ws://ha:8123/api/websocket");

实体状态显示为错误类型

症状： 状态应为“开”/“关”，但显示为数字或布尔值。

解决方案：

1. 检查实体的集成配置
2. 在开发者工具中重新加载集成
3. 确认模板/自定义组件没有转换类型
4. API 响应中的状态始终是字符串；请根据需要进行解析

state = entity["state"]  # String: "on", "off", "123"
is_on = state == "on"   # Convert to boolean

# For numeric states
temperature = float(entity["state"])  # Convert to float

服务调用失败，但未返回错误

症状： 服务调用返回 HTTP 200，但没有任何效果。

解决方案：

1. 确认服务存在：GET /api/services/{domain}
2. 检查 service_data 中的必填字段
3. 确认 entity_id 位于 entity_ids 列表中，而不是作为单独的参数
4. 检查实体是否支持该服务（light.turn_on 无法用于 switch）

# Correct - entity_id in service_data
curl -X POST "http://ha:8123/api/services/light/turn_on" \
  -d '{"entity_id": "light.living_room"}'

# Incorrect - entity_id as path parameter
curl -X POST "http://ha:8123/api/services/light/turn_on/light.living_room"

设置检查清单

使用此技能前，请确认：

• Home Assistant 实例正在运行且可访问（通过浏览器测试）
• 已创建长期访问令牌（设置 → 我的主页 → 创建令牌）
• 令牌已存储在环境变量 (HA_TOKEN) 或安全保管库中
• Home Assistant URL 设置正确（HA_URL 使用 http 而非 https）
• 防火墙允许 API/WebSocket 连接（端口 8123 或自定义端口）
• 如果使用 Python 示例，已安装 Python 包（pip install requests aiohttp）
• 在构建完整集成前，已使用简单的 curl/fetch 请求测试连通性