aioesphomeapi Python Library

For development, it is recommended to use a Python virtual environment and install the package in editable mode. To update protobuf definitions, use the provided Docker/Podman image.

# Setup virtualenv
$ python3 -m venv .
$ source bin/activate

# Install package and test dependencies
$ pip3 install -e .
$ pip3 install -r requirements/test.txt

# Run linters & test
$ script/lint

# Update protobuf definitions (requires docker or podman)
$ docker run --rm -v $PWD:/aioesphomeapi ghcr.io/esphome/aioesphomeapi-proto-builder:latest

Install the aioesphomeapi module from PyPI using pip.

An optional Cython extension is available for improved performance. The module attempts to build it automatically if a C compiler and Python development headers are present. To prevent the build attempt, set the environment variable SKIP_CYTHON=1.

$ pip3 install aioesphomeapi

To use aioesphomeapi, the ESPHome device must have the api: component enabled in its configuration.

For secure communication, it is highly recommended to use encryption with a key. You can generate a 32-byte base64 key using openssl rand -base64 32.

# Enable the Native API
api:

# Recommended: Enable encryption
api:
  encryption:
    key: !secret api_encryption_key

Use aioesphomeapi.APIClient to establish a connection.

• For modern devices (ESPHome 2026.1.0+): Use noise_psk for encryption. Password authentication has been removed.
• For older devices: Use the password argument.

After initializing the client, you must call await api.connect(login=True) to establish the session.

import aioesphomeapi
import asyncio

async def main():
    # Connect with encryption
    api = aioesphomeapi.APIClient(
        "api_test.local",
        6053,
        noise_psk="YOUR_ENCRYPTION_KEY",
    )
    await api.connect(login=True)

    # Connect to older devices with password
    # api = aioesphomeapi.APIClient("device.local", 6053, password="MyPassword")

asyncio.run(main())

You can listen for real-time state updates from the device by using api.subscribe_states(callback). The callback function will be executed whenever a state change occurs.

import aioesphomeapi
import asyncio

async def main():
    api = aioesphomeapi.APIClient(
        "api_test.local",
        6053,
        noise_psk="YOUR_ENCRYPTION_KEY",
    )
    await api.connect(login=True)

    def change_callback(state):
        """Print the state changes of the device.""
        print(state)

    # Subscribe to the state changes
    api.subscribe_states(change_callback)

    # Keep the loop running to receive updates
    while True:
        await asyncio.sleep(1)

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

Once connected via APIClient, you can access device metadata and its available services/entities using the following methods:

• api.api_version: Returns the API version of the device's firmware.
• await api.device_info(): Returns detailed information about the device.
• await api.list_entities_services(): Returns a list of all entities and services available on the device.

import aioesphomeapi
import asyncio

async def main():
    api = aioesphomeapi.APIClient(
        "api_test.local",
        6053,
        noise_psk="YOUR_ENCRYPTION_KEY",
    )
    await api.connect(login=True)

    # Get API version
    print(api.api_version)

    # Show device details
    device_info = await api.device_info()
    print(device_info)

    # List all entities of the device
    entities = await api.list_entities_services()
    print(entities)

asyncio.run(main())

The package includes two CLI tools:

• aioesphomeapi-logs: Used for watching device logs.
• aioesphomeapi-discover: Used for discovering devices on the network.

aioesphomeapi-logs --help

aioesphomeapi-discover --help

To discover what entities (sensors, lights, etc.) and services are available on the device, use:

• await list_entities_services(): Returns a tuple of (list[EntityInfo], list[UserService]).
• await device_info_and_list_entities(): A more efficient method that fetches both device info and the entity list in a single network packet to reduce latency. Returns (DeviceInfo, list[EntityInfo], list[UserService]).

entities, services = await client.list_entities_services()
for entity in entities:
    print(f"Entity: {entity.name}, State: {entity.state}")

Use water_heater_command() to control water heater entities. It supports setting the mode, target_temperature, and granular away or on states using WaterHeaterCommandField flags.

client.water_heater_command(
    key=456,
    mode=1,
    target_temperature=50.0,
    away=True
)