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

async_step_user

The initial step when users manually add the integration:

async def async_step_user(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Handle user initiated configuration."""
    if user_input is None:
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required(CONF_HOST): cv.string,
            }),
        )
    
    # Validate and create entry
    return self.async_create_entry(
        title=user_input[CONF_HOST],
        data=user_input,
    )

async_step_discovery

Called when the integration is discovered:

async def async_step_zeroconf(
    self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
    """Handle zeroconf discovery."""
    # Extract info from discovery
    host = discovery_info.host
    
    # Check if already configured
    await self.async_set_unique_id(discovery_info.properties["id"])
    self._abort_if_unique_id_configured()
    
    # Continue to user confirmation
    return await self.async_step_confirm()

async_step_reauth

Called when authentication fails and needs updating:

async def async_step_reauth(
    self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
    """Handle reauth."""
    return await self.async_step_reauth_confirm()


async def async_step_reauth_confirm(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Confirm reauth."""
    if user_input is None:
        return self.async_show_form(
            step_id="reauth_confirm",
            data_schema=vol.Schema({
                vol.Required(CONF_PASSWORD): cv.string,
            }),
        )
    
    # Update existing entry
    entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
    self.hass.config_entries.async_update_entry(
        entry,
        data={**entry.data, CONF_PASSWORD: user_input[CONF_PASSWORD]},
    )
    
    return self.async_abort(reason="reauth_successful")

async_step_reconfigure

Allows users to modify configuration:

async def async_step_reconfigure(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Handle reconfiguration."""
    entry = self.hass.config_entries.async_get_entry(
        self.context["entry_id"]
    )
    
    if user_input is not None:
        # Update entry
        self.hass.config_entries.async_update_entry(
            entry,
            data=user_input,
        )
        return self.async_abort(reason="reconfigure_successful")
    
    # Show form with current values
    return self.async_show_form(
        step_id="reconfigure",
        data_schema=vol.Schema({
            vol.Required(CONF_HOST, default=entry.data[CONF_HOST]): cv.string,
        }),
    )

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()

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

Build docs developers (and LLMs) love

Getting Started

Core Architecture

Building Integrations

Core Concepts

Development

​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

Build docs developers (and LLMs) love

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