aiokwikset 0.6.2

Python interface for Kwikset Smart Locks

aiokwikset

An asynchronous Python client library for the Kwikset Smart Lock API.

This library provides a fully async interface for interacting with Kwikset Smart Locks (including Halo and Aura models) via the Kwikset cloud API.

Table of Contents

• Features
• Requirements
• Installation
• Quick Start
• Usage
  • Basic Authentication
  • Using Context Manager
  • MFA Authentication
  • Session Restoration
  • Device Operations
  • Real-Time Subscriptions
  • Token Persistence
  • Home Assistant Integration
• API Reference
  • API Class
  • Device Class
  • User Class
  • SubscriptionManager Class
  • Exceptions
• Error Handling
• Contributing
• License

Features

• Fully Asynchronous: Built with asyncio and aiohttp for non-blocking I/O
• Real-Time Subscriptions: Push-based device updates via AWS AppSync WebSocket connections
• Type Annotated: Complete type hints for better IDE support and code quality
• MFA Support: Full support for SMS and Software Token (TOTP) multi-factor authentication
• Context Manager: Proper resource cleanup with async context manager support
• Token Management: Automatic token refresh with optional persistence callbacks
• Session Restoration: Restore sessions from saved tokens without re-entering password
• Retry Logic: Automatic retries with exponential backoff for transient failures
• Auto-Reconnect: WebSocket connections automatically reconnect with exponential backoff
• Configurable Timeouts: Customizable request timeouts (default: 30 seconds)
• Home Assistant Ready: Purpose-built exceptions and patterns for HA integrations
• Secure: AWS Cognito authentication with secure token handling
• Extensible: Clean architecture with modular endpoint handlers

Supported Operations

Category	Operations
Authentication	Login, MFA (SMS/TOTP), Token Refresh
Lock Control	Lock, Unlock
Device Settings	LED Status, Audio Status, Secure Screen Status
Information	User Info, Homes, Devices, Device Details
Real-Time Events	Device Status, Battery, Door Sensor, Lock Events, WiFi Signal, and more

Requirements

• Python: 3.10 or higher
• Kwikset Account: You must have a registered Kwikset account with at least one home configured
• Kwikset App: Initial setup requires the official Kwikset mobile app
  • iOS App Store
  • Google Play Store

Installation

Using pip

pip install aiokwikset

Using Poetry

poetry add aiokwikset

From Source

git clone https://github.com/explosivo22/aiokwikset.git
cd aiokwikset
pip install -e .

Development Installation

pip install -e ".[dev]"

Quick Start

import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("[e-mail]", "your-password")
        
        # Get all homes
        homes = await api.user.get_homes()
        
        # Get devices for the first home
        devices = await api.device.get_devices(homes[0]["homeid"])
        
        for device in devices:
            print(f"Device: {device['devicename']} - {device['deviceid']}")

asyncio.run(main())

Usage

Basic Authentication

import asyncio
from aiokwikset import API

async def main() -> None:
    # Create API instance
    api = API()
    
    try:
        # Authenticate with Kwikset cloud
        await api.async_login("[e-mail]", "your-password")
        
        # Check authentication status
        if api.is_authenticated:
            print("Successfully authenticated!")
            
            # Get user information
            user_info = await api.user.get_info()
            print(f"Welcome, {user_info['firstname']} {user_info['lastname']}!")
            
    finally:
        # Clean up resources
        await api.async_close()

asyncio.run(main())

Using Context Manager

The recommended way to use the API is with an async context manager, which ensures proper cleanup:

import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("[e-mail]", "your-password")
        
        homes = await api.user.get_homes()
        for home in homes:
            print(f"Home: {home['homename']} (ID: {home['homeid']})")
        
        # Resources are automatically cleaned up when exiting the context

asyncio.run(main())

MFA Authentication

If your account has Multi-Factor Authentication enabled, handle the MFA challenge:

import asyncio
from aiokwikset import API, MFAChallengeRequired

async def main() -> None:
    async with API() as api:
        try:
            await api.async_login("[e-mail]", "your-password")
            
        except MFAChallengeRequired as mfa_error:
            # Determine MFA type
            if mfa_error.mfa_type == "SOFTWARE_TOKEN_MFA":
                print("Enter the code from your authenticator app:")
            else:  # SMS_MFA
                print("Enter the code sent to your phone:")
            
            mfa_code = input("MFA Code: ")
            
            # Complete MFA authentication
            await api.async_respond_to_mfa_challenge(
                mfa_code=mfa_code,
                mfa_type=mfa_error.mfa_type,
                mfa_tokens=mfa_error.mfa_tokens,
            )
            
            print("MFA authentication successful!")
        
        # Continue with normal API usage
        homes = await api.user.get_homes()
        print(f"Found {len(homes)} home(s)")

asyncio.run(main())

For more detailed MFA examples, see MFA_USAGE.md.

Session Restoration

Restore sessions from previously saved tokens without requiring the user to re-enter their password. This is ideal for Home Assistant integrations and long-running applications:

import asyncio
from aiokwikset import API, TokenExpiredError

async def main() -> None:
    # Create API with stored tokens
    api = API(
        id_token="stored_id_token",
        access_token="stored_access_token",
        refresh_token="stored_refresh_token",
    )
    
    async with api:
        try:
            # Authenticate using stored tokens (validates and refreshes if needed)
            await api.async_authenticate_with_tokens(
                id_token="stored_id_token",
                access_token="stored_access_token",
                refresh_token="stored_refresh_token",
            )
            
            # Session restored - continue with API usage
            homes = await api.user.get_homes()
            print(f"Session restored! Found {len(homes)} home(s)")
            
        except TokenExpiredError:
            # Tokens are no longer valid - re-authentication required
            print("Stored tokens expired. Please login again.")
            await api.async_login("[e-mail]", "your-password")

asyncio.run(main())

Device Operations

Locking and Unlocking

import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("[e-mail]", "your-password")
        
        # Get user info (required for lock/unlock operations)
        user_info = await api.user.get_info()
        
        # Get homes and devices
        homes = await api.user.get_homes()
        devices = await api.device.get_devices(homes[0]["homeid"])
        
        # Get detailed info for the first device
        device_info = await api.device.get_device_info(devices[0]["deviceid"])
        
        # Lock the device
        await api.device.lock_device(device_info, user_info)
        print("Device locked!")
        
        # Unlock the device
        await api.device.unlock_device(device_info, user_info)
        print("Device unlocked!")

asyncio.run(main())

Device Information and Settings

import asyncio
from aiokwikset import API

async def main() -> None:
    async with API() as api:
        await api.async_login("[e-mail]", "your-password")
        
        # Get user info (needed for lock/unlock operations)
        user_info = await api.user.get_info()
        
        # Get all homes and devices
        homes = await api.user.get_homes()
        devices = await api.device.get_devices(homes[0]["homeid"])
        
        for device in devices:
            # Get detailed device information
            device_info = await api.device.get_device_info(device["deviceid"])
            
            if device_info:
                print(f"Device: {device_info['devicename']}")
                print(f"  Model: {device_info.get('modelnumber', 'Unknown')}")
                print(f"  Battery: {device_info.get('batterypercentage', 'N/A')}%")
                print(f"  Lock Status: {device_info.get('doorstatus', 'Unknown')}")
                
                # Configure device settings
                await api.device.set_ledstatus(device_info, "true")      # Enable LED
                await api.device.set_audiostatus(device_info, "true")    # Enable audio
                await api.device.set_securescreenstatus(device_info, "false")  # Disable secure screen

asyncio.run(main())

Real-Time Subscriptions

Receive instant push notifications when device state changes, instead of polling. The library connects to Kwikset's AWS AppSync WebSocket endpoints and delivers events via a callback.

Basic Subscription

import asyncio
from aiokwikset import API


def handle_event(subscription_name: str, data: dict) -> None:
    """Called whenever a real-time event arrives."""
    print(f"[{subscription_name}] {data}")


async def main() -> None:
    async with API() as api:
        await api.async_login("[e-mail]", "your-password")

        # Set up real-time events
        api.subscriptions.set_callback(handle_event)
        await api.subscriptions.async_subscribe_device("[e-mail]")

        # Keep running to receive events (Ctrl+C to stop)
        try:
            while True:
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            pass


asyncio.run(main())

The onManageDevice subscription delivers all device changes in a single stream:

# Example callback data:
# subscription_name = "onManageDevice"
# data = {
#     "deviceid": "abc123",
#     "devicestatus": "Locked",
#     "batterypercentage": "95",
#     "connectivitystatus": "online",
#     "doorposition": "closed",
#     "operationtype": "statusUpdate",
#     ...
# }

Available Subscriptions

Two AppSync endpoints are available — the Core API for broad device/user/home events, and the Halo V API for granular, per-attribute updates:

Core API (recommended for most use cases):

Method	Event Name	Description
async_subscribe_device(email)	onManageDevice	Lock state, battery, door position, tamper, connectivity, and all device settings
async_subscribe_user(email)	onManageUser	Account profile changes (name, phone, MFA status)
async_subscribe_home(email)	onManageHome	Home sharing, invitations, access level changes

Halo V API (fine-grained updates for newer devices):

Method	Event Name	Description
async_subscribe_hav_device_status(email)	onUpdateHavDeviceStatus	Lock/unlock state changes
async_subscribe_hav_device(email)	onUpdateHavDevice	Full device metadata updates
async_subscribe_hav_event(email)	onUpdateHavEvent	Activity log (who locked/unlocked, garage events)
async_subscribe_hav_battery_voltage(email)	onUpdateHavBatteryVoltage	Battery voltage changes
async_subscribe_hav_sensor(email)	onUpdateHavSensor	Door sensor alignment and alarm flags
async_subscribe_hav_led_status(email)	onUpdateHavLedStatus	LED indicator changes
async_subscribe_hav_percentage(email)	onUpdateHavPercentage	Garage door open percentage
async_subscribe_hav_vacation_mode(email)	onUpdateHavVacationMode	Vacation mode toggle
async_subscribe_hav_wifi_signal(email)	onUpdateHavWifiSignal	WiFi signal strength
async_subscribe_hav_cycles(email)	onUpdateHavCycles	Lock mechanism cycle count
async_subscribe_hav_auto_close_schedule(email)	onUpdateHavAutoCloseSchedule	Auto-lock schedule changes
async_subscribe_hav_temporary_key(email)	onUpdateHavTemporaryKey	Guest/temporary access keys
async_subscribe_hav_brightness_level(email)	onUpdateHavBrightnessLevel	Keypad brightness changes

Managing Subscriptions

Each subscribe method returns a subscription ID that can be used to unsubscribe:

# Subscribe
sub_id = await api.subscriptions.async_subscribe_device(email)
hav_id = await api.subscriptions.async_subscribe_hav_device_status(email)

# Unsubscribe from a specific subscription
await api.subscriptions.async_unsubscribe(sub_id)

# Close all subscriptions and WebSocket connections
await api.subscriptions.async_close()

Key Behaviors

• Lazy connections: WebSocket connections are only opened when you first subscribe to an endpoint
• Auto-reconnect: If the connection drops, all subscriptions are automatically re-established with exponential backoff
• Token refresh: Credentials are automatically refreshed on reconnection
• Session reuse: When a websession is injected (e.g., from Home Assistant), it is reused for WebSocket connections
• Cleanup: Calling api.async_close() or exiting the async with block closes all WebSocket connections

Token Persistence

For long-running applications or integrations, you can persist tokens and receive callbacks when they're refreshed:

import asyncio
import json
from pathlib import Path
from aiokwikset import API

TOKEN_FILE = Path("tokens.json")

async def save_tokens(id_token: str, access_token: str, refresh_token: str) -> None:
    """Callback to save tokens when they're refreshed."""
    TOKEN_FILE.write_text(json.dumps({
        "id_token": id_token,
        "access_token": access_token,
        "refresh_token": refresh_token,
    }))

async def main() -> None:
    # Load existing tokens if available
    tokens = {}
    if TOKEN_FILE.exists():
        tokens = json.loads(TOKEN_FILE.read_text())
    
    # Create API with token persistence callback
    api = API(
        id_token=tokens.get("id_token"),
        access_token=tokens.get("access_token"),
        refresh_token=tokens.get("refresh_token"),
        token_update_callback=save_tokens,
    )
    
    async with api:
        if not api.is_authenticated:
            await api.async_login("[e-mail]", "your-password")
        
        # Tokens will be automatically saved when refreshed
        homes = await api.user.get_homes()
        print(f"Found {len(homes)} home(s)")

asyncio.run(main())

Home Assistant Integration

When using with Home Assistant, pass the shared aiohttp session and handle the specific exceptions for proper integration behavior:

from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from aiokwikset import API, TokenExpiredError, ConnectionError, MFAChallengeRequired

async def async_setup_entry(hass, entry):
    """Set up Kwikset from a config entry."""
    session = async_get_clientsession(hass)
    
    async def async_update_tokens(id_token, access_token, refresh_token):
        """Update config entry with new tokens."""
        hass.config_entries.async_update_entry(
            entry,
            data={
                **entry.data,
                "id_token": id_token,
                "access_token": access_token,
                "refresh_token": refresh_token,
            },
        )
    
    api = API(
        websession=session,  # Reuses HA's session (won't be closed by API)
        timeout=30,  # Configurable timeout
        token_update_callback=async_update_tokens,
    )
    
    try:
        # Restore session from stored tokens
        await api.async_authenticate_with_tokens(
            id_token=entry.data["id_token"],
            access_token=entry.data["access_token"],
            refresh_token=entry.data["refresh_token"],
        )
    except TokenExpiredError:
        # Tokens expired - trigger reauth flow
        raise ConfigEntryAuthFailed("Authentication tokens expired")
    except ConnectionError:
        # Network issue - retry later
        raise ConfigEntryNotReady("Cannot connect to Kwikset API")
    
    # Store API instance for use by platforms
    hass.data.setdefault("kwikset", {})[entry.entry_id] = api
    
    # Set up real-time event subscriptions
    email = entry.data["email"]
    
    def handle_realtime_event(name: str, data: dict) -> None:
        """Handle real-time device updates."""
        if name == "onManageDevice":
            device_id = data.get("deviceid")
            if device_id:
                async_dispatcher_send(hass, f"kwikset_update_{device_id}", data)
    
    api.subscriptions.set_callback(handle_realtime_event)
    await api.subscriptions.async_subscribe_device(email)
    
    # Set up platforms
    await hass.config_entries.async_forward_entry_setups(entry, ["lock", "sensor"])
    
    return True

Config Flow Example

from aiokwikset import API, MFAChallengeRequired, Unauthenticated

class KwiksetConfigFlow(ConfigFlow, domain="kwikset"):
    """Handle a config flow for Kwikset."""
    
    async def async_step_user(self, user_input=None):
        """Handle the initial step."""
        errors = {}
        
        if user_input is not None:
            api = API()
            try:
                await api.async_login(
                    user_input["email"],
                    user_input["password"],
                )
                
                # Success - create entry with tokens
                return self.async_create_entry(
                    title=user_input["email"],
                    data={
                        "email": user_input["email"],
                        "id_token": api.id_token,
                        "access_token": api.access_token,
                        "refresh_token": api.refresh_token,
                    },
                )
                
            except MFAChallengeRequired as err:
                # Store MFA context for next step
                self._mfa_tokens = err.mfa_tokens
                self._mfa_type = err.mfa_type
                self._email = user_input["email"]
                return await self.async_step_mfa()
                
            except Unauthenticated:
                errors["base"] = "invalid_auth"
            finally:
                await api.async_close()
        
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required("email"): str,
                vol.Required("password"): str,
            }),
            errors=errors,
        )

API Reference

API Class

The main entry point for interacting with the Kwikset cloud API.

API(
    websession: Optional[ClientSession] = None,
    user_pool_region: str = "us-east-1",
    username: Optional[str] = None,
    timeout: int = 30,
    id_token: Optional[str] = None,
    access_token: Optional[str] = None,
    refresh_token: Optional[str] = None,
    token_update_callback: Optional[Callable[[str, str, str], Awaitable[None]]] = None,
)

Parameter	Type	Description
websession	Optional[ClientSession]	Existing aiohttp session to reuse (won't be closed by API)
user_pool_region	str	AWS Cognito region (default: "us-east-1")
username	Optional[str]	Pre-set username (email)
timeout	int	Request timeout in seconds (default: 30)
id_token	Optional[str]	Pre-existing ID token for session restoration
access_token	Optional[str]	Pre-existing access token
refresh_token	Optional[str]	Pre-existing refresh token
token_update_callback	Optional[Callable]	Async callback when tokens are refreshed

Properties

Property	Type	Description
is_authenticated	bool	Whether the client is currently authenticated
device	Device	Device endpoint handler (available after authentication)
user	User	User endpoint handler (available after authentication)
subscriptions	SubscriptionManager	Real-time event subscription manager (lazy, available after authentication)

Methods

Method	Description
async_login(email, password)	Authenticate with email and password
async_authenticate_with_tokens(id_token, access_token, refresh_token)	Restore session from saved tokens
async_respond_to_mfa_challenge(mfa_code, mfa_type, mfa_tokens)	Complete MFA authentication
async_renew_access_token()	Manually refresh access token
async_close()	Clean up resources and clear tokens

Device Class

Handler for device-related API endpoints.

Method	Parameters	Description
get_devices(home_id)	home_id: str	Get all devices for a home
get_device_info(device_id)	device_id: str	Get detailed device information
lock_device(device, user)	device: dict, user: dict	Lock the device
unlock_device(device, user)	device: dict, user: dict	Unlock the device
set_ledstatus(device, status)	device: dict, status: str	Set LED status ("true"/"false")
set_audiostatus(device, status)	device: dict, status: str	Set audio status ("true"/"false")
set_securescreenstatus(device, status)	device: dict, status: str	Set secure screen status ("true"/"false")
set_led_enabled(device, enabled)	device: dict, enabled: bool	Enable/disable LED (convenience method)
set_audio_enabled(device, enabled)	device: dict, enabled: bool	Enable/disable audio (convenience method)
set_secure_screen_enabled(device, enabled)	device: dict, enabled: bool	Enable/disable secure screen (convenience method)

User Class

Handler for user-related API endpoints.

Method	Description
get_info()	Get current user account information
get_homes()	Get all homes associated with the user

SubscriptionManager Class

Manages real-time WebSocket subscriptions to both AppSync endpoints. Accessed via api.subscriptions.

Method	Parameters	Description
set_callback(callback)	callback: Callable[[str, dict], None]	Set the event handler for all subscriptions
async_subscribe_device(email)	email: str	Subscribe to Core device events (returns subscription ID)
async_subscribe_user(email)	email: str	Subscribe to Core user events
async_subscribe_home(email)	email: str	Subscribe to Core home events
async_subscribe_hav_device_status(email)	email: str	Subscribe to Halo V device status
async_subscribe_hav_device(email)	email: str	Subscribe to Halo V device metadata
async_subscribe_hav_event(email)	email: str	Subscribe to Halo V activity events
async_subscribe_hav_battery_voltage(email)	email: str	Subscribe to Halo V battery updates
async_subscribe_hav_sensor(email)	email: str	Subscribe to Halo V door sensor/alarms
async_subscribe_hav_led_status(email)	email: str	Subscribe to Halo V LED status
async_subscribe_hav_percentage(email)	email: str	Subscribe to Halo V garage percentage
async_subscribe_hav_vacation_mode(email)	email: str	Subscribe to Halo V vacation mode
async_subscribe_hav_wifi_signal(email)	email: str	Subscribe to Halo V WiFi signal
async_subscribe_hav_cycles(email)	email: str	Subscribe to Halo V lock cycles
async_subscribe_hav_auto_close_schedule(email)	email: str	Subscribe to Halo V auto-close schedule
async_subscribe_hav_temporary_key(email)	email: str	Subscribe to Halo V temporary keys
async_subscribe_hav_brightness_level(email)	email: str	Subscribe to Halo V brightness
async_unsubscribe(sub_id)	sub_id: str	Unsubscribe by subscription ID
async_close()	—	Close all WebSocket connections

Exceptions

All exceptions inherit from KwiksetError.

Authentication Exceptions

Exception	Description
AuthenticationError	Base exception for authentication-related errors
MFAChallengeRequired	MFA verification is required (contains mfa_type and mfa_tokens)
Unauthenticated	Invalid credentials or missing authentication
TokenExpiredError	Tokens expired and cannot be refreshed (triggers HA reauth)
UserNotFound	User account does not exist
UserNotConfirmed	User has not confirmed their email
UserExists	User already exists (during registration)
PasswordChangeRequired	Password must be changed

Request Exceptions

Exception	Description
RequestError	HTTP request failed (non-retryable)
ConnectionError	Network connection failed after retries (HA marks unavailable)

Device Exceptions

Exception	Description
DeviceError	Base exception for device-related errors
InvalidDeviceError	Device data is invalid or missing required fields
InvalidUserError	User data is invalid or missing required fields
InvalidActionError	Invalid action requested
InvalidStatusError	Invalid status value provided
DeviceCommandError	Device command failed

General Exceptions

Exception	Description
KwiksetError	Base exception for all library errors
UnknownError	Unknown error occurred

Subscription Exceptions

Exception	Description
SubscriptionError	Base exception for subscription/WebSocket errors
SubscriptionTimeout	WebSocket connection or subscription acknowledgment timed out

Error Handling

Implement comprehensive error handling for production applications:

import asyncio
from aiokwikset import (
    API,
    KwiksetError,
    MFAChallengeRequired,
    RequestError,
    ConnectionError,
    TokenExpiredError,
    Unauthenticated,
    UserNotFound,
)

async def main() -> None:
    async with API() as api:
        try:
            await api.async_login("[e-mail]", "password")
            
        except MFAChallengeRequired as err:
            # Handle MFA challenge
            print(f"MFA required: {err.mfa_type}")
            
        except Unauthenticated:
            print("Invalid email or password")
            
        except UserNotFound:
            print("Account does not exist")
            
        except ConnectionError as err:
            # Network issues - device should be marked unavailable
            print(f"Connection failed (after retries): {err}")
            
        except TokenExpiredError:
            # Tokens expired - need to re-authenticate
            print("Session expired. Please login again.")
            
        except RequestError as err:
            print(f"Request error: {err}")
            
        except KwiksetError as err:
            print(f"Unexpected Kwikset error: {err}")

asyncio.run(main())

Home Assistant Error Handling

For Home Assistant integrations, map exceptions to appropriate HA behaviors:

from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from aiokwikset import ConnectionError, TokenExpiredError, Unauthenticated

async def async_setup_entry(hass, entry):
    try:
        await api.async_authenticate_with_tokens(...)
    except (TokenExpiredError, Unauthenticated):
        # Trigger reauth flow in HA
        raise ConfigEntryAuthFailed("Re-authentication required")
    except ConnectionError:
        # HA will retry setup later
        raise ConfigEntryNotReady("Cannot connect to Kwikset API")

Contributing

Contributions are welcome! This is a community-maintained library.

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Run tests (pytest)
5. Commit your changes (git commit -m 'Add amazing feature')
6. Push to the branch (git push origin feature/amazing-feature)
7. Open a Pull Request

Development Setup

# Clone the repository
git clone https://github.com/explosivo22/aiokwikset.git
cd aiokwikset

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run linting
ruff check .

# Run type checking
mypy aiokwikset

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

---

Disclaimer: This library is not affiliated with or endorsed by Kwikset. Use at your own risk.