Config Flow

Config Flows enable users to configure integrations through the Home Assistant UI instead of manually editing YAML files. This guide covers implementing config flows for your integration.

Overview

Config flows are defined by implementing a ConfigFlow class in your integration’s config_flow.py file. The system manages the multi-step configuration process, validation, and storage.

Config flows are the recommended way to configure integrations. They provide a better user experience and enable features like discovery and reconfiguration.

Basic Structure

A config flow inherits from ConfigFlow and defines steps:

config_flow.py

"""Config flow for My Integration."""
from __future__ import annotations

import logging
from typing import Any

import voluptuous as vol

from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
import homeassistant.helpers.config_validation as cv

from .const import DOMAIN

_LOGGER = logging.getLogger(__name__)


class MyIntegrationConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for My Integration."""
    
    VERSION = 1
    
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        errors = {}
        
        if user_input is not None:
            # Validate user input
            try:
                # Test connection
                await self._test_connection(
                    user_input[CONF_HOST],
                    user_input[CONF_USERNAME],
                    user_input[CONF_PASSWORD],
                )
            except ConnectionError:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                # Create entry
                return self.async_create_entry(
                    title=user_input[CONF_HOST],
                    data=user_input,
                )
        
        # Show form
        data_schema = vol.Schema(
            {
                vol.Required(CONF_HOST): cv.string,
                vol.Required(CONF_USERNAME): cv.string,
                vol.Required(CONF_PASSWORD): cv.string,
            }
        )
        
        return self.async_show_form(
            step_id="user",
            data_schema=data_schema,
            errors=errors,
        )
    
    async def _test_connection(
        self, host: str, username: str, password: str
    ) -> None:
        """Test if we can connect."""
        # Implement your connection test
        pass

Flow Result Types

From data_entry_flow.py:28, flows return different result types:

class FlowResultType(StrEnum):
    """Result type for a data entry flow."""
    FORM = "form"
    CREATE_ENTRY = "create_entry"
    ABORT = "abort"
    EXTERNAL_STEP = "external"
    EXTERNAL_STEP_DONE = "external_done"
    SHOW_PROGRESS = "progress"
    SHOW_PROGRESS_DONE = "progress_done"
    MENU = "menu"

Common Flow Steps

Creating Entries

async_create_entry

Creates a config entry and completes the flow:

return self.async_create_entry(
    title="My Device",
    data={
        CONF_HOST: "192.168.1.100",
        CONF_USERNAME: "admin",
    },
    options={
        "scan_interval": 30,
    },
)

title: Displayed name in the UI
data: Immutable configuration data
options: User-configurable options (can be changed via options flow)

Aborting Flows

async_abort

Stops the flow with a reason:

# Already configured
return self.async_abort(reason="already_configured")

# User cancelled
return self.async_abort(reason="user_cancelled")

# Not supported
return self.async_abort(reason="not_supported")

Unique IDs

Prevent duplicate configurations:

# Set unique ID
await self.async_set_unique_id(device_id)

# Abort if already configured
self._abort_if_unique_id_configured()

# Update host if changed
self._abort_if_unique_id_configured(
    updates={CONF_HOST: discovery_info.host}
)

Multi-Step Flows

Complex configuration can span multiple steps:

class MyConfigFlow(ConfigFlow, domain=DOMAIN):
    """Multi-step config flow."""
    
    def __init__(self) -> None:
        """Initialize."""
        self._device_type: str | None = None
    
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Step 1: Select device type."""
        if user_input is not None:
            self._device_type = user_input["device_type"]
            return await self.async_step_credentials()
        
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required("device_type"): vol.In(["type_a", "type_b"]),
            }),
        )
    
    async def async_step_credentials(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Step 2: Enter credentials."""
        if user_input is not None:
            # Create entry with both steps' data
            return self.async_create_entry(
                title=f"{self._device_type} Device",
                data={
                    "device_type": self._device_type,
                    **user_input,
                },
            )
        
        return self.async_show_form(
            step_id="credentials",
            data_schema=vol.Schema({
                vol.Required(CONF_USERNAME): cv.string,
                vol.Required(CONF_PASSWORD): cv.string,
            }),
        )

Options Flow

Allow users to modify options after configuration:

class MyOptionsFlow(OptionsFlow):
    """Handle options."""
    
    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Manage options."""
        if user_input is not None:
            return self.async_create_entry(title="", data=user_input)
        
        return self.async_show_form(
            step_id="init",
            data_schema=vol.Schema({
                vol.Optional(
                    "scan_interval",
                    default=self.config_entry.options.get("scan_interval", 60),
                ): cv.positive_int,
            }),
        )


class MyConfigFlow(ConfigFlow, domain=DOMAIN):
    """Config flow with options."""
    
    @staticmethod
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> MyOptionsFlow:
        """Get the options flow."""
        return MyOptionsFlow(config_entry)

Discovery Sources

Different discovery methods have specific steps:

async def async_step_zeroconf(
    self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
    """Handle zeroconf discovery."""
    await self.async_set_unique_id(discovery_info.properties["id"])
    self._abort_if_unique_id_configured()
    return await self.async_step_confirm()

async def async_step_ssdp(
    self, discovery_info: SsdpServiceInfo
) -> ConfigFlowResult:
    """Handle SSDP discovery."""
    await self.async_set_unique_id(discovery_info.upnp["UDN"])
    self._abort_if_unique_id_configured()
    return await self.async_step_confirm()

async def async_step_bluetooth(
    self, discovery_info: BluetoothServiceInfoBleak
) -> ConfigFlowResult:
    """Handle bluetooth discovery."""
    await self.async_set_unique_id(discovery_info.address)
    self._abort_if_unique_id_configured()
    return await self.async_step_confirm()

Error Handling

Display errors to users:

errors = {}

try:
    await self._validate_input(user_input)
except ValueError:
    errors["base"] = "invalid_value"
except ConnectionError:
    errors["base"] = "cannot_connect"
except Exception:
    _LOGGER.exception("Unexpected exception")
    errors["base"] = "unknown"

if errors:
    return self.async_show_form(
        step_id="user",
        data_schema=data_schema,
        errors=errors,
    )

Define error messages in strings.json:

strings.json

{
  "config": {
    "error": {
      "cannot_connect": "Failed to connect to device",
      "invalid_value": "Invalid configuration value",
      "unknown": "Unexpected error occurred"
    }
  }
}

Progress Steps

Show progress for long-running operations:

async def async_step_user(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Handle user step."""
    if user_input is not None:
        # Show progress
        return self.async_show_progress(
            step_id="user",
            progress_action="connecting",
        )
    
    return self.async_show_form(
        step_id="user",
        data_schema=data_schema,
    )


async def async_step_connecting(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Handle connection."""
    try:
        await self._connect()
    except Exception as err:
        return self.async_show_progress_done(next_step_id="error")
    
    return self.async_show_progress_done(next_step_id="finish")

Provide multiple options:

async def async_step_user(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Show menu."""
    return self.async_show_menu(
        step_id="user",
        menu_options=["manual", "automatic"],
    )


async def async_step_manual(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Manual configuration."""
    # Manual setup
    pass


async def async_step_automatic(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Automatic discovery."""
    # Auto discovery
    pass

Real-World Example: MQTT Config Flow

From homeassistant/components/mqtt/config_flow.py:66:

class MqttFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for MQTT."""
    
    VERSION = CONFIG_ENTRY_VERSION
    MINOR_VERSION = CONFIG_ENTRY_MINOR_VERSION

The MQTT integration has a comprehensive config flow supporting:

Manual configuration
Hassio discovery
Reconfiguration
Options flow for discovery settings

Best Practices

Security: Never log passwords or sensitive data during validation.

Validate Input

async def _validate_input(self, data: dict[str, Any]) -> None:
    """Validate user input."""
    # Test connection
    api = MyAPI(data[CONF_HOST])
    await api.authenticate(data[CONF_USERNAME], data[CONF_PASSWORD])
    
    # Validate data
    if not await api.validate():
        raise ValueError("Invalid configuration")

Use Selectors

From config_flow.py:121, use modern selectors:

from homeassistant.helpers.selector import (
    TextSelector,
    TextSelectorConfig,
    TextSelectorType,
)

data_schema = vol.Schema({
    vol.Required(CONF_HOST): TextSelector(
        TextSelectorConfig(type=TextSelectorType.TEXT)
    ),
    vol.Required(CONF_PASSWORD): TextSelector(
        TextSelectorConfig(type=TextSelectorType.PASSWORD)
    ),
})

Handle Reauth Properly

# Trigger reauth from integration
self.hass.config_entries.async_schedule_reauth(
    config_entry=self.config_entry,
)

Getting Started

Core Architecture

Building Integrations

Core Concepts

Development

Config Flow

Config Flow

Overview

Basic Structure

Flow Result Types

Common Flow Steps

Creating Entries

async_create_entry

Aborting Flows

async_abort

Unique IDs

Multi-Step Flows

Options Flow

Discovery Sources

Error Handling

Progress Steps

Menu Steps

Real-World Example: MQTT Config Flow

Best Practices

Validate Input

Use Selectors

Handle Reauth Properly

Next Steps

Entity Platforms

Creating Components

​Config Flow

​Overview

​Basic Structure

​Flow Result Types

​Common Flow Steps

​Creating Entries

​async_create_entry

​Aborting Flows

​async_abort

​Unique IDs

​Multi-Step Flows

​Options Flow

​Discovery Sources

​Error Handling

​Progress Steps

​Menu Steps

​Real-World Example: MQTT Config Flow

​Best Practices

​Validate Input

​Use Selectors

​Handle Reauth Properly

​Next Steps

Entity Platforms

Creating Components

Config Flow

Overview

Basic Structure

Flow Result Types

Common Flow Steps

Creating Entries

async_create_entry

Aborting Flows

async_abort

Unique IDs

Multi-Step Flows

Options Flow

Discovery Sources

Error Handling

Progress Steps

Menu Steps

Real-World Example: MQTT Config Flow

Best Practices

Validate Input

Use Selectors

Handle Reauth Properly

Next Steps