"""The OpenAI Conversation integration.""" from pathlib import Path from types import MappingProxyType import openai from openai.types.images_response import ImagesResponse from openai.types.responses import ( EasyInputMessageParam, Response, ResponseInputMessageContentListParam, ResponseInputParam, ResponseInputTextParam, ) import voluptuous as vol from homeassistant.config_entries import ConfigEntry, ConfigSubentry from homeassistant.const import CONF_API_KEY, CONF_PROMPT, Platform from homeassistant.core import ( HomeAssistant, ServiceCall, ServiceResponse, SupportsResponse, ) from homeassistant.exceptions import ( ConfigEntryAuthFailed, ConfigEntryNotReady, HomeAssistantError, ServiceValidationError, ) from homeassistant.helpers import ( config_validation as cv, device_registry as dr, entity_registry as er, issue_registry as ir, selector, ) from homeassistant.helpers.httpx_client import get_async_client from homeassistant.helpers.typing import UNDEFINED, ConfigType, UndefinedType from .const import ( CONF_CHAT_MODEL, CONF_FILENAMES, CONF_MAX_TOKENS, CONF_REASONING_EFFORT, CONF_REASONING_SUMMARY, CONF_STORE_RESPONSES, CONF_TEMPERATURE, CONF_TOP_P, DEFAULT_AI_TASK_NAME, DEFAULT_NAME, DEFAULT_STT_NAME, DEFAULT_TTS_NAME, DOMAIN, LOGGER, RECOMMENDED_AI_TASK_OPTIONS, RECOMMENDED_CHAT_MODEL, RECOMMENDED_MAX_TOKENS, RECOMMENDED_REASONING_EFFORT, RECOMMENDED_REASONING_SUMMARY, RECOMMENDED_STORE_RESPONSES, RECOMMENDED_STT_OPTIONS, RECOMMENDED_TEMPERATURE, RECOMMENDED_TOP_P, RECOMMENDED_TTS_OPTIONS, ) from .entity import async_prepare_files_for_prompt SERVICE_GENERATE_IMAGE = "generate_image" SERVICE_GENERATE_CONTENT = "generate_content" PLATFORMS = (Platform.AI_TASK, Platform.CONVERSATION, Platform.STT, Platform.TTS) CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) type OpenAIConfigEntry = ConfigEntry[openai.AsyncClient] async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up OpenAI Conversation.""" await async_migrate_integration(hass) async def render_image(call: ServiceCall) -> ServiceResponse: """Render an image with dall-e.""" LOGGER.warning( "Action '%s.%s' is deprecated and will be removed in the 2026.9.0 release. " "Please use the 'ai_task.generate_image' action instead", DOMAIN, SERVICE_GENERATE_IMAGE, ) ir.async_create_issue( hass, DOMAIN, "deprecated_generate_image", breaks_in_ha_version="2026.9.0", is_fixable=False, severity=ir.IssueSeverity.WARNING, translation_key="deprecated_generate_image", ) entry_id = call.data["config_entry"] entry = hass.config_entries.async_get_entry(entry_id) if entry is None or entry.domain != DOMAIN: raise ServiceValidationError( translation_domain=DOMAIN, translation_key="invalid_config_entry", translation_placeholders={"config_entry": entry_id}, ) client: openai.AsyncClient = entry.runtime_data try: response: ImagesResponse = await client.images.generate( model="dall-e-3", prompt=call.data[CONF_PROMPT], size=call.data["size"], quality=call.data["quality"], style=call.data["style"], response_format="url", n=1, ) except openai.AuthenticationError as err: entry.async_start_reauth(hass) raise HomeAssistantError("Authentication error") from err except openai.OpenAIError as err: raise HomeAssistantError(f"Error generating image: {err}") from err if not response.data or not response.data[0].url: raise HomeAssistantError("No image returned") return response.data[0].model_dump(exclude={"b64_json"}) async def send_prompt(call: ServiceCall) -> ServiceResponse: """Send a prompt to ChatGPT and return the response.""" LOGGER.warning( "Action '%s.%s' is deprecated and will be removed in the 2026.9.0 release. " "Please use the 'ai_task.generate_data' action instead", DOMAIN, SERVICE_GENERATE_CONTENT, ) ir.async_create_issue( hass, DOMAIN, "deprecated_generate_content", breaks_in_ha_version="2026.9.0", is_fixable=False, severity=ir.IssueSeverity.WARNING, translation_key="deprecated_generate_content", ) entry_id = call.data["config_entry"] entry = hass.config_entries.async_get_entry(entry_id) if entry is None or entry.domain != DOMAIN: raise ServiceValidationError( translation_domain=DOMAIN, translation_key="invalid_config_entry", translation_placeholders={"config_entry": entry_id}, ) # Get first conversation subentry for options conversation_subentry = next( ( sub for sub in entry.subentries.values() if sub.subentry_type == "conversation" ), None, ) if not conversation_subentry: raise ServiceValidationError("No conversation configuration found") model: str = conversation_subentry.data.get( CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL ) client: openai.AsyncClient = entry.runtime_data content: ResponseInputMessageContentListParam = [ ResponseInputTextParam(type="input_text", text=call.data[CONF_PROMPT]) ] if filenames := call.data.get(CONF_FILENAMES): for filename in filenames: if not hass.config.is_allowed_path(filename): raise HomeAssistantError( f"Cannot read `{filename}`, no access to path; " "`allowlist_external_dirs` may need to be adjusted in " "`configuration.yaml`" ) content.extend( await async_prepare_files_for_prompt( hass, [(Path(filename), None) for filename in filenames] ) ) messages: ResponseInputParam = [ EasyInputMessageParam(type="message", role="user", content=content) ] model_args = { "model": model, "input": messages, "max_output_tokens": conversation_subentry.data.get( CONF_MAX_TOKENS, RECOMMENDED_MAX_TOKENS ), "top_p": conversation_subentry.data.get(CONF_TOP_P, RECOMMENDED_TOP_P), "temperature": conversation_subentry.data.get( CONF_TEMPERATURE, RECOMMENDED_TEMPERATURE ), "user": call.context.user_id, "store": conversation_subentry.data.get( CONF_STORE_RESPONSES, RECOMMENDED_STORE_RESPONSES ), } if model.startswith("o"): model_args["reasoning"] = { "effort": conversation_subentry.data.get( CONF_REASONING_EFFORT, RECOMMENDED_REASONING_EFFORT ) } try: response: Response = await client.responses.create(**model_args) except openai.AuthenticationError as err: entry.async_start_reauth(hass) raise HomeAssistantError("Authentication error") from err except openai.OpenAIError as err: raise HomeAssistantError(f"Error generating content: {err}") from err except FileNotFoundError as err: raise HomeAssistantError(f"Error generating content: {err}") from err return {"text": response.output_text} hass.services.async_register( DOMAIN, SERVICE_GENERATE_CONTENT, send_prompt, schema=vol.Schema( { vol.Required("config_entry"): selector.ConfigEntrySelector( { "integration": DOMAIN, } ), vol.Required(CONF_PROMPT): cv.string, vol.Optional(CONF_FILENAMES, default=[]): vol.All( cv.ensure_list, [cv.string] ), } ), supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GENERATE_IMAGE, render_image, schema=vol.Schema( { vol.Required("config_entry"): selector.ConfigEntrySelector( { "integration": DOMAIN, } ), vol.Required(CONF_PROMPT): cv.string, vol.Optional("size", default="1024x1024"): vol.In( ("1024x1024", "1024x1792", "1792x1024") ), vol.Optional("quality", default="standard"): vol.In(("standard", "hd")), vol.Optional("style", default="vivid"): vol.In(("vivid", "natural")), } ), supports_response=SupportsResponse.ONLY, ) return True async def async_setup_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> bool: """Set up OpenAI Conversation from a config entry.""" client = openai.AsyncOpenAI( api_key=entry.data[CONF_API_KEY], http_client=get_async_client(hass), ) # Cache current platform data which gets added to each request # (caching done by library) _ = await hass.async_add_executor_job(client.platform_headers) try: await hass.async_add_executor_job(client.with_options(timeout=10.0).models.list) except openai.AuthenticationError as err: raise ConfigEntryAuthFailed(err) from err except openai.OpenAIError as err: raise ConfigEntryNotReady(err) from err entry.runtime_data = client await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) entry.async_on_unload(entry.add_update_listener(async_update_options)) return True async def async_unload_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> bool: """Unload OpenAI.""" return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) async def async_update_options(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None: """Update options.""" await hass.config_entries.async_reload(entry.entry_id) async def async_migrate_integration(hass: HomeAssistant) -> None: """Migrate integration entry structure.""" # Make sure we get enabled config entries first entries = sorted( hass.config_entries.async_entries(DOMAIN), key=lambda e: e.disabled_by is not None, ) if not any(entry.version == 1 for entry in entries): return api_keys_entries: dict[str, tuple[OpenAIConfigEntry, bool]] = {} entity_registry = er.async_get(hass) device_registry = dr.async_get(hass) for entry in entries: use_existing = False subentry = ConfigSubentry( data=entry.options, subentry_type="conversation", title=entry.title, unique_id=None, ) if entry.data[CONF_API_KEY] not in api_keys_entries: use_existing = True all_disabled = all( e.disabled_by is not None for e in entries if e.data[CONF_API_KEY] == entry.data[CONF_API_KEY] ) api_keys_entries[entry.data[CONF_API_KEY]] = (entry, all_disabled) parent_entry, all_disabled = api_keys_entries[entry.data[CONF_API_KEY]] hass.config_entries.async_add_subentry(parent_entry, subentry) conversation_entity_id = entity_registry.async_get_entity_id( "conversation", DOMAIN, entry.entry_id, ) device = device_registry.async_get_device_by_identifier( (DOMAIN, entry.entry_id), entry.entry_id ) if conversation_entity_id is not None: conversation_entity_entry = entity_registry.entities[conversation_entity_id] entity_disabled_by = conversation_entity_entry.disabled_by if ( entity_disabled_by is er.RegistryEntryDisabler.CONFIG_ENTRY and not all_disabled ): # Device and entity registries will set the disabled_by flag to None # when moving a device or entity disabled by CONFIG_ENTRY to an enabled # config entry, but we want to set it to DEVICE or USER instead, entity_disabled_by = ( er.RegistryEntryDisabler.DEVICE if device else er.RegistryEntryDisabler.USER ) entity_registry.async_update_entity( conversation_entity_id, config_entry_id=parent_entry.entry_id, config_subentry_id=subentry.subentry_id, disabled_by=entity_disabled_by, new_unique_id=subentry.subentry_id, ) if device is not None: # Device and entity registries will set the disabled_by flag to None # when moving a device or entity disabled by CONFIG_ENTRY to an enabled # config entry, but we want to set it to USER instead, device_disabled_by: dr.DeviceEntryDisabler | UndefinedType = UNDEFINED if ( device.disabled_by is dr.DeviceEntryDisabler.CONFIG_ENTRY and not all_disabled ): device_disabled_by = dr.DeviceEntryDisabler.USER device_registry.async_update_device( device.id, disabled_by=device_disabled_by, new_identifiers={(DOMAIN, subentry.subentry_id)}, new_config_entry_id=parent_entry.entry_id, new_config_subentry_id=subentry.subentry_id, ) if not use_existing: await hass.config_entries.async_remove(entry.entry_id) else: _add_ai_task_subentry(hass, entry) hass.config_entries.async_update_entry( entry, title=DEFAULT_NAME, options={}, version=2, minor_version=4, ) async def async_migrate_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> bool: """Migrate entry.""" LOGGER.debug("Migrating from version %s:%s", entry.version, entry.minor_version) if entry.version == 2 and entry.minor_version == 1: # Devices left in both the config entry and its subentry by Home Assistant Core # 2025.7.0b0-2025.7.0b1 are collapsed onto the subentry by the device registry # migration, so there's nothing to correct here. hass.config_entries.async_update_entry(entry, minor_version=2) if entry.version == 2 and entry.minor_version == 2: _add_ai_task_subentry(hass, entry) hass.config_entries.async_update_entry(entry, minor_version=3) if entry.version == 2 and entry.minor_version == 3: # Fix migration where the disabled_by flag was not set correctly. # We can currently only correct this for enabled config entries, # because migration does not run for disabled config entries. This # is asserted in tests, and if that behavior is changed, we should # correct also disabled config entries. device_registry = dr.async_get(hass) entity_registry = er.async_get(hass) devices = dr.async_entries_for_config_entry(device_registry, entry.entry_id) entity_entries = er.async_entries_for_config_entry( entity_registry, entry.entry_id ) if entry.disabled_by is None: # If the config entry is not disabled, we need to set the disabled_by # flag on devices to USER, and on entities to DEVICE, if they are set # to CONFIG_ENTRY. for device in devices: if device.disabled_by is not dr.DeviceEntryDisabler.CONFIG_ENTRY: continue device_registry.async_update_device( device.id, disabled_by=dr.DeviceEntryDisabler.USER, ) for entity in entity_entries: if entity.disabled_by is not er.RegistryEntryDisabler.CONFIG_ENTRY: continue entity_registry.async_update_entity( entity.entity_id, disabled_by=er.RegistryEntryDisabler.DEVICE, ) hass.config_entries.async_update_entry(entry, minor_version=4) if entry.version == 2 and entry.minor_version == 4: _add_tts_subentry(hass, entry) hass.config_entries.async_update_entry(entry, minor_version=5) if entry.version == 2 and entry.minor_version == 5: _add_stt_subentry(hass, entry) hass.config_entries.async_update_entry(entry, minor_version=6) if entry.version == 2 and entry.minor_version == 6: for subentry in entry.subentries.values(): if subentry.subentry_type in ("conversation", "ai_task_data"): data = dict(subentry.data) updated = False if data.get(CONF_REASONING_SUMMARY) == "short": data[CONF_REASONING_SUMMARY] = "concise" updated = True if data.get(CONF_REASONING_SUMMARY) == "concise" and not data.get( CONF_CHAT_MODEL, "" ).startswith("gpt-5"): data[CONF_REASONING_SUMMARY] = RECOMMENDED_REASONING_SUMMARY updated = True if updated: hass.config_entries.async_update_subentry( entry, subentry, data=data ) hass.config_entries.async_update_entry(entry, minor_version=7) LOGGER.debug( "Migration to version %s:%s successful", entry.version, entry.minor_version ) return True def _add_ai_task_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None: """Add AI Task subentry to the config entry.""" hass.config_entries.async_add_subentry( entry, ConfigSubentry( data=MappingProxyType(RECOMMENDED_AI_TASK_OPTIONS), subentry_type="ai_task_data", title=DEFAULT_AI_TASK_NAME, unique_id=None, ), ) def _add_stt_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None: """Add STT subentry to the config entry.""" hass.config_entries.async_add_subentry( entry, ConfigSubentry( data=MappingProxyType(RECOMMENDED_STT_OPTIONS), subentry_type="stt", title=DEFAULT_STT_NAME, unique_id=None, ), ) def _add_tts_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None: """Add TTS subentry to the config entry.""" hass.config_entries.async_add_subentry( entry, ConfigSubentry( data=MappingProxyType(RECOMMENDED_TTS_OPTIONS), subentry_type="tts", title=DEFAULT_TTS_NAME, unique_id=None, ), )