/
/
/
1"""OpenAI Compatible plugin provider for Music Assistant."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
8from music_assistant_models.enums import ConfigEntryType, ProviderFeature
9from music_assistant_models.errors import (
10 MusicAssistantError,
11 SetupFailedError,
12 UnsupportedFeaturedException,
13)
14
15from music_assistant.models.plugin import AIEngine, PluginProvider
16
17from .constants import (
18 CHAT_REQUEST_TIMEOUT,
19 CONF_API_KEY,
20 CONF_BASE_URL,
21 CONF_MODELS,
22 MODELS_REQUEST_TIMEOUT,
23)
24from .helpers import chat_completion, list_models
25
26if TYPE_CHECKING:
27 from music_assistant_models.config_entries import ProviderConfig
28 from music_assistant_models.provider import ProviderManifest
29
30 from music_assistant.mass import MusicAssistant
31 from music_assistant.models import ProviderInstanceType
32
33
34SUPPORTED_FEATURES = {ProviderFeature.AI_QUERY}
35
36
37async def setup(
38 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
39) -> ProviderInstanceType:
40 """Initialize provider(instance) with given configuration."""
41 return OpenAICompatibleProvider(mass, manifest, config, SUPPORTED_FEATURES)
42
43
44class OpenAICompatibleProvider(PluginProvider):
45 """Exposes the models of an OpenAI-compatible API as selectable AI engines."""
46
47 async def handle_async_init(self) -> None:
48 """Handle async initialization of the provider."""
49 if not self._base_url:
50 msg = "No API base URL is configured"
51 raise SetupFailedError(msg)
52
53 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
54 """Return the (options) config entries to configure this provider instance."""
55 selected = self._configured_models()
56 # the listing is the natural source for the options, but it is optional in the
57 # OpenAI API: with no options at all the frontend accepts freely typed names,
58 # and keeping the current selection in the list survives a model disappearing
59 models = sorted({*await self._discover_models(), *selected})
60 return (
61 ConfigEntry(
62 key=CONF_MODELS,
63 type=ConfigEntryType.STRING,
64 multi_value=True,
65 required=False,
66 default_value=[],
67 options=[ConfigValueOption(model, title=model) for model in models],
68 category="features",
69 ),
70 )
71
72 async def get_ai_engines(self) -> list[AIEngine]:
73 """Return the AI engines this plugin exposes."""
74 # the engine name carries no provider name: consumers label a picker option
75 # as "<provider name> | <engine name>" themselves
76 return [
77 AIEngine(id=model, name=model, provider=self) for model in self._configured_models()
78 ]
79
80 async def ai_query(self, query: str, engine_id: str | None = None) -> str:
81 """Handle an AI query."""
82 model = engine_id or next(iter(self._configured_models()), None)
83 if model is None:
84 msg = "No model is selected for this provider"
85 raise UnsupportedFeaturedException(msg)
86 return await chat_completion(
87 self.mass,
88 self._base_url,
89 self._api_key,
90 model=model,
91 prompt=query,
92 timeout=CHAT_REQUEST_TIMEOUT,
93 )
94
95 @property
96 def _base_url(self) -> str:
97 """Return the configured API endpoint, without trailing slash."""
98 # read on demand: the config entries are resolved before async init runs,
99 # so this has to be readable without it
100 return str(self.get_setup_value(CONF_BASE_URL) or "").strip().rstrip("/")
101
102 @property
103 def _api_key(self) -> str:
104 """Return the configured API key, empty for an endpoint without auth."""
105 api_key = self.get_setup_value(CONF_API_KEY)
106 return api_key.strip() if isinstance(api_key, str) else ""
107
108 def _configured_models(self) -> list[str]:
109 """Return the models the user selected, in a stable order."""
110 value = self.get_config_value(CONF_MODELS)
111 if not isinstance(value, list):
112 return []
113 return sorted(
114 {model.strip() for model in value if isinstance(model, str) and model.strip()}
115 )
116
117 async def _discover_models(self) -> list[str]:
118 """Return the models the endpoint advertises, empty when it cannot be asked."""
119 try:
120 return await list_models(
121 self.mass, self._base_url, self._api_key, MODELS_REQUEST_TIMEOUT
122 )
123 except MusicAssistantError as err:
124 self.logger.warning("Could not retrieve the list of available models: %s", err)
125 return []
126