Authentication Client

The NavienAuthClient handles all authentication with the Navien Smart Control API, including sign-in, token management, and automatic token refresh.

Overview

The authentication client:

  • Signs in with email and password

  • Manages JWT tokens (ID, access, refresh)

  • Provides AWS credentials for MQTT

  • Automatically refreshes expired tokens

  • Works as async context manager

Quick Start

Basic Authentication

from nwp500 import NavienAuthClient
import asyncio

async def main():
    # Use as context manager (recommended)
    async with NavienAuthClient("email@example.com", "password") as auth:
        print(f"Authenticated as: {auth.user_email}")
        print(f"User: {auth.current_user.full_name}")

        # auth is ready to use with API and MQTT clients
        # Tokens are automatically refreshed

asyncio.run(main())

Environment Variables

import os

# Set credentials in environment
os.environ['NAVIEN_EMAIL'] = 'your@email.com'
os.environ['NAVIEN_PASSWORD'] = 'your_password'

# Create without parameters
async with NavienAuthClient() as auth:
    # Credentials loaded from environment
    print(f"Logged in as: {auth.user_email}")

API Reference

Authentication Methods

sign_in()

sign_in(email=None, password=None)

Sign in to Navien Smart Control API.

Parameters:

  • email (str or None) – User email (uses constructor value if None)

  • password (str or None) – User password (uses constructor value if None)

Returns:

Authentication response with user info and tokens

Return type:

AuthenticationResponse

Raises:

Example:

auth = NavienAuthClient()

try:
    response = await auth.sign_in("email@example.com", "password")
    print(f"Signed in as: {response.user_info.full_name}")
    print(f"Tokens expire in: {response.tokens.time_until_expiry}")
except InvalidCredentialsError:
    print("Wrong email or password")

refresh_token()

refresh_token(refresh_token)

Refresh access token using refresh token.

Parameters:

refresh_token (str) – Refresh token from previous sign-in

Returns:

New auth tokens

Return type:

AuthTokens

Raises:

TokenRefreshError – If refresh fails

Note

This is usually called automatically by ensure_valid_token(). You rarely need to call it manually.

Example:

try:
    new_tokens = await auth.refresh_token(old_refresh_token)
    print(f"Token refreshed, expires: {new_tokens.expires_at}")
except TokenRefreshError:
    print("Refresh failed - need to sign in again")

ensure_valid_token()

ensure_valid_token()

Ensure access token is valid, refreshing if needed.

Returns:

Current valid tokens or None if not authenticated

Return type:

AuthTokens or None

Example:

# This is called automatically by API/MQTT clients
tokens = await auth.ensure_valid_token()
if tokens:
    print(f"Valid until: {tokens.expires_at}")

Token and Session Management

close()

close()

Close the HTTP session.

Note

Called automatically when using context manager.

Example:

auth = NavienAuthClient(email, password)
try:
    await auth.sign_in()
    # ... operations ...
finally:
    await auth.close()

get_auth_headers()

get_auth_headers()

Get HTTP headers for authenticated requests.

Returns:

Headers dictionary with Authorization bearer token

Return type:

dict[str, str]

Example:

headers = auth.get_auth_headers()
# {'Authorization': 'Bearer eyJ0eXAiOiJKV1...'}

# Used internally by API client
async with aiohttp.ClientSession() as session:
    async with session.get(url, headers=headers) as resp:
        data = await resp.json()

Properties

is_authenticated

is_authenticated

Check if currently authenticated.

Type:

bool

Example:

if auth.is_authenticated:
    print("Ready to make API calls")
else:
    await auth.sign_in(email, password)

current_user

current_user

Get current user information.

Type:

UserInfo or None

Example:

if auth.current_user:
    print(f"Name: {auth.current_user.full_name}")
    print(f"Type: {auth.current_user.user_type}")
    print(f"Status: {auth.current_user.user_status}")

current_tokens

current_tokens

Get current authentication tokens.

Type:

AuthTokens or None

Example:

if auth.current_tokens:
    tokens = auth.current_tokens
    print(f"Expires: {tokens.expires_at}")
    print(f"Time left: {tokens.time_until_expiry}")

    if tokens.is_expired:
        await auth.ensure_valid_token()

user_email

user_email

Get authenticated user’s email.

Type:

str or None

Example:

print(f"Logged in as: {auth.user_email}")

Data Models

UserInfo

class UserInfo

User information from authentication.

Parameters:

  • user_first_name – First name

  • user_last_name – Last name

  • user_type – User type

  • user_status – Account status

Properties:

  • full_name - Full name (first + last)

AuthTokens

class AuthTokens

Authentication tokens and AWS credentials.

Parameters:

  • id_token – JWT ID token

  • access_token – JWT access token

  • refresh_token – Refresh token

  • authentication_expires_in – Expiry in seconds

  • access_key_id – AWS access key (for MQTT)

  • secret_key – AWS secret key (for MQTT)

  • session_token – AWS session token (for MQTT)

  • issued_at – Token issue timestamp (auto-set if not provided)

Properties:

  • expires_at - Expiration timestamp

  • is_expired - Check if expired

  • time_until_expiry - Time remaining

  • bearer_token - Formatted bearer token

  • are_aws_credentials_expired - Check if AWS credentials expired

Methods:

classmethod model_validate(data)

Create AuthTokens from dictionary (API response or saved data).

Parameters:

data (dict[str, Any]) – Token data dictionary

Returns:

AuthTokens instance

Return type:

AuthTokens

Supports both camelCase keys (API response) and snake_case keys (saved data).

to_dict()

Serialize tokens to dictionary for storage.

Returns:

Dictionary with all token data including issued_at timestamp

Return type:

dict[str, Any]

Example:

# Save tokens
tokens = auth.current_tokens
token_data = tokens.to_dict()

# Later, restore tokens
restored = AuthTokens.model_validate(token_data)

AuthenticationResponse

class AuthenticationResponse

Complete sign-in response.

Parameters:

  • user_info – User information

  • tokens – Authentication tokens

Examples

Example 1: Basic Authentication

from nwp500 import NavienAuthClient

async def basic_auth():
    async with NavienAuthClient("email@example.com", "password") as auth:
        print(f"Authenticated: {auth.is_authenticated}")
        print(f"User: {auth.current_user.full_name}")
        print(f"Email: {auth.user_email}")

        tokens = auth.current_tokens
        print(f"Token expires: {tokens.expires_at}")
        print(f"Time remaining: {tokens.time_until_expiry}")

Example 2: Environment Variables

import os
from nwp500 import NavienAuthClient

os.environ['NAVIEN_EMAIL'] = 'your@email.com'
os.environ['NAVIEN_PASSWORD'] = 'your_password'

async def env_auth():
    async with NavienAuthClient() as auth:
        print(f"Logged in as: {auth.user_email}")

Example 3: Manual Token Management

from nwp500 import NavienAuthClient, InvalidCredentialsError

async def manual_auth():
    auth = NavienAuthClient()

    try:
        # Sign in
        response = await auth.sign_in("email@example.com", "password")
        print(f"Signed in: {response.user_info.full_name}")

        # Check token status
        if auth.current_tokens.is_expired:
            print("Token expired, refreshing...")
            await auth.ensure_valid_token()

        # Use for API calls
        headers = auth.get_auth_headers()

    except InvalidCredentialsError:
        print("Invalid credentials")
    finally:
        await auth.close()

Example 4: Long-Running Application

from nwp500 import NavienAuthClient

async def long_running():
    async with NavienAuthClient(email, password) as auth:
        while True:
            # Token is automatically refreshed
            await auth.ensure_valid_token()

            # Do work
            await perform_operations(auth)

            # Sleep
            await asyncio.sleep(3600)

Example 5: Token Restoration (Skip Re-authentication)

import json
from nwp500 import NavienAuthClient
from nwp500.auth import AuthTokens

async def save_tokens():
    """Save tokens for later reuse."""
    async with NavienAuthClient(email, password) as auth:
        tokens = auth.current_tokens

        # Serialize tokens to dictionary
        token_data = tokens.to_dict()

        # Save to file (or database, cache, etc.)
        with open('tokens.json', 'w') as f:
            json.dump(token_data, f)

        print("Tokens saved for future use")

async def restore_tokens():
    """Restore authentication from saved tokens."""
    # Load saved tokens
    with open('tokens.json') as f:
        token_data = json.load(f)

    # Deserialize tokens
    stored_tokens = AuthTokens.model_validate(token_data)

    # Initialize client with stored tokens
    # This skips initial authentication if tokens are still valid
    async with NavienAuthClient(
        email, password,
        stored_tokens=stored_tokens
    ) as auth:
        # If tokens were expired, they're automatically refreshed
        # If AWS credentials expired, re-authentication occurs
        print(f"Authenticated (from stored tokens): {auth.user_email}")

        # Always save updated tokens after refresh
        new_tokens = auth.current_tokens
        if new_tokens.issued_at != stored_tokens.issued_at:
            token_data = new_tokens.to_dict()
            with open('tokens.json', 'w') as f:
                json.dump(token_data, f)
            print("Tokens were refreshed and re-saved")

Note

Token restoration is especially useful for applications that restart frequently (like Home Assistant) to avoid unnecessary authentication requests on every restart.

Error Handling

from nwp500 import (
    InvalidCredentialsError,
    TokenExpiredError,
    TokenRefreshError,
    AuthenticationError
)

async def handle_auth_errors():
    try:
        async with NavienAuthClient(email, password) as auth:
            # Operations
            pass

    except InvalidCredentialsError:
        print("Wrong email or password")

    except TokenExpiredError:
        print("Token expired and refresh failed")

    except TokenRefreshError:
        print("Could not refresh token - sign in again")

    except AuthenticationError as e:
        print(f"Auth error: {e.message}")

Best Practices

  1. Always use context manager:

    # [OK] Correct
async with NavienAuthClient(email, password) as auth:
    # operations

# ✗ Wrong
auth = NavienAuthClient(email, password)
await auth.sign_in()
# ... forgot to call auth.close()

  2. Use environment variables for credentials:

    # Don't hardcode credentials
async with NavienAuthClient() as auth:
    # Loaded from NAVIEN_EMAIL and NAVIEN_PASSWORD
    pass

  3. Share auth client:

    async with NavienAuthClient(email, password) as auth:
    # Use same auth for both clients
    api = NavienAPIClient(auth)
    mqtt = NavienMqttClient(auth)

  4. Let automatic refresh work:

    # Don't manually check/refresh
# The client does it automatically