From 187fcb895f502ed40f9b3b5000e1ca89f32f9403 Mon Sep 17 00:00:00 2001 From: Robot Date: Thu, 26 Feb 2026 20:00:05 +0300 Subject: [PATCH] Initial commit: Anthropic integration with custom base_url support --- README.md | 28 + .../anthropic_custom/__init__.py | 259 +++++ custom_components/anthropic_custom/ai_task.py | 87 ++ .../anthropic_custom/config_flow.py | 569 +++++++++++ custom_components/anthropic_custom/const.py | 78 ++ .../anthropic_custom/conversation.py | 76 ++ custom_components/anthropic_custom/entity.py | 924 ++++++++++++++++++ custom_components/anthropic_custom/icons.json | 15 + .../anthropic_custom/manifest.json | 13 + .../anthropic_custom/quality_scale.yaml | 109 +++ custom_components/anthropic_custom/repairs.py | 194 ++++ .../anthropic_custom/strings.json | 185 ++++ 12 files changed, 2537 insertions(+) create mode 100644 README.md create mode 100644 custom_components/anthropic_custom/__init__.py create mode 100644 custom_components/anthropic_custom/ai_task.py create mode 100644 custom_components/anthropic_custom/config_flow.py create mode 100644 custom_components/anthropic_custom/const.py create mode 100644 custom_components/anthropic_custom/conversation.py create mode 100644 custom_components/anthropic_custom/entity.py create mode 100644 custom_components/anthropic_custom/icons.json create mode 100644 custom_components/anthropic_custom/manifest.json create mode 100644 custom_components/anthropic_custom/quality_scale.yaml create mode 100644 custom_components/anthropic_custom/repairs.py create mode 100644 custom_components/anthropic_custom/strings.json diff --git a/README.md b/README.md new file mode 100644 index 0000000..b7c7e05 --- /dev/null +++ b/README.md @@ -0,0 +1,28 @@ +# ha-anthropic-custom + +A custom Home Assistant integration forked from the official [Anthropic integration](https://github.com/home-assistant/core/tree/dev/homeassistant/components/anthropic), with added support for a **custom `base_url`** parameter. + +## Features + +All features from the official integration, plus: + +- **Custom Base URL**: Configure a custom API endpoint (e.g., a proxy, a compatible LLM API, or a self-hosted endpoint) via the `base_url` field in the config flow. + +## Installation + +1. Copy the `custom_components/anthropic_custom` folder to your Home Assistant `custom_components` directory. +2. Restart Home Assistant. +3. Go to **Settings → Integrations → Add Integration** and search for **Anthropic (Custom Base URL)**. +4. Enter your API key and optionally a custom `base_url`. + +## Configuration + +| Field | Required | Description | +|-------|----------|-------------| +| `api_key` | Yes | Your Anthropic API key | +| `base_url` | No | Custom base URL for the Anthropic API (e.g. `https://my-proxy.example.com`) | + +## Notes + +- If `base_url` is left empty, the default Anthropic API URL is used. +- This integration uses domain `anthropic_custom` and will not conflict with the official `anthropic` integration. diff --git a/custom_components/anthropic_custom/__init__.py b/custom_components/anthropic_custom/__init__.py new file mode 100644 index 0000000..ac2c001 --- /dev/null +++ b/custom_components/anthropic_custom/__init__.py @@ -0,0 +1,259 @@ +"""The Anthropic integration.""" + +from __future__ import annotations + +import anthropic + +from homeassistant.config_entries import ConfigEntry, ConfigSubentry +from homeassistant.const import CONF_API_KEY, Platform +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady +from homeassistant.helpers import ( + config_validation as cv, + device_registry as dr, + entity_registry as er, + issue_registry as ir, +) +from homeassistant.helpers.httpx_client import get_async_client +from homeassistant.helpers.typing import ConfigType + +from .const import ( + CONF_BASE_URL, + CONF_CHAT_MODEL, + DEFAULT_CONVERSATION_NAME, + DEPRECATED_MODELS, + DOMAIN, + LOGGER, +) + +PLATFORMS = (Platform.AI_TASK, Platform.CONVERSATION) +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) + +type AnthropicConfigEntry = ConfigEntry[anthropic.AsyncClient] + + +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + """Set up Anthropic.""" + await async_migrate_integration(hass) + return True + + +async def async_setup_entry(hass: HomeAssistant, entry: AnthropicConfigEntry) -> bool: + """Set up Anthropic from a config entry.""" + client = anthropic.AsyncAnthropic( + api_key=entry.data[CONF_API_KEY], + base_url=entry.data.get(CONF_BASE_URL) or None, + http_client=get_async_client(hass) + ) + try: + await client.models.list(timeout=10.0) + except anthropic.AuthenticationError as err: + raise ConfigEntryAuthFailed(err) from err + except anthropic.AnthropicError 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)) + + for subentry in entry.subentries.values(): + if (model := subentry.data.get(CONF_CHAT_MODEL)) and model.startswith( + tuple(DEPRECATED_MODELS) + ): + ir.async_create_issue( + hass, + DOMAIN, + "model_deprecated", + is_fixable=True, + is_persistent=False, + learn_more_url="https://platform.claude.com/docs/en/about-claude/model-deprecations", + severity=ir.IssueSeverity.WARNING, + translation_key="model_deprecated", + ) + break + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: AnthropicConfigEntry) -> bool: + """Unload Anthropic.""" + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + +async def async_update_options( + hass: HomeAssistant, entry: AnthropicConfigEntry +) -> 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[AnthropicConfigEntry, 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( + identifiers={(DOMAIN, 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 = device.disabled_by + 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)}, + add_config_subentry_id=subentry.subentry_id, + add_config_entry_id=parent_entry.entry_id, + ) + if parent_entry.entry_id != entry.entry_id: + device_registry.async_update_device( + device.id, + remove_config_entry_id=entry.entry_id, + ) + else: + device_registry.async_update_device( + device.id, + remove_config_entry_id=entry.entry_id, + remove_config_subentry_id=None, + ) + + if not use_existing: + await hass.config_entries.async_remove(entry.entry_id) + else: + hass.config_entries.async_update_entry( + entry, + title=DEFAULT_CONVERSATION_NAME, + options={}, + version=2, + minor_version=3, + ) + + +async def async_migrate_entry(hass: HomeAssistant, entry: AnthropicConfigEntry) -> bool: + """Migrate entry.""" + LOGGER.debug("Migrating from version %s:%s", entry.version, entry.minor_version) + + if entry.version > 2: + # This means the user has downgraded from a future version + return False + + if entry.version == 2 and entry.minor_version == 1: + # Correct broken device migration in Home Assistant Core 2025.7.0b0-2025.7.0b1 + device_registry = dr.async_get(hass) + for device in dr.async_entries_for_config_entry( + device_registry, entry.entry_id + ): + device_registry.async_update_device( + device.id, + remove_config_entry_id=entry.entry_id, + remove_config_subentry_id=None, + ) + + hass.config_entries.async_update_entry(entry, minor_version=2) + + if entry.version == 2 and entry.minor_version == 2: + # 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=3) + + LOGGER.debug( + "Migration to version %s:%s successful", entry.version, entry.minor_version + ) + + return True + diff --git a/custom_components/anthropic_custom/ai_task.py b/custom_components/anthropic_custom/ai_task.py new file mode 100644 index 0000000..56b9ea4 --- /dev/null +++ b/custom_components/anthropic_custom/ai_task.py @@ -0,0 +1,87 @@ +"""AI Task integration for Anthropic.""" + +from __future__ import annotations + +from json import JSONDecodeError +import logging +from typing import TYPE_CHECKING + +from homeassistant.components import ai_task, conversation +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from homeassistant.util.json import json_loads + +from .entity import AnthropicBaseLLMEntity + +if TYPE_CHECKING: + from . import AnthropicConfigEntry + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: AnthropicConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up AI Task entities.""" + for subentry in config_entry.subentries.values(): + if subentry.subentry_type != "ai_task_data": + continue + + async_add_entities( + [AnthropicTaskEntity(config_entry, subentry)], + config_subentry_id=subentry.subentry_id, + ) + + +class AnthropicTaskEntity( + ai_task.AITaskEntity, + AnthropicBaseLLMEntity, +): + """Anthropic AI Task entity.""" + + _attr_supported_features = ( + ai_task.AITaskEntityFeature.GENERATE_DATA + | ai_task.AITaskEntityFeature.SUPPORT_ATTACHMENTS + ) + _attr_translation_key = "ai_task_data" + + async def _async_generate_data( + self, + task: ai_task.GenDataTask, + chat_log: conversation.ChatLog, + ) -> ai_task.GenDataTaskResult: + """Handle a generate data task.""" + await self._async_handle_chat_log( + chat_log, task.name, task.structure, max_iterations=1000 + ) + + if not isinstance(chat_log.content[-1], conversation.AssistantContent): + raise HomeAssistantError( + "Last content in chat log is not an AssistantContent" + ) + + text = chat_log.content[-1].content or "" + + if not task.structure: + return ai_task.GenDataTaskResult( + conversation_id=chat_log.conversation_id, + data=text, + ) + try: + data = json_loads(text) + except JSONDecodeError as err: + _LOGGER.error( + "Failed to parse JSON response: %s. Response: %s", + err, + text, + ) + raise HomeAssistantError("Error with Claude structured response") from err + + return ai_task.GenDataTaskResult( + conversation_id=chat_log.conversation_id, + data=data, + ) + diff --git a/custom_components/anthropic_custom/config_flow.py b/custom_components/anthropic_custom/config_flow.py new file mode 100644 index 0000000..c754da6 --- /dev/null +++ b/custom_components/anthropic_custom/config_flow.py @@ -0,0 +1,569 @@ +"""Config flow for Anthropic integration.""" + +from __future__ import annotations + +from collections.abc import Mapping +import json +import logging +import re +from typing import TYPE_CHECKING, Any, cast + +import anthropic +import voluptuous as vol +from voluptuous_openapi import convert + +from homeassistant.components.zone import ENTITY_ID_HOME +from homeassistant.config_entries import ( + SOURCE_REAUTH, + ConfigEntryState, + ConfigFlow, + ConfigFlowResult, + ConfigSubentryFlow, + SubentryFlowResult, +) +from homeassistant.const import ( + ATTR_LATITUDE, + ATTR_LONGITUDE, + CONF_API_KEY, + CONF_LLM_HASS_API, + CONF_NAME, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import llm +from homeassistant.helpers.httpx_client import get_async_client +from homeassistant.helpers.selector import ( + NumberSelector, + NumberSelectorConfig, + SelectOptionDict, + SelectSelector, + SelectSelectorConfig, + SelectSelectorMode, + TemplateSelector, +) +from homeassistant.helpers.typing import VolDictType + +from .const import ( + CODE_EXECUTION_UNSUPPORTED_MODELS, + CONF_BASE_URL, + CONF_CHAT_MODEL, + CONF_CODE_EXECUTION, + CONF_MAX_TOKENS, + CONF_PROMPT, + CONF_RECOMMENDED, + CONF_TEMPERATURE, + CONF_THINKING_BUDGET, + CONF_THINKING_EFFORT, + CONF_WEB_SEARCH, + CONF_WEB_SEARCH_CITY, + CONF_WEB_SEARCH_COUNTRY, + CONF_WEB_SEARCH_MAX_USES, + CONF_WEB_SEARCH_REGION, + CONF_WEB_SEARCH_TIMEZONE, + CONF_WEB_SEARCH_USER_LOCATION, + DEFAULT, + DEFAULT_AI_TASK_NAME, + DEFAULT_CONVERSATION_NAME, + DOMAIN, + NON_ADAPTIVE_THINKING_MODELS, + NON_THINKING_MODELS, + WEB_SEARCH_UNSUPPORTED_MODELS, +) + +if TYPE_CHECKING: + from . import AnthropicConfigEntry + +_LOGGER = logging.getLogger(__name__) + +STEP_USER_DATA_SCHEMA = vol.Schema( + { + vol.Required(CONF_API_KEY): str, + vol.Optional(CONF_BASE_URL): str, + } +) + +DEFAULT_CONVERSATION_OPTIONS = { + CONF_RECOMMENDED: True, + CONF_LLM_HASS_API: [llm.LLM_API_ASSIST], + CONF_PROMPT: llm.DEFAULT_INSTRUCTIONS_PROMPT, +} + +DEFAULT_AI_TASK_OPTIONS = { + CONF_RECOMMENDED: True, +} + + +async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None: + """Validate the user input allows us to connect. + + Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user. + """ + client = anthropic.AsyncAnthropic( + api_key=data[CONF_API_KEY], + base_url=data.get(CONF_BASE_URL) or None, + http_client=get_async_client(hass) + ) + await client.models.list(timeout=10.0) + + +async def get_model_list(client: anthropic.AsyncAnthropic) -> list[SelectOptionDict]: + """Get list of available models.""" + try: + models = (await client.models.list()).data + except anthropic.AnthropicError: + models = [] + _LOGGER.debug("Available models: %s", models) + model_options: list[SelectOptionDict] = [] + short_form = re.compile(r"[^\d]-\d$") + for model_info in models: + # Resolve alias from versioned model name: + model_alias = ( + model_info.id[:-9] + if model_info.id != "claude-3-haiku-20240307" + and model_info.id[-2:-1] != "-" + else model_info.id + ) + if short_form.search(model_alias): + model_alias += "-0" + model_options.append( + SelectOptionDict( + label=model_info.display_name, + value=model_alias, + ) + ) + return model_options + + +class AnthropicConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Anthropic.""" + + VERSION = 2 + MINOR_VERSION = 3 + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the initial step.""" + errors: dict[str, str] = {} + + if user_input is not None: + self._async_abort_entries_match(user_input) + try: + await validate_input(self.hass, user_input) + except anthropic.APITimeoutError: + errors["base"] = "timeout_connect" + except anthropic.APIConnectionError: + errors["base"] = "cannot_connect" + except anthropic.APIStatusError as e: + errors["base"] = "unknown" + if ( + isinstance(e.body, dict) + and (error := e.body.get("error")) + and error.get("type") == "authentication_error" + ): + errors["base"] = "authentication_error" + except Exception: + _LOGGER.exception("Unexpected exception") + errors["base"] = "unknown" + else: + if self.source == SOURCE_REAUTH: + return self.async_update_reload_and_abort( + self._get_reauth_entry(), data_updates=user_input + ) + return self.async_create_entry( + title="Claude", + data=user_input, + subentries=[ + { + "subentry_type": "conversation", + "data": DEFAULT_CONVERSATION_OPTIONS, + "title": DEFAULT_CONVERSATION_NAME, + "unique_id": None, + }, + { + "subentry_type": "ai_task_data", + "data": DEFAULT_AI_TASK_OPTIONS, + "title": DEFAULT_AI_TASK_NAME, + "unique_id": None, + }, + ], + ) + + return self.async_show_form( + step_id="user", + data_schema=STEP_USER_DATA_SCHEMA, + errors=errors or None, + description_placeholders={ + "instructions_url": "https://www.home-assistant.io/integrations/anthropic/#generating-an-api-key", + }, + ) + + async def async_step_reauth( + self, entry_data: Mapping[str, Any] + ) -> ConfigFlowResult: + """Perform reauth upon an API authentication error.""" + return await self.async_step_reauth_confirm() + + async def async_step_reauth_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Dialog that informs the user that reauth is required.""" + if not user_input: + return self.async_show_form( + step_id="reauth_confirm", data_schema=STEP_USER_DATA_SCHEMA + ) + return await self.async_step_user(user_input) + + @classmethod + @callback + def async_get_supported_subentry_types( + cls, config_entry: AnthropicConfigEntry + ) -> dict[str, type[ConfigSubentryFlow]]: + """Return subentries supported by this integration.""" + return { + "conversation": ConversationSubentryFlowHandler, + "ai_task_data": ConversationSubentryFlowHandler, + } + + +class ConversationSubentryFlowHandler(ConfigSubentryFlow): + """Flow for managing conversation subentries.""" + + options: dict[str, Any] + + @property + def _is_new(self) -> bool: + """Return if this is a new subentry.""" + return self.source == "user" + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> SubentryFlowResult: + """Add a subentry.""" + if self._subentry_type == "ai_task_data": + self.options = DEFAULT_AI_TASK_OPTIONS.copy() + else: + self.options = DEFAULT_CONVERSATION_OPTIONS.copy() + return await self.async_step_init() + + async def async_step_reconfigure( + self, user_input: dict[str, Any] | None = None + ) -> SubentryFlowResult: + """Handle reconfiguration of a subentry.""" + self.options = self._get_reconfigure_subentry().data.copy() + return await self.async_step_init() + + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> SubentryFlowResult: + """Set initial options.""" + # abort if entry is not loaded + if self._get_entry().state != ConfigEntryState.LOADED: + return self.async_abort(reason="entry_not_loaded") + + hass_apis: list[SelectOptionDict] = [ + SelectOptionDict( + label=api.name, + value=api.id, + ) + for api in llm.async_get_apis(self.hass) + ] + if suggested_llm_apis := self.options.get(CONF_LLM_HASS_API): + if isinstance(suggested_llm_apis, str): + suggested_llm_apis = [suggested_llm_apis] + known_apis = {api.id for api in llm.async_get_apis(self.hass)} + self.options[CONF_LLM_HASS_API] = [ + api for api in suggested_llm_apis if api in known_apis + ] + + step_schema: VolDictType = {} + errors: dict[str, str] = {} + + if self._is_new: + if self._subentry_type == "ai_task_data": + default_name = DEFAULT_AI_TASK_NAME + else: + default_name = DEFAULT_CONVERSATION_NAME + step_schema[vol.Required(CONF_NAME, default=default_name)] = str + + if self._subentry_type == "conversation": + step_schema.update( + { + vol.Optional(CONF_PROMPT): TemplateSelector(), + vol.Optional( + CONF_LLM_HASS_API, + ): SelectSelector( + SelectSelectorConfig(options=hass_apis, multiple=True) + ), + } + ) + + step_schema[ + vol.Required( + CONF_RECOMMENDED, default=self.options.get(CONF_RECOMMENDED, False) + ) + ] = bool + + if user_input is not None: + if not user_input.get(CONF_LLM_HASS_API): + user_input.pop(CONF_LLM_HASS_API, None) + + if user_input[CONF_RECOMMENDED]: + if not errors: + if self._is_new: + return self.async_create_entry( + title=user_input.pop(CONF_NAME), + data=user_input, + ) + return self.async_update_and_abort( + self._get_entry(), + self._get_reconfigure_subentry(), + data=user_input, + ) + else: + self.options.update(user_input) + if ( + CONF_LLM_HASS_API in self.options + and CONF_LLM_HASS_API not in user_input + ): + self.options.pop(CONF_LLM_HASS_API) + if not errors: + return await self.async_step_advanced() + + return self.async_show_form( + step_id="init", + data_schema=self.add_suggested_values_to_schema( + vol.Schema(step_schema), self.options + ), + errors=errors or None, + ) + + async def async_step_advanced( + self, user_input: dict[str, Any] | None = None + ) -> SubentryFlowResult: + """Manage advanced options.""" + errors: dict[str, str] = {} + + step_schema: VolDictType = { + vol.Optional( + CONF_CHAT_MODEL, + default=DEFAULT[CONF_CHAT_MODEL], + ): SelectSelector( + SelectSelectorConfig( + options=await self._get_model_list(), custom_value=True + ) + ), + vol.Optional( + CONF_MAX_TOKENS, + default=DEFAULT[CONF_MAX_TOKENS], + ): int, + vol.Optional( + CONF_TEMPERATURE, + default=DEFAULT[CONF_TEMPERATURE], + ): NumberSelector(NumberSelectorConfig(min=0, max=1, step=0.05)), + } + + if user_input is not None: + self.options.update(user_input) + + if not errors: + return await self.async_step_model() + + return self.async_show_form( + step_id="advanced", + data_schema=self.add_suggested_values_to_schema( + vol.Schema(step_schema), self.options + ), + errors=errors, + ) + + async def async_step_model( + self, user_input: dict[str, Any] | None = None + ) -> SubentryFlowResult: + """Manage model-specific options.""" + errors: dict[str, str] = {} + + step_schema: VolDictType = {} + + model = self.options[CONF_CHAT_MODEL] + + if not model.startswith(tuple(NON_THINKING_MODELS)) and model.startswith( + tuple(NON_ADAPTIVE_THINKING_MODELS) + ): + step_schema[ + vol.Optional( + CONF_THINKING_BUDGET, default=DEFAULT[CONF_THINKING_BUDGET] + ) + ] = vol.All( + NumberSelector( + NumberSelectorConfig( + min=0, + max=self.options.get(CONF_MAX_TOKENS, DEFAULT[CONF_MAX_TOKENS]), + ) + ), + vol.Coerce(int), + ) + else: + self.options.pop(CONF_THINKING_BUDGET, None) + + if not model.startswith(tuple(NON_ADAPTIVE_THINKING_MODELS)): + step_schema[ + vol.Optional( + CONF_THINKING_EFFORT, + default=DEFAULT[CONF_THINKING_EFFORT], + ) + ] = SelectSelector( + SelectSelectorConfig( + options=["none", "low", "medium", "high", "max"], + translation_key=CONF_THINKING_EFFORT, + mode=SelectSelectorMode.DROPDOWN, + ) + ) + else: + self.options.pop(CONF_THINKING_EFFORT, None) + + if not model.startswith(tuple(CODE_EXECUTION_UNSUPPORTED_MODELS)): + step_schema[ + vol.Optional( + CONF_CODE_EXECUTION, + default=DEFAULT[CONF_CODE_EXECUTION], + ) + ] = bool + else: + self.options.pop(CONF_CODE_EXECUTION, None) + + if not model.startswith(tuple(WEB_SEARCH_UNSUPPORTED_MODELS)): + step_schema.update( + { + vol.Optional( + CONF_WEB_SEARCH, + default=DEFAULT[CONF_WEB_SEARCH], + ): bool, + vol.Optional( + CONF_WEB_SEARCH_MAX_USES, + default=DEFAULT[CONF_WEB_SEARCH_MAX_USES], + ): int, + vol.Optional( + CONF_WEB_SEARCH_USER_LOCATION, + default=DEFAULT[CONF_WEB_SEARCH_USER_LOCATION], + ): bool, + } + ) + else: + self.options.pop(CONF_WEB_SEARCH, None) + self.options.pop(CONF_WEB_SEARCH_MAX_USES, None) + self.options.pop(CONF_WEB_SEARCH_USER_LOCATION, None) + + self.options.pop(CONF_WEB_SEARCH_CITY, None) + self.options.pop(CONF_WEB_SEARCH_REGION, None) + self.options.pop(CONF_WEB_SEARCH_COUNTRY, None) + self.options.pop(CONF_WEB_SEARCH_TIMEZONE, None) + + if not step_schema: + user_input = {} + + if user_input is not None: + if user_input.get(CONF_WEB_SEARCH, DEFAULT[CONF_WEB_SEARCH]) and not errors: + if user_input.get( + CONF_WEB_SEARCH_USER_LOCATION, + DEFAULT[CONF_WEB_SEARCH_USER_LOCATION], + ): + user_input.update(await self._get_location_data()) + + self.options.update(user_input) + + if not errors: + if self._is_new: + return self.async_create_entry( + title=self.options.pop(CONF_NAME), + data=self.options, + ) + + return self.async_update_and_abort( + self._get_entry(), + self._get_reconfigure_subentry(), + data=self.options, + ) + + return self.async_show_form( + step_id="model", + data_schema=self.add_suggested_values_to_schema( + vol.Schema(step_schema), self.options + ), + errors=errors or None, + last_step=True, + ) + + async def _get_model_list(self) -> list[SelectOptionDict]: + """Get list of available models.""" + entry = self._get_entry() + client = anthropic.AsyncAnthropic( + api_key=entry.data[CONF_API_KEY], + base_url=entry.data.get(CONF_BASE_URL) or None, + http_client=get_async_client(self.hass), + ) + return await get_model_list(client) + + async def _get_location_data(self) -> dict[str, str]: + """Get approximate location data of the user.""" + location_data: dict[str, str] = {} + zone_home = self.hass.states.get(ENTITY_ID_HOME) + if zone_home is not None: + entry = self._get_entry() + client = anthropic.AsyncAnthropic( + api_key=entry.data[CONF_API_KEY], + base_url=entry.data.get(CONF_BASE_URL) or None, + http_client=get_async_client(self.hass), + ) + location_schema = vol.Schema( + { + vol.Optional( + CONF_WEB_SEARCH_CITY, + description="Free text input for the city, e.g. `San Francisco`", + ): str, + vol.Optional( + CONF_WEB_SEARCH_REGION, + description="Free text input for the region, e.g. `California`", + ): str, + } + ) + response = await client.messages.create( + model=cast(str, DEFAULT[CONF_CHAT_MODEL]), + messages=[ + { + "role": "user", + "content": "Where are the following coordinates located: " + f"({zone_home.attributes[ATTR_LATITUDE]}," + f" {zone_home.attributes[ATTR_LONGITUDE]})?", + } + ], + max_tokens=cast(int, DEFAULT[CONF_MAX_TOKENS]), + output_config={ + "format": { + "type": "json_schema", + "schema": { + **convert(location_schema), + "additionalProperties": False, + }, + } + }, + ) + _LOGGER.debug("Model response: %s", response.content) + location_data = location_schema( + json.loads( + "".join( + block.text + for block in response.content + if isinstance(block, anthropic.types.TextBlock) + ) + ) + or {} + ) + + if self.hass.config.country: + location_data[CONF_WEB_SEARCH_COUNTRY] = self.hass.config.country + location_data[CONF_WEB_SEARCH_TIMEZONE] = self.hass.config.time_zone + + _LOGGER.debug("Location data: %s", location_data) + + return location_data + diff --git a/custom_components/anthropic_custom/const.py b/custom_components/anthropic_custom/const.py new file mode 100644 index 0000000..b68b00a --- /dev/null +++ b/custom_components/anthropic_custom/const.py @@ -0,0 +1,78 @@ +"""Constants for the Anthropic (Custom Base URL) integration.""" + +import logging + +DOMAIN = "anthropic_custom" +CONF_BASE_URL = "base_url" +LOGGER = logging.getLogger(__package__) + +DEFAULT_CONVERSATION_NAME = "Claude conversation" +DEFAULT_AI_TASK_NAME = "Claude AI Task" + +CONF_RECOMMENDED = "recommended" +CONF_PROMPT = "prompt" +CONF_CHAT_MODEL = "chat_model" +CONF_CODE_EXECUTION = "code_execution" +CONF_MAX_TOKENS = "max_tokens" +CONF_TEMPERATURE = "temperature" +CONF_THINKING_BUDGET = "thinking_budget" +CONF_THINKING_EFFORT = "thinking_effort" +CONF_WEB_SEARCH = "web_search" +CONF_WEB_SEARCH_USER_LOCATION = "user_location" +CONF_WEB_SEARCH_MAX_USES = "web_search_max_uses" +CONF_WEB_SEARCH_CITY = "city" +CONF_WEB_SEARCH_REGION = "region" +CONF_WEB_SEARCH_COUNTRY = "country" +CONF_WEB_SEARCH_TIMEZONE = "timezone" + +DEFAULT = { + CONF_CHAT_MODEL: "claude-haiku-4-5", + CONF_CODE_EXECUTION: False, + CONF_MAX_TOKENS: 3000, + CONF_TEMPERATURE: 1.0, + CONF_THINKING_BUDGET: 0, + CONF_THINKING_EFFORT: "low", + CONF_WEB_SEARCH: False, + CONF_WEB_SEARCH_USER_LOCATION: False, + CONF_WEB_SEARCH_MAX_USES: 5, +} + +MIN_THINKING_BUDGET = 1024 + +NON_THINKING_MODELS = [ + "claude-3-haiku", +] + +NON_ADAPTIVE_THINKING_MODELS = [ + "claude-opus-4-5", + "claude-sonnet-4-5", + "claude-haiku-4-5", + "claude-opus-4-1", + "claude-opus-4-0", + "claude-opus-4-20250514", + "claude-sonnet-4-0", + "claude-sonnet-4-20250514", + "claude-3-haiku", +] + +UNSUPPORTED_STRUCTURED_OUTPUT_MODELS = [ + "claude-opus-4-1", + "claude-opus-4-0", + "claude-opus-4-20250514", + "claude-sonnet-4-0", + "claude-sonnet-4-20250514", + "claude-3-haiku", +] + +WEB_SEARCH_UNSUPPORTED_MODELS = [ + "claude-3-haiku", +] + +CODE_EXECUTION_UNSUPPORTED_MODELS = [ + "claude-3-haiku", +] + +DEPRECATED_MODELS = [ + "claude-3", +] + diff --git a/custom_components/anthropic_custom/conversation.py b/custom_components/anthropic_custom/conversation.py new file mode 100644 index 0000000..bd3ec31 --- /dev/null +++ b/custom_components/anthropic_custom/conversation.py @@ -0,0 +1,76 @@ +"""Conversation support for Anthropic.""" + +from typing import Literal + +from homeassistant.components import conversation +from homeassistant.config_entries import ConfigSubentry +from homeassistant.const import CONF_LLM_HASS_API, MATCH_ALL +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from . import AnthropicConfigEntry +from .const import CONF_PROMPT, DOMAIN +from .entity import AnthropicBaseLLMEntity + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: AnthropicConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up conversation entities.""" + for subentry in config_entry.subentries.values(): + if subentry.subentry_type != "conversation": + continue + + async_add_entities( + [AnthropicConversationEntity(config_entry, subentry)], + config_subentry_id=subentry.subentry_id, + ) + + +class AnthropicConversationEntity( + conversation.ConversationEntity, + conversation.AbstractConversationAgent, + AnthropicBaseLLMEntity, +): + """Anthropic conversation agent.""" + + _attr_supports_streaming = True + _attr_translation_key = "conversation" + + def __init__(self, entry: AnthropicConfigEntry, subentry: ConfigSubentry) -> None: + """Initialize the agent.""" + super().__init__(entry, subentry) + if self.subentry.data.get(CONF_LLM_HASS_API): + self._attr_supported_features = ( + conversation.ConversationEntityFeature.CONTROL + ) + + @property + def supported_languages(self) -> list[str] | Literal["*"]: + """Return a list of supported languages.""" + return MATCH_ALL + + async def _async_handle_message( + self, + user_input: conversation.ConversationInput, + chat_log: conversation.ChatLog, + ) -> conversation.ConversationResult: + """Call the API.""" + options = self.subentry.data + + try: + await chat_log.async_provide_llm_data( + user_input.as_llm_context(DOMAIN), + options.get(CONF_LLM_HASS_API), + options.get(CONF_PROMPT), + user_input.extra_system_prompt, + ) + except conversation.ConverseError as err: + return err.as_conversation_result() + + await self._async_handle_chat_log(chat_log) + + return conversation.async_get_result_from_chat_log(user_input, chat_log) + diff --git a/custom_components/anthropic_custom/entity.py b/custom_components/anthropic_custom/entity.py new file mode 100644 index 0000000..569c88a --- /dev/null +++ b/custom_components/anthropic_custom/entity.py @@ -0,0 +1,924 @@ +"""Base entity for Anthropic.""" + +import base64 +from collections.abc import AsyncGenerator, Callable, Iterable +from dataclasses import dataclass, field +from datetime import UTC, datetime +import json +from mimetypes import guess_file_type +from pathlib import Path +from typing import Any, Literal, cast + +import anthropic +from anthropic import AsyncStream +from anthropic.types import ( + Base64ImageSourceParam, + Base64PDFSourceParam, + BashCodeExecutionToolResultBlock, + CitationsDelta, + CitationsWebSearchResultLocation, + CitationWebSearchResultLocationParam, + CodeExecutionTool20250825Param, + Container, + ContentBlockParam, + DocumentBlockParam, + ImageBlockParam, + InputJSONDelta, + JSONOutputFormatParam, + MessageDeltaUsage, + MessageParam, + MessageStreamEvent, + OutputConfigParam, + RawContentBlockDeltaEvent, + RawContentBlockStartEvent, + RawContentBlockStopEvent, + RawMessageDeltaEvent, + RawMessageStartEvent, + RawMessageStopEvent, + RedactedThinkingBlock, + RedactedThinkingBlockParam, + ServerToolUseBlock, + ServerToolUseBlockParam, + SignatureDelta, + TextBlock, + TextBlockParam, + TextCitation, + TextCitationParam, + TextDelta, + TextEditorCodeExecutionToolResultBlock, + ThinkingBlock, + ThinkingBlockParam, + ThinkingConfigAdaptiveParam, + ThinkingConfigDisabledParam, + ThinkingConfigEnabledParam, + ThinkingDelta, + ToolChoiceAnyParam, + ToolChoiceAutoParam, + ToolChoiceToolParam, + ToolParam, + ToolUnionParam, + ToolUseBlock, + ToolUseBlockParam, + Usage, + WebSearchTool20250305Param, + WebSearchToolResultBlock, + WebSearchToolResultBlockParamContentParam, +) +from anthropic.types.bash_code_execution_tool_result_block_param import ( + Content as BashCodeExecutionToolResultContentParam, +) +from anthropic.types.message_create_params import MessageCreateParamsStreaming +from anthropic.types.text_editor_code_execution_tool_result_block_param import ( + Content as TextEditorCodeExecutionToolResultContentParam, +) +import voluptuous as vol +from voluptuous_openapi import convert + +from homeassistant.components import conversation +from homeassistant.config_entries import ConfigSubentry +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry as dr, llm +from homeassistant.helpers.entity import Entity +from homeassistant.helpers.json import json_dumps +from homeassistant.util import slugify +from homeassistant.util.json import JsonObjectType + +from . import AnthropicConfigEntry +from .const import ( + CONF_CHAT_MODEL, + CONF_CODE_EXECUTION, + CONF_MAX_TOKENS, + CONF_TEMPERATURE, + CONF_THINKING_BUDGET, + CONF_THINKING_EFFORT, + CONF_WEB_SEARCH, + CONF_WEB_SEARCH_CITY, + CONF_WEB_SEARCH_COUNTRY, + CONF_WEB_SEARCH_MAX_USES, + CONF_WEB_SEARCH_REGION, + CONF_WEB_SEARCH_TIMEZONE, + CONF_WEB_SEARCH_USER_LOCATION, + DEFAULT, + DOMAIN, + LOGGER, + MIN_THINKING_BUDGET, + NON_ADAPTIVE_THINKING_MODELS, + NON_THINKING_MODELS, + UNSUPPORTED_STRUCTURED_OUTPUT_MODELS, +) + +# Max number of back and forth with the LLM to generate a response +MAX_TOOL_ITERATIONS = 10 + + +def _format_tool( + tool: llm.Tool, custom_serializer: Callable[[Any], Any] | None +) -> ToolParam: + """Format tool specification.""" + return ToolParam( + name=tool.name, + description=tool.description or "", + input_schema=convert(tool.parameters, custom_serializer=custom_serializer), + ) + + +@dataclass(slots=True) +class CitationDetails: + """Citation details for a content part.""" + + index: int = 0 + """Start position of the text.""" + + length: int = 0 + """Length of the relevant data.""" + + citations: list[TextCitationParam] = field(default_factory=list) + """Citations for the content part.""" + + +@dataclass(slots=True) +class ContentDetails: + """Native data for AssistantContent.""" + + citation_details: list[CitationDetails] = field(default_factory=list) + thinking_signature: str | None = None + redacted_thinking: str | None = None + container: Container | None = None + + def has_content(self) -> bool: + """Check if there is any text content.""" + return any(detail.length > 0 for detail in self.citation_details) + + def __bool__(self) -> bool: + """Check if there is any thinking content or citations.""" + return ( + self.thinking_signature is not None + or self.redacted_thinking is not None + or self.container is not None + or self.has_citations() + ) + + def has_citations(self) -> bool: + """Check if there are any citations.""" + return any(detail.citations for detail in self.citation_details) + + def add_citation_detail(self) -> None: + """Add a new citation detail.""" + if not self.citation_details or self.citation_details[-1].length > 0: + self.citation_details.append( + CitationDetails( + index=self.citation_details[-1].index + + self.citation_details[-1].length + if self.citation_details + else 0 + ) + ) + + def add_citation(self, citation: TextCitation) -> None: + """Add a citation to the current detail.""" + if not self.citation_details: + self.citation_details.append(CitationDetails()) + citation_param: TextCitationParam | None = None + if isinstance(citation, CitationsWebSearchResultLocation): + citation_param = CitationWebSearchResultLocationParam( + type="web_search_result_location", + title=citation.title, + url=citation.url, + cited_text=citation.cited_text, + encrypted_index=citation.encrypted_index, + ) + if citation_param: + self.citation_details[-1].citations.append(citation_param) + + def delete_empty(self) -> None: + """Delete empty citation details.""" + self.citation_details = [ + detail for detail in self.citation_details if detail.citations + ] + + +def _convert_content( + chat_content: Iterable[conversation.Content], +) -> tuple[list[MessageParam], str | None]: + """Transform HA chat_log content into Anthropic API format.""" + messages: list[MessageParam] = [] + container_id: str | None = None + + for content in chat_content: + if isinstance(content, conversation.ToolResultContent): + external_tool = True + if content.tool_name == "web_search": + tool_result_block: ContentBlockParam = { + "type": "web_search_tool_result", + "tool_use_id": content.tool_call_id, + "content": cast( + WebSearchToolResultBlockParamContentParam, + content.tool_result["content"] + if "content" in content.tool_result + else { + "type": "web_search_tool_result_error", + "error_code": content.tool_result.get( + "error_code", "unavailable" + ), + }, + ), + } + elif content.tool_name == "bash_code_execution": + tool_result_block = { + "type": "bash_code_execution_tool_result", + "tool_use_id": content.tool_call_id, + "content": cast( + BashCodeExecutionToolResultContentParam, content.tool_result + ), + } + elif content.tool_name == "text_editor_code_execution": + tool_result_block = { + "type": "text_editor_code_execution_tool_result", + "tool_use_id": content.tool_call_id, + "content": cast( + TextEditorCodeExecutionToolResultContentParam, + content.tool_result, + ), + } + else: + tool_result_block = { + "type": "tool_result", + "tool_use_id": content.tool_call_id, + "content": json_dumps(content.tool_result), + } + external_tool = False + if not messages or messages[-1]["role"] != ( + "assistant" if external_tool else "user" + ): + messages.append( + MessageParam( + role="assistant" if external_tool else "user", + content=[tool_result_block], + ) + ) + elif isinstance(messages[-1]["content"], str): + messages[-1]["content"] = [ + TextBlockParam(type="text", text=messages[-1]["content"]), + tool_result_block, + ] + else: + messages[-1]["content"].append(tool_result_block) # type: ignore[attr-defined] + elif isinstance(content, conversation.UserContent): + # Combine consequent user messages + if not messages or messages[-1]["role"] != "user": + messages.append( + MessageParam( + role="user", + content=content.content, + ) + ) + elif isinstance(messages[-1]["content"], str): + messages[-1]["content"] = [ + TextBlockParam(type="text", text=messages[-1]["content"]), + TextBlockParam(type="text", text=content.content), + ] + else: + messages[-1]["content"].append( # type: ignore[attr-defined] + TextBlockParam(type="text", text=content.content) + ) + elif isinstance(content, conversation.AssistantContent): + # Combine consequent assistant messages + if not messages or messages[-1]["role"] != "assistant": + messages.append( + MessageParam( + role="assistant", + content=[], + ) + ) + elif isinstance(messages[-1]["content"], str): + messages[-1]["content"] = [ + TextBlockParam(type="text", text=messages[-1]["content"]), + ] + + if isinstance(content.native, ContentDetails): + if content.native.thinking_signature: + messages[-1]["content"].append( # type: ignore[union-attr] + ThinkingBlockParam( + type="thinking", + thinking=content.thinking_content or "", + signature=content.native.thinking_signature, + ) + ) + if content.native.redacted_thinking: + messages[-1]["content"].append( # type: ignore[union-attr] + RedactedThinkingBlockParam( + type="redacted_thinking", + data=content.native.redacted_thinking, + ) + ) + if ( + content.native.container is not None + and content.native.container.expires_at > datetime.now(UTC) + ): + container_id = content.native.container.id + + if content.content: + current_index = 0 + for detail in ( + content.native.citation_details + if isinstance(content.native, ContentDetails) + else [CitationDetails(length=len(content.content))] + ): + if detail.index > current_index: + # Add text block for any text without citations + messages[-1]["content"].append( # type: ignore[union-attr] + TextBlockParam( + type="text", + text=content.content[current_index : detail.index], + ) + ) + messages[-1]["content"].append( # type: ignore[union-attr] + TextBlockParam( + type="text", + text=content.content[ + detail.index : detail.index + detail.length + ], + citations=detail.citations, + ) + if detail.citations + else TextBlockParam( + type="text", + text=content.content[ + detail.index : detail.index + detail.length + ], + ) + ) + current_index = detail.index + detail.length + if current_index < len(content.content): + # Add text block for any remaining text without citations + messages[-1]["content"].append( # type: ignore[union-attr] + TextBlockParam( + type="text", + text=content.content[current_index:], + ) + ) + + if content.tool_calls: + messages[-1]["content"].extend( # type: ignore[union-attr] + [ + ServerToolUseBlockParam( + type="server_tool_use", + id=tool_call.id, + name=cast( + Literal[ + "web_search", + "bash_code_execution", + "text_editor_code_execution", + ], + tool_call.tool_name, + ), + input=tool_call.tool_args, + ) + if tool_call.external + and tool_call.tool_name + in [ + "web_search", + "bash_code_execution", + "text_editor_code_execution", + ] + else ToolUseBlockParam( + type="tool_use", + id=tool_call.id, + name=tool_call.tool_name, + input=tool_call.tool_args, + ) + for tool_call in content.tool_calls + ] + ) + + if ( + isinstance(messages[-1]["content"], list) + and len(messages[-1]["content"]) == 1 + and messages[-1]["content"][0]["type"] == "text" + ): + # If there is only one text block, simplify the content to a string + messages[-1]["content"] = messages[-1]["content"][0]["text"] + else: + # Note: We don't pass SystemContent here as its passed to the API as the prompt + raise TypeError(f"Unexpected content type: {type(content)}") + + return messages, container_id + + +async def _transform_stream( # noqa: C901 - This is complex, but better to have it in one place + chat_log: conversation.ChatLog, + stream: AsyncStream[MessageStreamEvent], + output_tool: str | None = None, +) -> AsyncGenerator[ + conversation.AssistantContentDeltaDict | conversation.ToolResultContentDeltaDict +]: + """Transform the response stream into HA format. + + A typical stream of responses might look something like the following: + - RawMessageStartEvent with no content + - RawContentBlockStartEvent with an empty ThinkingBlock (if extended thinking is enabled) + - RawContentBlockDeltaEvent with a ThinkingDelta + - RawContentBlockDeltaEvent with a ThinkingDelta + - RawContentBlockDeltaEvent with a ThinkingDelta + - ... + - RawContentBlockDeltaEvent with a SignatureDelta + - RawContentBlockStopEvent + - RawContentBlockStartEvent with a RedactedThinkingBlock (occasionally) + - RawContentBlockStopEvent (RedactedThinkingBlock does not have a delta) + - RawContentBlockStartEvent with an empty TextBlock + - RawContentBlockDeltaEvent with a TextDelta + - RawContentBlockDeltaEvent with a TextDelta + - RawContentBlockDeltaEvent with a TextDelta + - ... + - RawContentBlockStopEvent + - RawContentBlockStartEvent with ToolUseBlock specifying the function name + - RawContentBlockDeltaEvent with a InputJSONDelta + - RawContentBlockDeltaEvent with a InputJSONDelta + - ... + - RawContentBlockStopEvent + - RawMessageDeltaEvent with a stop_reason='tool_use' + - RawMessageStopEvent(type='message_stop') + + Each message could contain multiple blocks of the same type. + """ + if stream is None: + raise TypeError("Expected a stream of messages") + + current_tool_block: ToolUseBlockParam | ServerToolUseBlockParam | None = None + current_tool_args: str + content_details = ContentDetails() + content_details.add_citation_detail() + input_usage: Usage | None = None + first_block: bool = True + + async for response in stream: + LOGGER.debug("Received response: %s", response) + + if isinstance(response, RawMessageStartEvent): + if response.message.role != "assistant": + raise ValueError("Unexpected message role") + input_usage = response.message.usage + first_block = True + elif isinstance(response, RawContentBlockStartEvent): + if isinstance(response.content_block, ToolUseBlock): + current_tool_block = ToolUseBlockParam( + type="tool_use", + id=response.content_block.id, + name=response.content_block.name, + input={}, + ) + current_tool_args = "" + if response.content_block.name == output_tool: + if first_block or content_details.has_content(): + if content_details: + content_details.delete_empty() + yield {"native": content_details} + content_details = ContentDetails() + content_details.add_citation_detail() + yield {"role": "assistant"} + first_block = False + elif isinstance(response.content_block, TextBlock): + if ( # Do not start a new assistant content just for citations, concatenate consecutive blocks with citations instead. + first_block + or ( + not content_details.has_citations() + and response.content_block.citations is None + and content_details.has_content() + ) + ): + if content_details: + content_details.delete_empty() + yield {"native": content_details} + content_details = ContentDetails() + yield {"role": "assistant"} + first_block = False + content_details.add_citation_detail() + if response.content_block.text: + content_details.citation_details[-1].length += len( + response.content_block.text + ) + yield {"content": response.content_block.text} + elif isinstance(response.content_block, ThinkingBlock): + if first_block or content_details.thinking_signature: + if content_details: + content_details.delete_empty() + yield {"native": content_details} + content_details = ContentDetails() + content_details.add_citation_detail() + yield {"role": "assistant"} + first_block = False + elif isinstance(response.content_block, RedactedThinkingBlock): + LOGGER.debug( + "Some of Claude’s internal reasoning has been automatically " + "encrypted for safety reasons. This doesn’t affect the quality of " + "responses" + ) + if first_block or content_details.redacted_thinking: + if content_details: + content_details.delete_empty() + yield {"native": content_details} + content_details = ContentDetails() + content_details.add_citation_detail() + yield {"role": "assistant"} + first_block = False + content_details.redacted_thinking = response.content_block.data + elif isinstance(response.content_block, ServerToolUseBlock): + current_tool_block = ServerToolUseBlockParam( + type="server_tool_use", + id=response.content_block.id, + name=response.content_block.name, + input={}, + ) + current_tool_args = "" + elif isinstance( + response.content_block, + ( + WebSearchToolResultBlock, + BashCodeExecutionToolResultBlock, + TextEditorCodeExecutionToolResultBlock, + ), + ): + if content_details: + content_details.delete_empty() + yield {"native": content_details} + content_details = ContentDetails() + content_details.add_citation_detail() + yield { + "role": "tool_result", + "tool_call_id": response.content_block.tool_use_id, + "tool_name": response.content_block.type.removesuffix( + "_tool_result" + ), + "tool_result": { + "content": cast( + JsonObjectType, response.content_block.to_dict()["content"] + ) + } + if isinstance(response.content_block.content, list) + else cast(JsonObjectType, response.content_block.content.to_dict()), + } + first_block = True + elif isinstance(response, RawContentBlockDeltaEvent): + if isinstance(response.delta, InputJSONDelta): + if ( + current_tool_block is not None + and current_tool_block["name"] == output_tool + ): + content_details.citation_details[-1].length += len( + response.delta.partial_json + ) + yield {"content": response.delta.partial_json} + else: + current_tool_args += response.delta.partial_json + elif isinstance(response.delta, TextDelta): + if response.delta.text: + content_details.citation_details[-1].length += len( + response.delta.text + ) + yield {"content": response.delta.text} + elif isinstance(response.delta, ThinkingDelta): + if response.delta.thinking: + yield {"thinking_content": response.delta.thinking} + elif isinstance(response.delta, SignatureDelta): + content_details.thinking_signature = response.delta.signature + elif isinstance(response.delta, CitationsDelta): + content_details.add_citation(response.delta.citation) + elif isinstance(response, RawContentBlockStopEvent): + if current_tool_block is not None: + if current_tool_block["name"] == output_tool: + current_tool_block = None + continue + tool_args = json.loads(current_tool_args) if current_tool_args else {} + current_tool_block["input"] = tool_args + yield { + "tool_calls": [ + llm.ToolInput( + id=current_tool_block["id"], + tool_name=current_tool_block["name"], + tool_args=tool_args, + external=current_tool_block["type"] == "server_tool_use", + ) + ] + } + current_tool_block = None + elif isinstance(response, RawMessageDeltaEvent): + if (usage := response.usage) is not None: + chat_log.async_trace(_create_token_stats(input_usage, usage)) + content_details.container = response.delta.container + if response.delta.stop_reason == "refusal": + raise HomeAssistantError("Potential policy violation detected") + elif isinstance(response, RawMessageStopEvent): + if content_details: + content_details.delete_empty() + yield {"native": content_details} + content_details = ContentDetails() + content_details.add_citation_detail() + + +def _create_token_stats( + input_usage: Usage | None, response_usage: MessageDeltaUsage +) -> dict[str, Any]: + """Create token stats for conversation agent tracing.""" + input_tokens = 0 + cached_input_tokens = 0 + if input_usage: + input_tokens = input_usage.input_tokens + cached_input_tokens = input_usage.cache_creation_input_tokens or 0 + output_tokens = response_usage.output_tokens + return { + "stats": { + "input_tokens": input_tokens, + "cached_input_tokens": cached_input_tokens, + "output_tokens": output_tokens, + } + } + + +class AnthropicBaseLLMEntity(Entity): + """Anthropic base LLM entity.""" + + _attr_has_entity_name = True + _attr_name = None + + def __init__(self, entry: AnthropicConfigEntry, subentry: ConfigSubentry) -> None: + """Initialize the entity.""" + self.entry = entry + self.subentry = subentry + self._attr_unique_id = subentry.subentry_id + self._attr_device_info = dr.DeviceInfo( + identifiers={(DOMAIN, subentry.subentry_id)}, + name=subentry.title, + manufacturer="Anthropic", + model=subentry.data.get(CONF_CHAT_MODEL, DEFAULT[CONF_CHAT_MODEL]), + entry_type=dr.DeviceEntryType.SERVICE, + ) + + async def _async_handle_chat_log( + self, + chat_log: conversation.ChatLog, + structure_name: str | None = None, + structure: vol.Schema | None = None, + max_iterations: int = MAX_TOOL_ITERATIONS, + ) -> None: + """Generate an answer for the chat log.""" + options = self.subentry.data + + system = chat_log.content[0] + if not isinstance(system, conversation.SystemContent): + raise TypeError("First message must be a system message") + + # System prompt with caching enabled + system_prompt: list[TextBlockParam] = [ + TextBlockParam( + type="text", + text=system.content, + cache_control={"type": "ephemeral"}, + ) + ] + + messages, container_id = _convert_content(chat_log.content[1:]) + + model = options.get(CONF_CHAT_MODEL, DEFAULT[CONF_CHAT_MODEL]) + + model_args = MessageCreateParamsStreaming( + model=model, + messages=messages, + max_tokens=options.get(CONF_MAX_TOKENS, DEFAULT[CONF_MAX_TOKENS]), + system=system_prompt, + stream=True, + container=container_id, + ) + + if not model.startswith(tuple(NON_ADAPTIVE_THINKING_MODELS)): + thinking_effort = options.get( + CONF_THINKING_EFFORT, DEFAULT[CONF_THINKING_EFFORT] + ) + if thinking_effort != "none": + model_args["thinking"] = ThinkingConfigAdaptiveParam(type="adaptive") + model_args["output_config"] = OutputConfigParam(effort=thinking_effort) + else: + model_args["thinking"] = ThinkingConfigDisabledParam(type="disabled") + model_args["temperature"] = options.get( + CONF_TEMPERATURE, DEFAULT[CONF_TEMPERATURE] + ) + else: + thinking_budget = options.get( + CONF_THINKING_BUDGET, DEFAULT[CONF_THINKING_BUDGET] + ) + if ( + not model.startswith(tuple(NON_THINKING_MODELS)) + and thinking_budget >= MIN_THINKING_BUDGET + ): + model_args["thinking"] = ThinkingConfigEnabledParam( + type="enabled", budget_tokens=thinking_budget + ) + else: + model_args["thinking"] = ThinkingConfigDisabledParam(type="disabled") + model_args["temperature"] = options.get( + CONF_TEMPERATURE, DEFAULT[CONF_TEMPERATURE] + ) + + tools: list[ToolUnionParam] = [] + if chat_log.llm_api: + tools = [ + _format_tool(tool, chat_log.llm_api.custom_serializer) + for tool in chat_log.llm_api.tools + ] + + if options.get(CONF_CODE_EXECUTION): + tools.append( + CodeExecutionTool20250825Param( + name="code_execution", + type="code_execution_20250825", + ), + ) + + if options.get(CONF_WEB_SEARCH): + web_search = WebSearchTool20250305Param( + name="web_search", + type="web_search_20250305", + max_uses=options.get(CONF_WEB_SEARCH_MAX_USES), + ) + if options.get(CONF_WEB_SEARCH_USER_LOCATION): + web_search["user_location"] = { + "type": "approximate", + "city": options.get(CONF_WEB_SEARCH_CITY, ""), + "region": options.get(CONF_WEB_SEARCH_REGION, ""), + "country": options.get(CONF_WEB_SEARCH_COUNTRY, ""), + "timezone": options.get(CONF_WEB_SEARCH_TIMEZONE, ""), + } + tools.append(web_search) + + # Handle attachments by adding them to the last user message + last_content = chat_log.content[-1] + if last_content.role == "user" and last_content.attachments: + last_message = messages[-1] + if last_message["role"] != "user": + raise HomeAssistantError( + "Last message must be a user message to add attachments" + ) + if isinstance(last_message["content"], str): + last_message["content"] = [ + TextBlockParam(type="text", text=last_message["content"]) + ] + last_message["content"].extend( # type: ignore[union-attr] + await async_prepare_files_for_prompt( + self.hass, [(a.path, a.mime_type) for a in last_content.attachments] + ) + ) + + if structure and structure_name: + if not model.startswith(tuple(UNSUPPORTED_STRUCTURED_OUTPUT_MODELS)): + # Native structured output for those models who support it. + structure_name = None + model_args.setdefault("output_config", OutputConfigParam())[ + "format" + ] = JSONOutputFormatParam( + type="json_schema", + schema={ + **convert( + structure, + custom_serializer=chat_log.llm_api.custom_serializer + if chat_log.llm_api + else llm.selector_serializer, + ), + "additionalProperties": False, + }, + ) + elif model_args["thinking"]["type"] == "disabled": + structure_name = slugify(structure_name) + if not tools: + # Simplest case: no tools and no extended thinking + # Add a tool and force its use + model_args["tool_choice"] = ToolChoiceToolParam( + type="tool", + name=structure_name, + ) + else: + # Second case: tools present but no extended thinking + # Allow the model to use any tool but not text response + # The model should know to use the right tool by its description + model_args["tool_choice"] = ToolChoiceAnyParam( + type="any", + ) + else: + # Extended thinking is enabled. With extended thinking, we cannot + # force tool use or disable text responses, so we add a hint to the + # system prompt instead. With extended thinking, the model should be + # smart enough to use the tool. + structure_name = slugify(structure_name) + model_args["tool_choice"] = ToolChoiceAutoParam( + type="auto", + ) + + model_args["system"].append( # type: ignore[union-attr] + TextBlockParam( + type="text", + text=f"Claude MUST use the '{structure_name}' tool to provide " + "the final answer instead of plain text.", + ) + ) + + if structure_name: + tools.append( + ToolParam( + name=structure_name, + description="Use this tool to reply to the user", + input_schema=convert( + structure, + custom_serializer=chat_log.llm_api.custom_serializer + if chat_log.llm_api + else llm.selector_serializer, + ), + ) + ) + + if tools: + model_args["tools"] = tools + + client = self.entry.runtime_data + + # To prevent infinite loops, we limit the number of iterations + for _iteration in range(max_iterations): + try: + stream = await client.messages.create(**model_args) + + new_messages, model_args["container"] = _convert_content( + [ + content + async for content in chat_log.async_add_delta_content_stream( + self.entity_id, + _transform_stream( + chat_log, + stream, + output_tool=structure_name or None, + ), + ) + ] + ) + messages.extend(new_messages) + except anthropic.AnthropicError as err: + raise HomeAssistantError( + f"Sorry, I had a problem talking to Anthropic: {err}" + ) from err + + if not chat_log.unresponded_tool_results: + break + + +async def async_prepare_files_for_prompt( + hass: HomeAssistant, files: list[tuple[Path, str | None]] +) -> Iterable[ImageBlockParam | DocumentBlockParam]: + """Append files to a prompt. + + Caller needs to ensure that the files are allowed. + """ + + def append_files_to_content() -> Iterable[ImageBlockParam | DocumentBlockParam]: + content: list[ImageBlockParam | DocumentBlockParam] = [] + + for file_path, mime_type in files: + if not file_path.exists(): + raise HomeAssistantError(f"`{file_path}` does not exist") + + if mime_type is None: + mime_type = guess_file_type(file_path)[0] + + if not mime_type or not mime_type.startswith(("image/", "application/pdf")): + raise HomeAssistantError( + "Only images and PDF are supported by the Anthropic API," + f"`{file_path}` is not an image file or PDF" + ) + if mime_type == "image/jpg": + mime_type = "image/jpeg" + + base64_file = base64.b64encode(file_path.read_bytes()).decode("utf-8") + + if mime_type.startswith("image/"): + content.append( + ImageBlockParam( + type="image", + source=Base64ImageSourceParam( + type="base64", + media_type=mime_type, # type: ignore[typeddict-item] + data=base64_file, + ), + ) + ) + elif mime_type.startswith("application/pdf"): + content.append( + DocumentBlockParam( + type="document", + source=Base64PDFSourceParam( + type="base64", + media_type=mime_type, # type: ignore[typeddict-item] + data=base64_file, + ), + ) + ) + + return content + + return await hass.async_add_executor_job(append_files_to_content) + diff --git a/custom_components/anthropic_custom/icons.json b/custom_components/anthropic_custom/icons.json new file mode 100644 index 0000000..7ec3e20 --- /dev/null +++ b/custom_components/anthropic_custom/icons.json @@ -0,0 +1,15 @@ +{ + "entity": { + "ai_task": { + "ai_task_data": { + "default": "mdi:asterisk" + } + }, + "conversation": { + "conversation": { + "default": "mdi:asterisk" + } + } + } +} + diff --git a/custom_components/anthropic_custom/manifest.json b/custom_components/anthropic_custom/manifest.json new file mode 100644 index 0000000..c1da10a --- /dev/null +++ b/custom_components/anthropic_custom/manifest.json @@ -0,0 +1,13 @@ +{ + "domain": "anthropic_custom", + "name": "Anthropic (Custom Base URL)", + "after_dependencies": ["assist_pipeline", "intent"], + "codeowners": [], + "config_flow": true, + "dependencies": ["conversation"], + "documentation": "https://github.com/robot/ha-anthropic-custom", + "integration_type": "service", + "iot_class": "cloud_polling", + "quality_scale": "bronze", + "requirements": ["anthropic==0.83.0"] +} diff --git a/custom_components/anthropic_custom/quality_scale.yaml b/custom_components/anthropic_custom/quality_scale.yaml new file mode 100644 index 0000000..78ad74c --- /dev/null +++ b/custom_components/anthropic_custom/quality_scale.yaml @@ -0,0 +1,109 @@ +rules: + # Bronze + action-setup: + status: exempt + comment: | + Integration has no actions. + appropriate-polling: + status: exempt + comment: | + Integration does not poll. + brands: done + common-modules: done + config-flow-test-coverage: done + config-flow: done + dependency-transparency: done + docs-actions: + status: exempt + comment: | + Integration has no actions. + docs-high-level-description: done + docs-installation-instructions: done + docs-removal-instructions: done + entity-event-setup: + status: exempt + comment: | + Integration does not subscribe to events. + entity-unique-id: done + has-entity-name: done + runtime-data: done + test-before-configure: done + test-before-setup: done + unique-config-entry: done + # Silver + action-exceptions: + status: todo + comment: | + Reevaluate exceptions for entity services. + config-entry-unloading: done + docs-configuration-parameters: done + docs-installation-parameters: done + entity-unavailable: todo + integration-owner: done + log-when-unavailable: todo + parallel-updates: + status: exempt + comment: | + The API does not limit parallel updates. + reauthentication-flow: done + test-coverage: done + # Gold + devices: done + diagnostics: todo + discovery-update-info: + status: exempt + comment: | + Service integration, no discovery. + discovery: + status: exempt + comment: | + Service integration, no discovery. + docs-data-update: + status: exempt + comment: | + No data updates. + docs-examples: + status: todo + comment: | + To give examples of how people use the integration + docs-known-limitations: done + docs-supported-devices: + status: todo + comment: | + To write something about what models we support. + docs-supported-functions: done + docs-troubleshooting: todo + docs-use-cases: done + dynamic-devices: + status: exempt + comment: | + Service integration, no devices. + entity-category: + status: exempt + comment: | + No entities with categories. + entity-device-class: + status: exempt + comment: | + No entities with device classes. + entity-disabled-by-default: + status: exempt + comment: | + No entities disabled by default. + entity-translations: todo + exception-translations: todo + icon-translations: done + reconfiguration-flow: done + repair-issues: done + stale-devices: + status: exempt + comment: | + Service integration, no devices. + # Platinum + async-dependency: done + inject-websession: + status: done + comment: | + Uses `httpx` session. + strict-typing: done + diff --git a/custom_components/anthropic_custom/repairs.py b/custom_components/anthropic_custom/repairs.py new file mode 100644 index 0000000..2199f62 --- /dev/null +++ b/custom_components/anthropic_custom/repairs.py @@ -0,0 +1,194 @@ +"""Issue repair flow for Anthropic.""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import TYPE_CHECKING + +import voluptuous as vol + +from homeassistant import data_entry_flow +from homeassistant.components.repairs import RepairsFlow +from homeassistant.config_entries import ConfigEntryState, ConfigSubentry +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.selector import ( + SelectOptionDict, + SelectSelector, + SelectSelectorConfig, +) + +from .config_flow import get_model_list +from .const import CONF_CHAT_MODEL, DEPRECATED_MODELS, DOMAIN + +if TYPE_CHECKING: + from . import AnthropicConfigEntry + + +class ModelDeprecatedRepairFlow(RepairsFlow): + """Handler for an issue fixing flow.""" + + _subentry_iter: Iterator[tuple[str, str]] | None + _current_entry_id: str | None + _current_subentry_id: str | None + _model_list_cache: dict[str, list[SelectOptionDict]] | None + + def __init__(self) -> None: + """Initialize the flow.""" + super().__init__() + self._subentry_iter = None + self._current_entry_id = None + self._current_subentry_id = None + self._model_list_cache = None + + async def async_step_init( + self, user_input: dict[str, str] + ) -> data_entry_flow.FlowResult: + """Handle the steps of a fix flow.""" + if user_input.get(CONF_CHAT_MODEL): + self._async_update_current_subentry(user_input) + + target = await self._async_next_target() + if target is None: + return self.async_create_entry(data={}) + + entry, subentry, model = target + if self._model_list_cache is None: + self._model_list_cache = {} + if entry.entry_id in self._model_list_cache: + model_list = self._model_list_cache[entry.entry_id] + else: + client = entry.runtime_data + model_list = [ + model_option + for model_option in await get_model_list(client) + if not model_option["value"].startswith(tuple(DEPRECATED_MODELS)) + ] + self._model_list_cache[entry.entry_id] = model_list + + if "opus" in model: + family = "claude-opus" + elif "sonnet" in model: + family = "claude-sonnet" + else: + family = "claude-haiku" + + suggested_model = next( + ( + model_option["value"] + for model_option in sorted( + (m for m in model_list if family in m["value"]), + key=lambda x: x["value"], + reverse=True, + ) + ), + vol.UNDEFINED, + ) + + schema = vol.Schema( + { + vol.Required( + CONF_CHAT_MODEL, + default=suggested_model, + ): SelectSelector( + SelectSelectorConfig(options=model_list, custom_value=True) + ), + } + ) + + return self.async_show_form( + step_id="init", + data_schema=schema, + description_placeholders={ + "entry_name": entry.title, + "model": model, + "subentry_name": subentry.title, + "subentry_type": self._format_subentry_type(subentry.subentry_type), + }, + ) + + def _iter_deprecated_subentries(self) -> Iterator[tuple[str, str]]: + """Yield entry/subentry pairs that use deprecated models.""" + for entry in self.hass.config_entries.async_entries(DOMAIN): + if entry.state is not ConfigEntryState.LOADED: + continue + for subentry in entry.subentries.values(): + model = subentry.data.get(CONF_CHAT_MODEL) + if model and model.startswith(tuple(DEPRECATED_MODELS)): + yield entry.entry_id, subentry.subentry_id + + async def _async_next_target( + self, + ) -> tuple[AnthropicConfigEntry, ConfigSubentry, str] | None: + """Return the next deprecated subentry target.""" + if self._subentry_iter is None: + self._subentry_iter = self._iter_deprecated_subentries() + + while True: + try: + entry_id, subentry_id = next(self._subentry_iter) + except StopIteration: + return None + + # Verify that the entry/subentry still exists and the model is still + # deprecated. This may have changed since we started the repair flow. + entry = self.hass.config_entries.async_get_entry(entry_id) + if entry is None: + continue + + subentry = entry.subentries.get(subentry_id) + if subentry is None: + continue + + model = subentry.data.get(CONF_CHAT_MODEL) + if not model or not model.startswith(tuple(DEPRECATED_MODELS)): + continue + + self._current_entry_id = entry_id + self._current_subentry_id = subentry_id + return entry, subentry, model + + def _async_update_current_subentry(self, user_input: dict[str, str]) -> None: + """Update the currently selected subentry.""" + if ( + self._current_entry_id is None + or self._current_subentry_id is None + or ( + entry := self.hass.config_entries.async_get_entry( + self._current_entry_id + ) + ) + is None + or (subentry := entry.subentries.get(self._current_subentry_id)) is None + ): + raise HomeAssistantError("Subentry not found") + + updated_data = { + **subentry.data, + CONF_CHAT_MODEL: user_input[CONF_CHAT_MODEL], + } + self.hass.config_entries.async_update_subentry( + entry, + subentry, + data=updated_data, + ) + + def _format_subentry_type(self, subentry_type: str) -> str: + """Return a user-friendly subentry type label.""" + if subentry_type == "conversation": + return "Conversation agent" + if subentry_type in ("ai_task", "ai_task_data"): + return "AI task" + return subentry_type + + +async def async_create_fix_flow( + hass: HomeAssistant, + issue_id: str, + data: dict[str, str | int | float | None] | None, +) -> RepairsFlow: + """Create flow.""" + if issue_id == "model_deprecated": + return ModelDeprecatedRepairFlow() + raise HomeAssistantError("Unknown issue ID") + diff --git a/custom_components/anthropic_custom/strings.json b/custom_components/anthropic_custom/strings.json new file mode 100644 index 0000000..c274d4a --- /dev/null +++ b/custom_components/anthropic_custom/strings.json @@ -0,0 +1,185 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_service%]", + "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]" + }, + "error": { + "authentication_error": "[%key:common::config_flow::error::invalid_auth%]", + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "timeout_connect": "[%key:common::config_flow::error::timeout_connect%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "step": { + "reauth_confirm": { + "data": { + "api_key": "[%key:common::config_flow::data::api_key%]" + }, + "data_description": { + "api_key": "[%key:component::anthropic::config::step::user::data_description::api_key%]" + }, + "description": "Reauthentication required. Please enter your updated API key." + }, + "user": { + "data": { + "api_key": "[%key:common::config_flow::data::api_key%]", + "base_url": "Base URL" + }, + "data_description": { + "api_key": "Your Anthropic API key.", + "base_url": "Optional custom base URL for the Anthropic API (e.g. for a proxy or compatible endpoint). Leave empty to use the default Anthropic API URL." + }, + "description": "Set up Anthropic integration by providing your Anthropic API key. Instructions to obtain an API key can be found in [the documentation]({instructions_url})." + } + } + }, + "config_subentries": { + "ai_task_data": { + "abort": { + "entry_not_loaded": "[%key:component::anthropic::config_subentries::conversation::abort::entry_not_loaded%]", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]" + }, + "entry_type": "AI task", + "initiate_flow": { + "reconfigure": "Reconfigure AI task", + "user": "Add AI task" + }, + "step": { + "advanced": { + "data": { + "chat_model": "[%key:common::generic::model%]", + "max_tokens": "[%key:component::anthropic::config_subentries::conversation::step::advanced::data::max_tokens%]", + "temperature": "[%key:component::anthropic::config_subentries::conversation::step::advanced::data::temperature%]" + }, + "data_description": { + "chat_model": "[%key:component::anthropic::config_subentries::conversation::step::advanced::data_description::chat_model%]", + "max_tokens": "[%key:component::anthropic::config_subentries::conversation::step::advanced::data_description::max_tokens%]", + "temperature": "[%key:component::anthropic::config_subentries::conversation::step::advanced::data_description::temperature%]" + }, + "title": "[%key:component::anthropic::config_subentries::conversation::step::advanced::title%]" + }, + "init": { + "data": { + "name": "[%key:common::config_flow::data::name%]", + "recommended": "[%key:component::anthropic::config_subentries::conversation::step::init::data::recommended%]" + }, + "data_description": { + "name": "[%key:component::anthropic::config_subentries::conversation::step::init::data_description::name%]", + "recommended": "[%key:component::anthropic::config_subentries::conversation::step::init::data_description::recommended%]" + }, + "title": "[%key:component::anthropic::config_subentries::conversation::step::init::title%]" + }, + "model": { + "data": { + "code_execution": "[%key:component::anthropic::config_subentries::conversation::step::model::data::code_execution%]", + "thinking_budget": "[%key:component::anthropic::config_subentries::conversation::step::model::data::thinking_budget%]", + "thinking_effort": "[%key:component::anthropic::config_subentries::conversation::step::model::data::thinking_effort%]", + "user_location": "[%key:component::anthropic::config_subentries::conversation::step::model::data::user_location%]", + "web_search": "[%key:component::anthropic::config_subentries::conversation::step::model::data::web_search%]", + "web_search_max_uses": "[%key:component::anthropic::config_subentries::conversation::step::model::data::web_search_max_uses%]" + }, + "data_description": { + "code_execution": "[%key:component::anthropic::config_subentries::conversation::step::model::data_description::code_execution%]", + "thinking_budget": "[%key:component::anthropic::config_subentries::conversation::step::model::data_description::thinking_budget%]", + "thinking_effort": "[%key:component::anthropic::config_subentries::conversation::step::model::data_description::thinking_effort%]", + "user_location": "[%key:component::anthropic::config_subentries::conversation::step::model::data_description::user_location%]", + "web_search": "[%key:component::anthropic::config_subentries::conversation::step::model::data_description::web_search%]", + "web_search_max_uses": "[%key:component::anthropic::config_subentries::conversation::step::model::data_description::web_search_max_uses%]" + }, + "title": "[%key:component::anthropic::config_subentries::conversation::step::model::title%]" + } + } + }, + "conversation": { + "abort": { + "entry_not_loaded": "Cannot add things while the configuration is disabled.", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]" + }, + "entry_type": "Conversation agent", + "initiate_flow": { + "reconfigure": "Reconfigure conversation agent", + "user": "Add conversation agent" + }, + "step": { + "advanced": { + "data": { + "chat_model": "[%key:common::generic::model%]", + "max_tokens": "Maximum tokens to return in response", + "temperature": "Temperature" + }, + "data_description": { + "chat_model": "The model to serve the responses.", + "max_tokens": "Limit the number of response tokens.", + "temperature": "Control the randomness of the response, trading off between creativity and coherence." + }, + "title": "Advanced settings" + }, + "init": { + "data": { + "llm_hass_api": "[%key:common::config_flow::data::llm_hass_api%]", + "name": "[%key:common::config_flow::data::name%]", + "prompt": "[%key:common::config_flow::data::prompt%]", + "recommended": "Recommended model settings" + }, + "data_description": { + "llm_hass_api": "Allow the LLM to control Home Assistant.", + "name": "The name of this configuration", + "prompt": "Instruct how the LLM should respond. This can be a template.", + "recommended": "Use default configuration" + }, + "title": "Basic settings" + }, + "model": { + "data": { + "code_execution": "Code execution", + "thinking_budget": "Thinking budget", + "thinking_effort": "Thinking effort", + "user_location": "Include home location", + "web_search": "Enable web search", + "web_search_max_uses": "Maximum web searches" + }, + "data_description": { + "code_execution": "Allow the model to execute code in a secure sandbox environment, enabling it to analyze data and perform complex calculations.", + "thinking_budget": "The number of tokens the model can use to think about the response out of the total maximum number of tokens. Set to 1024 or greater to enable extended thinking.", + "thinking_effort": "Control how many tokens Claude uses when responding, trading off between response thoroughness and token efficiency", + "user_location": "Localize search results based on home location", + "web_search": "The web search tool gives Claude direct access to real-time web content, allowing it to answer questions with up-to-date information beyond its knowledge cutoff", + "web_search_max_uses": "Limit the number of searches performed per response" + }, + "title": "Model-specific options" + } + } + } + }, + "issues": { + "model_deprecated": { + "fix_flow": { + "step": { + "init": { + "data": { + "chat_model": "[%key:common::generic::model%]" + }, + "data_description": { + "chat_model": "Select the new model to use." + }, + "description": "You are updating {subentry_name} ({subentry_type}) in {entry_name}. The current model {model} is deprecated. Select a supported model to continue.", + "title": "Update model" + } + } + }, + "title": "Model deprecated" + } + }, + "selector": { + "thinking_effort": { + "options": { + "high": "[%key:common::state::high%]", + "low": "[%key:common::state::low%]", + "max": "Max", + "medium": "[%key:common::state::medium%]", + "none": "None" + } + } + } +} +