From b6447d9b05319c667432955f20fb2696bb54e3fd Mon Sep 17 00:00:00 2001 From: Marcel van der Veldt Date: Mon, 27 Oct 2025 21:07:50 +0100 Subject: [PATCH] Add automatically generated API documentation (#2559) --- music_assistant/controllers/webserver.py | 45 + music_assistant/helpers/api_docs.py | 1027 +++++++++++++++++ .../helpers/resources/api_docs.html | 487 ++++++++ .../helpers/resources/redoc_ui.html | 28 + .../helpers/resources/swagger_ui.html | 61 + 5 files changed, 1648 insertions(+) create mode 100644 music_assistant/helpers/api_docs.py create mode 100644 music_assistant/helpers/resources/api_docs.html create mode 100644 music_assistant/helpers/resources/redoc_ui.html create mode 100644 music_assistant/helpers/resources/swagger_ui.html diff --git a/music_assistant/controllers/webserver.py b/music_assistant/controllers/webserver.py index 04e2f7c8..ffa91ef9 100644 --- a/music_assistant/controllers/webserver.py +++ b/music_assistant/controllers/webserver.py @@ -8,6 +8,7 @@ this webserver allows for more fine grained configuration to better secure it. from __future__ import annotations import asyncio +import html import logging import os import urllib.parse @@ -16,6 +17,7 @@ from contextlib import suppress from functools import partial from typing import TYPE_CHECKING, Any, Final +import aiofiles from aiohttp import WSMsgType, web from music_assistant_frontend import where as locate_frontend from music_assistant_models.api import ( @@ -30,6 +32,7 @@ from music_assistant_models.errors import InvalidCommand from music_assistant.constants import CONF_BIND_IP, CONF_BIND_PORT, VERBOSE_LOG_LEVEL from music_assistant.helpers.api import APICommandHandler, parse_arguments +from music_assistant.helpers.api_docs import generate_openapi_spec from music_assistant.helpers.audio import get_preview_stream from music_assistant.helpers.json import json_dumps from music_assistant.helpers.util import get_ip_addresses @@ -151,6 +154,11 @@ class WebserverController(CoreController): routes.append(("GET", "/preview", self.serve_preview_stream)) # add jsonrpc api routes.append(("POST", "/api", self._handle_jsonrpc_api_command)) + # add api documentation + routes.append(("GET", "/api-docs", self._handle_api_intro)) + routes.append(("GET", "/api-docs/openapi.json", self._handle_openapi_spec)) + routes.append(("GET", "/api-docs/swagger", self._handle_swagger_ui)) + routes.append(("GET", "/api-docs/redoc", self._handle_redoc_ui)) # start the webserver all_ip_addresses = await get_ip_addresses() default_publish_ip = all_ip_addresses[0] @@ -260,6 +268,43 @@ class WebserverController(CoreController): log_data = await self.mass.get_application_log() return web.Response(text=log_data, content_type="text/text") + async def _handle_api_intro(self, request: web.Request) -> web.Response: + """Handle request for API introduction/documentation page.""" + intro_html_path = os.path.join( + os.path.dirname(__file__), "..", "helpers", "resources", "api_docs.html" + ) + # Read the template + async with aiofiles.open(intro_html_path) as f: + html_content = await f.read() + + # Replace placeholders (escape values to prevent XSS) + html_content = html_content.replace("{VERSION}", html.escape(self.mass.version)) + html_content = html_content.replace("{BASE_URL}", html.escape(self.base_url)) + html_content = html_content.replace("{SERVER_HOST}", html.escape(request.host)) + + return web.Response(text=html_content, content_type="text/html") + + async def _handle_openapi_spec(self, request: web.Request) -> web.Response: + """Handle request for OpenAPI specification (generated on-the-fly).""" + spec = generate_openapi_spec( + self.mass.command_handlers, server_url=self.base_url, version=self.mass.version + ) + return web.json_response(spec) + + async def _handle_swagger_ui(self, request: web.Request) -> web.Response: + """Handle request for Swagger UI.""" + swagger_html_path = os.path.join( + os.path.dirname(__file__), "..", "helpers", "resources", "swagger_ui.html" + ) + return await self._server.serve_static(swagger_html_path, request) + + async def _handle_redoc_ui(self, request: web.Request) -> web.Response: + """Handle request for ReDoc UI.""" + redoc_html_path = os.path.join( + os.path.dirname(__file__), "..", "helpers", "resources", "redoc_ui.html" + ) + return await self._server.serve_static(redoc_html_path, request) + class WebsocketClientHandler: """Handle an active websocket client connection.""" diff --git a/music_assistant/helpers/api_docs.py b/music_assistant/helpers/api_docs.py new file mode 100644 index 00000000..0a6a9d9b --- /dev/null +++ b/music_assistant/helpers/api_docs.py @@ -0,0 +1,1027 @@ +"""Helpers for generating API documentation and OpenAPI specifications.""" + +from __future__ import annotations + +import collections.abc +import inspect +from collections.abc import Callable +from dataclasses import MISSING +from datetime import datetime +from enum import Enum +from types import NoneType, UnionType +from typing import Any, Union, get_args, get_origin, get_type_hints + +from music_assistant_models.player import Player as PlayerState + +from music_assistant.helpers.api import APICommandHandler + + +def _format_type_name(type_hint: Any) -> str: + """Format a type hint as a user-friendly string, using JSON types instead of Python types.""" + if type_hint is NoneType or type_hint is type(None): + return "null" + + # Handle internal Player model - replace with PlayerState + if hasattr(type_hint, "__name__") and type_hint.__name__ == "Player": + if ( + hasattr(type_hint, "__module__") + and type_hint.__module__ == "music_assistant.models.player" + ): + return "PlayerState" + + # Map Python types to JSON types + type_name_mapping = { + "str": "string", + "int": "integer", + "float": "number", + "bool": "boolean", + "dict": "object", + "list": "array", + "tuple": "array", + "set": "array", + "frozenset": "array", + "Sequence": "array", + "UniqueList": "array", + "None": "null", + } + + if hasattr(type_hint, "__name__"): + type_name = str(type_hint.__name__) + return type_name_mapping.get(type_name, type_name) + + type_str = str(type_hint).replace("NoneType", "null") + # Replace Python types with JSON types in complex type strings + for python_type, json_type in type_name_mapping.items(): + type_str = type_str.replace(python_type, json_type) + return type_str + + +def _get_type_schema( # noqa: PLR0911, PLR0915 + type_hint: Any, definitions: dict[str, Any] +) -> dict[str, Any]: + """Convert a Python type hint to an OpenAPI schema.""" + # Handle string type hints from __future__ annotations + if isinstance(type_hint, str): + # Handle simple primitive type names + if type_hint in ("str", "string"): + return {"type": "string"} + if type_hint in ("int", "integer"): + return {"type": "integer"} + if type_hint in ("float", "number"): + return {"type": "number"} + if type_hint in ("bool", "boolean"): + return {"type": "boolean"} + + # Check if it looks like a simple class name (no special chars, starts with uppercase) + # Examples: "PlayerType", "DeviceInfo", "PlaybackState" + if type_hint.isidentifier() and type_hint[0].isupper(): + # Create a schema reference for this type + if type_hint not in definitions: + definitions[type_hint] = {"type": "object"} + return {"$ref": f"#/components/schemas/{type_hint}"} + + # For complex type expressions like "str | None", "list[str]", return generic object + return {"type": "object"} + + # Handle None type + if type_hint is NoneType or type_hint is type(None): + return {"type": "null"} + + # Handle internal Player model - replace with external PlayerState + if hasattr(type_hint, "__name__") and type_hint.__name__ == "Player": + # Check if this is the internal Player (from music_assistant.models.player) + if ( + hasattr(type_hint, "__module__") + and type_hint.__module__ == "music_assistant.models.player" + ): + # Replace with PlayerState from music_assistant_models + return _get_type_schema(PlayerState, definitions) + + # Handle Union types (including Optional) + origin = get_origin(type_hint) + if origin is Union or origin is UnionType: + args = get_args(type_hint) + # Check if it's Optional (Union with None) + non_none_args = [arg for arg in args if arg not in (NoneType, type(None))] + if (len(non_none_args) == 1 and NoneType in args) or type(None) in args: + # It's Optional[T], make it nullable + schema = _get_type_schema(non_none_args[0], definitions) + schema["nullable"] = True + return schema + # It's a union of multiple types + return {"oneOf": [_get_type_schema(arg, definitions) for arg in args]} + + # Handle UniqueList (treat as array) + if hasattr(type_hint, "__name__") and type_hint.__name__ == "UniqueList": + args = get_args(type_hint) + if args: + return {"type": "array", "items": _get_type_schema(args[0], definitions)} + return {"type": "array", "items": {}} + + # Handle Sequence types (from collections.abc or typing) + if origin is collections.abc.Sequence or ( + hasattr(origin, "__name__") and origin.__name__ == "Sequence" + ): + args = get_args(type_hint) + if args: + return {"type": "array", "items": _get_type_schema(args[0], definitions)} + return {"type": "array", "items": {}} + + # Handle set/frozenset types + if origin in (set, frozenset): + args = get_args(type_hint) + if args: + return {"type": "array", "items": _get_type_schema(args[0], definitions)} + return {"type": "array", "items": {}} + + # Handle list/tuple types + if origin in (list, tuple): + args = get_args(type_hint) + if args: + return {"type": "array", "items": _get_type_schema(args[0], definitions)} + return {"type": "array", "items": {}} + + # Handle dict types + if origin is dict: + args = get_args(type_hint) + if len(args) == 2: + return { + "type": "object", + "additionalProperties": _get_type_schema(args[1], definitions), + } + return {"type": "object", "additionalProperties": True} + + # Handle Enum types - add them to definitions as explorable objects + if inspect.isclass(type_hint) and issubclass(type_hint, Enum): + enum_name = type_hint.__name__ + if enum_name not in definitions: + enum_values = [item.value for item in type_hint] + enum_type = type(enum_values[0]).__name__ if enum_values else "string" + openapi_type = { + "str": "string", + "int": "integer", + "float": "number", + "bool": "boolean", + }.get(enum_type, "string") + + # Create a detailed enum definition with descriptions + enum_values_str = ", ".join(str(v) for v in enum_values) + definitions[enum_name] = { + "type": openapi_type, + "enum": enum_values, + "description": f"Enum: {enum_name}. Possible values: {enum_values_str}", + } + return {"$ref": f"#/components/schemas/{enum_name}"} + + # Handle datetime + if type_hint is datetime: + return {"type": "string", "format": "date-time"} + + # Handle primitive types - check both exact type and type name + if type_hint is str or (hasattr(type_hint, "__name__") and type_hint.__name__ == "str"): + return {"type": "string"} + if type_hint is int or (hasattr(type_hint, "__name__") and type_hint.__name__ == "int"): + return {"type": "integer"} + if type_hint is float or (hasattr(type_hint, "__name__") and type_hint.__name__ == "float"): + return {"type": "number"} + if type_hint is bool or (hasattr(type_hint, "__name__") and type_hint.__name__ == "bool"): + return {"type": "boolean"} + + # Handle complex types (dataclasses, models) + # Check for __annotations__ or if it's a class (not already handled above) + if hasattr(type_hint, "__annotations__") or ( + inspect.isclass(type_hint) and not issubclass(type_hint, (str, int, float, bool, Enum)) + ): + type_name = getattr(type_hint, "__name__", str(type_hint)) + # Add to definitions if not already there + if type_name not in definitions: + properties = {} + required = [] + + # Check if this is a dataclass with fields + if hasattr(type_hint, "__dataclass_fields__"): + # Resolve type hints to handle forward references from __future__ annotations + try: + resolved_hints = get_type_hints(type_hint) + except Exception: + resolved_hints = {} + + # Use dataclass fields to get proper info including defaults and metadata + for field_name, field_info in type_hint.__dataclass_fields__.items(): + # Skip fields marked with serialize="omit" in metadata + if field_info.metadata: + # Check for mashumaro field_options + if "serialize" in field_info.metadata: + if field_info.metadata["serialize"] == "omit": + continue + + # Use resolved type hint if available, otherwise fall back to field type + field_type = resolved_hints.get(field_name, field_info.type) + field_schema = _get_type_schema(field_type, definitions) + + # Add default value if present + if field_info.default is not MISSING: + field_schema["default"] = field_info.default + elif ( + hasattr(field_info, "default_factory") + and field_info.default_factory is not MISSING + ): + # Has a default factory - don't add anything, just skip + pass + + properties[field_name] = field_schema + + # Check if field is required (not Optional and no default) + has_default = field_info.default is not MISSING or ( + hasattr(field_info, "default_factory") + and field_info.default_factory is not MISSING + ) + is_optional = get_origin(field_type) in ( + Union, + UnionType, + ) and NoneType in get_args(field_type) + if not has_default and not is_optional: + required.append(field_name) + elif hasattr(type_hint, "__annotations__"): + # Fallback for non-dataclass types with annotations + for field_name, field_type in type_hint.__annotations__.items(): + properties[field_name] = _get_type_schema(field_type, definitions) + # Check if field is required (not Optional) + if not ( + get_origin(field_type) in (Union, UnionType) + and NoneType in get_args(field_type) + ): + required.append(field_name) + else: + # Class without dataclass fields or annotations - treat as generic object + pass # Will create empty properties + + definitions[type_name] = { + "type": "object", + "properties": properties, + } + if required: + definitions[type_name]["required"] = required + + return {"$ref": f"#/components/schemas/{type_name}"} + + # Handle Any + if type_hint is Any: + return {"type": "object"} + + # Fallback - for types we don't recognize, at least return a generic object type + return {"type": "object"} + + +def _parse_docstring( # noqa: PLR0915 + func: Callable[..., Any], +) -> tuple[str, str, dict[str, str]]: + """Parse docstring to extract summary, description and parameter descriptions. + + Returns: + Tuple of (short_summary, full_description, param_descriptions) + + Handles multiple docstring formats: + - reStructuredText (:param name: description) + - Google style (Args: section) + - NumPy style (Parameters section) + """ + docstring = inspect.getdoc(func) + if not docstring: + return "", "", {} + + lines = docstring.split("\n") + description_lines = [] + param_descriptions = {} + current_section = "description" + current_param = None + + for line in lines: + stripped = line.strip() + + # Check for section headers + if stripped.lower() in ("args:", "arguments:", "parameters:", "params:"): + current_section = "params" + current_param = None + continue + if stripped.lower() in ( + "returns:", + "return:", + "yields:", + "raises:", + "raises", + "examples:", + "example:", + "note:", + "notes:", + "see also:", + "warning:", + "warnings:", + ): + current_section = "other" + current_param = None + continue + + # Parse :param style + if stripped.startswith(":param "): + current_section = "params" + parts = stripped[7:].split(":", 1) + if len(parts) == 2: + current_param = parts[0].strip() + desc = parts[1].strip() + if desc: + param_descriptions[current_param] = desc + continue + + if stripped.startswith((":type ", ":rtype", ":return")): + current_section = "other" + current_param = None + continue + + # In params section, detect param lines (indented or starting with name) + if current_section == "params" and stripped: + # Google/NumPy style: "param_name: description" or "param_name (type): description" + if ":" in stripped and not stripped.startswith(" "): + # Likely a parameter definition + if "(" in stripped and ")" in stripped: + # Format: param_name (type): description + param_part = stripped.split(":")[0] + param_name = param_part.split("(")[0].strip() + desc_part = ":".join(stripped.split(":")[1:]).strip() + else: + # Format: param_name: description + parts = stripped.split(":", 1) + param_name = parts[0].strip() + desc_part = parts[1].strip() if len(parts) > 1 else "" + + if param_name and not param_name.startswith(("return", "yield", "raise")): + current_param = param_name + if desc_part: + param_descriptions[current_param] = desc_part + elif current_param and stripped: + # Continuation of previous parameter description + param_descriptions[current_param] = ( + param_descriptions.get(current_param, "") + " " + stripped + ).strip() + continue + + # Collect description lines (only before params/returns sections) + if current_section == "description" and stripped: + description_lines.append(stripped) + elif current_section == "description" and not stripped and description_lines: + # Empty line in description - keep it for paragraph breaks + description_lines.append("") + + # Join description lines, removing excessive empty lines + description = "\n".join(description_lines).strip() + # Collapse multiple empty lines into one + while "\n\n\n" in description: + description = description.replace("\n\n\n", "\n\n") + + # Extract first sentence/line as summary + summary = "" + if description: + # Get first line or first sentence (whichever is shorter) + first_line = description.split("\n")[0] + # Try to get first sentence (ending with .) + summary = first_line.split(".")[0] + "." if "." in first_line else first_line + + return summary, description, param_descriptions + + +def generate_openapi_spec( + command_handlers: dict[str, APICommandHandler], + server_url: str = "http://localhost:8095", + version: str = "1.0.0", +) -> dict[str, Any]: + """Generate OpenAPI 3.0 specification from API command handlers.""" + definitions: dict[str, Any] = {} + paths: dict[str, Any] = {} + + for command, handler in sorted(command_handlers.items()): + # Parse docstring + summary, description, param_descriptions = _parse_docstring(handler.target) + + # Build request body + request_body_properties = {} + request_body_required = [] + + for param_name, param in handler.signature.parameters.items(): + if param_name == "self": + continue + + param_type = handler.type_hints.get(param_name, Any) + param_schema = _get_type_schema(param_type, definitions) + param_description = param_descriptions.get(param_name, "") + + # Check if parameter is required + is_required = param.default is inspect.Parameter.empty + + # Add default value if present + if not is_required: + # Try to serialize the default value + try: + if param.default is None: + param_schema["default"] = None + elif isinstance(param.default, (str, int, float, bool)): + param_schema["default"] = param.default + elif isinstance(param.default, Enum): + param_schema["default"] = param.default.value + elif isinstance(param.default, (list, dict)): + param_schema["default"] = param.default + except Exception: # noqa: S110 + # If we can't serialize it, just skip the default + pass + + # Add to request body properties + request_body_properties[param_name] = { + **param_schema, + "description": param_description, + } + if is_required: + request_body_required.append(param_name) + + # Build response + return_type = handler.type_hints.get("return", Any) + response_schema = _get_type_schema(return_type, definitions) + + # Build path item + path = f"/{command}" + paths[path] = { + "post": { + "summary": summary or command, + "description": description, + "operationId": command.replace("/", "_"), + "tags": [command.split("/")[0]] if "/" in command else ["general"], + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": request_body_properties, + **( + {"required": request_body_required} + if request_body_required + else {} + ), + } + } + }, + } + if request_body_properties + else None, + "responses": { + "200": { + "description": "Successful response", + "content": {"application/json": {"schema": response_schema}}, + }, + "400": {"description": "Bad request - invalid parameters"}, + "500": {"description": "Internal server error"}, + }, + } + } + + # Remove requestBody if empty + if not request_body_properties: + del paths[path]["post"]["requestBody"] + + # Build OpenAPI spec + return { + "openapi": "3.0.0", + "info": { + "title": "Music Assistant API", + "version": version, + "description": """ +# Music Assistant API Documentation + +Music Assistant provides two ways to interact with the API: + +## WebSocket API (Recommended) +- **Endpoint:** `ws://{server}/ws` +- **Features:** + - Full API access to all commands + - Real-time event updates + - Bi-directional communication + - Best for applications that need live updates + +### WebSocket Message Format +Send commands as JSON messages: +```json +{ + "message_id": "unique-id", + "command": "command/name", + "args": { + "param1": "value1", + "param2": "value2" + } +} +``` + +Receive responses: +```json +{ + "message_id": "unique-id", + "result": { ... } +} +``` + +## REST API (Simple) +- **Endpoint:** `POST /api` +- **Features:** + - Simple HTTP POST requests + - JSON request/response + - Best for simple, incidental commands + +### REST Message Format +Send POST request to `/api` with JSON body: +```json +{ + "command": "command/name", + "args": { + "param1": "value1", + "param2": "value2" + } +} +``` + +Receive JSON response with result. + +## Authentication +Authentication is not yet implemented but will be added in a future release. + +## API Commands +All commands listed below are available via both WebSocket and REST interfaces. + """.strip(), + "contact": { + "name": "Music Assistant", + "url": "https://music-assistant.io", + }, + }, + "servers": [{"url": server_url, "description": "Music Assistant Server"}], + "paths": paths, + "components": {"schemas": definitions}, + } + + +def generate_html_docs( # noqa: PLR0915 + command_handlers: dict[str, APICommandHandler], + server_url: str = "http://localhost:8095", + version: str = "1.0.0", +) -> str: + """Generate HTML documentation from API command handlers.""" + # Group commands by category + categories: dict[str, list[tuple[str, APICommandHandler]]] = {} + for command, handler in sorted(command_handlers.items()): + category = command.split("/")[0] if "/" in command else "general" + if category not in categories: + categories[category] = [] + categories[category].append((command, handler)) + + # Start building HTML + html_parts = [ + """ + + + + + Music Assistant API Documentation + + + +
+
+

Music Assistant API Documentation

+

Version """, + version, + """

+
+ +
+

Getting Started

+

Music Assistant provides two ways to interact with the API:

+ +

🔌 WebSocket API (Recommended)

+

+ The WebSocket API provides full access to all commands + and real-time event updates. +

+
    +
  • Endpoint: ws://""", + server_url.replace("http://", "").replace("https://", ""), + """/ws
    • +
  • + Best for: Applications that need live + updates and real-time communication +
    • +
  • + Bonus: When connected, you automatically + receive event messages for state changes +
    • +
+

Sending commands:

+ 
{
+  "message_id": "unique-id-123",
+  "command": "players/all",
+  "args": {}
+}
+

Receiving events:

+

+ Once connected, you will automatically receive event messages + whenever something changes: +

+ 
{
+  "event": "player_updated",
+  "data": {
+    "player_id": "player_123",
+    ...player data...
+  }
+}
+ +

🌐 REST API (Simple)

+

+ The REST API provides a simple HTTP interface for + executing commands. +

+
    +
  • Endpoint: POST """, + server_url, + """/api
    • +
  • + Best for: Simple, incidental commands + without need for real-time updates +
    • +
+

Example request:

+ 
{
+  "command": "players/all",
+  "args": {}
+}
+ +

📥 OpenAPI Specification

+

Download the OpenAPI 3.0 specification for automated client generation:

+ Download openapi.json + +

🚀 Interactive API Explorers

+

+ Try out the API interactively with our API explorers. + Test endpoints, see live responses, and explore the full API: +

+
+ + Swagger UI Explorer + + + ReDoc Documentation + +
+ +

📡 WebSocket Events

+

+ When connected via WebSocket, you automatically receive + real-time event notifications: +

+
+ Player Events: +
    +
  • player_added - New player discovered
    • +
  • player_updated - Player state changed
    • +
  • player_removed - Player disconnected
    • +
  • player_config_updated - Player settings changed
    • +
+ + Queue Events: +
    +
  • queue_added - New queue created
    • +
  • queue_updated - Queue state changed
    • +
  • queue_items_updated - Queue content changed
    • +
  • queue_time_updated - Playback position updated
    • +
+ + Library Events: +
    +
  • media_item_added - New media added to library
    • +
  • media_item_updated - Media metadata updated
    • +
  • media_item_deleted - Media removed from library
    • +
  • media_item_played - Media playback started
    • +
+ + System Events: +
    +
  • providers_updated - Provider status changed
    • +
  • sync_tasks_updated - Sync progress updated
    • +
  • application_shutdown - Server shutting down
    • +
+
+
+ +
+

Quick Navigation

+
    +""", + ] + + # Add navigation links + for category in sorted(categories.keys()): + html_parts.append( + f'
  • {category} ' + f"({len(categories[category])} commands)
    • \n" + ) + + html_parts.append( + """
+
+""" + ) + + # Add commands by category + for category, commands in sorted(categories.items()): + html_parts.append(f'
\n') + html_parts.append(f'
{category}
\n') + + for command, handler in commands: + _, description, param_descriptions = _parse_docstring(handler.target) + + html_parts.append('
\n') + html_parts.append(f'
{command}
\n') + + if description: + html_parts.append( + f'
{description}
\n' + ) + + # Parameters + params_html = [] + for param_name, param in handler.signature.parameters.items(): + if param_name == "self": + continue + + param_type = handler.type_hints.get(param_name, Any) + is_required = param.default is inspect.Parameter.empty + param_desc = param_descriptions.get(param_name, "") + + # Format type name + type_name = _format_type_name(param_type) + if get_origin(param_type): + origin = get_origin(param_type) + args = get_args(param_type) + if origin is Union or origin is UnionType: + type_name = " | ".join(_format_type_name(arg) for arg in args) + elif origin in (list, tuple): + if args: + inner_type = _format_type_name(args[0]) + type_name = f"{origin.__name__}[{inner_type}]" + elif origin is dict: + if len(args) == 2: + key_type = _format_type_name(args[0]) + val_type = _format_type_name(args[1]) + type_name = f"dict[{key_type}, {val_type}]" + + required_badge = ( + 'required' + if is_required + else 'optional' + ) + + # Format default value + default_str = "" + if not is_required and param.default is not None: + try: + if isinstance(param.default, str): + default_str = f' = "{param.default}"' + elif isinstance(param.default, Enum): + default_str = f" = {param.default.value}" + elif isinstance(param.default, (int, float, bool, list, dict)): + default_str = f" = {param.default}" + except Exception: # noqa: S110 + pass # Can't serialize, skip default + + params_html.append( + f'
\n' + f' {param_name}\n' + f' ' + f"({type_name}{default_str})\n" + f" {required_badge}\n" + ) + if param_desc: + params_html.append( + f'
' + f"{param_desc}
\n" + ) + params_html.append("
\n") + + if params_html: + html_parts.append('
\n') + html_parts.append("

Parameters

\n") + html_parts.extend(params_html) + html_parts.append("
\n") + + # Return type + return_type = handler.type_hints.get("return", Any) + if return_type and return_type is not NoneType: + type_name = _format_type_name(return_type) + if get_origin(return_type): + origin = get_origin(return_type) + args = get_args(return_type) + if origin in (list, tuple) and args: + inner_type = _format_type_name(args[0]) + type_name = f"{origin.__name__}[{inner_type}]" + elif origin is Union or origin is UnionType: + type_name = " | ".join(_format_type_name(arg) for arg in args) + + html_parts.append('
\n') + html_parts.append("

Returns

\n") + html_parts.append( + f'
{type_name}
\n' + ) + html_parts.append("
\n") + + html_parts.append("
\n") + + html_parts.append("
\n") + + html_parts.append( + """
+ + +""" + ) + + return "".join(html_parts) diff --git a/music_assistant/helpers/resources/api_docs.html b/music_assistant/helpers/resources/api_docs.html new file mode 100644 index 00000000..ee3ee26c --- /dev/null +++ b/music_assistant/helpers/resources/api_docs.html @@ -0,0 +1,487 @@ + + + + + + Music Assistant API Documentation + + + +
+
+ +

Music Assistant API

+
Version {VERSION}
+
+ +
+
+

Welcome

+

+ Music Assistant provides a powerful API to control your music library, + manage players, and stream audio. Whether you're building a custom interface, + integrating with home automation, or creating a music app, our API gives you + complete control. +

+

+ This documentation will help you get started quickly with examples and best practices. +

+
+ +
+

API Explorers

+

+ Interactive documentation where you can test all API endpoints directly in your browser: +

+
+
+

📘 Swagger UI

+

Test endpoints interactively, see request/response examples, and try out the API

+ Open Swagger UI +
+
+

📗 ReDoc

+

Beautiful, detailed documentation with a clean three-panel layout

+ Open ReDoc +
+
+

📄 OpenAPI Spec

+

Download the OpenAPI 3.0 specification for automated client generation

+ Download JSON +
+
+
+ +
+

Quick Start

+ +

WebSocket API (Recommended)

+

+ The WebSocket API provides real-time bidirectional communication + and automatic event notifications. Perfect for applications that need live updates. +

+
+# Connect to WebSocket +ws://{SERVER_HOST}/ws + +# Send a command +{ + "message_id": "unique-id-123", + "command": "players/all", + "args": {} +} + +# Receive response +{ + "message_id": "unique-id-123", + "result": [...player data...] +} + +# Receive automatic events +{ + "event": "player_updated", + "data": {...updated player...} +} +
+ +

REST API (Simple)

+

+ For simple, one-off commands without needing real-time updates. +

+
+# Get all players +curl -X POST {BASE_URL}/api \ + -H "Content-Type: application/json" \ + -d '{ + "command": "players/all", + "args": {} + }' + +# Play media on a player +curl -X POST {BASE_URL}/api \ + -H "Content-Type: application/json" \ + -d '{ + "command": "player_queues/play_media", + "args": { + "queue_id": "player_123", + "media": ["library://track/456"] + } + }' + +# Get server info +curl {BASE_URL}/info +
+
+ +
+

WebSocket Events

+

+ When connected via WebSocket, you automatically receive real-time event notifications + for all state changes. No polling required! +

+ +
+

🔊 Player Events

+
    +
  • player_added - New player discovered
    • +
  • player_updated - Player state changed
    • +
  • player_removed - Player disconnected
    • +
  • player_config_updated - Settings changed
    • +
+
+ +
+

🎵 Queue Events

+
    +
  • queue_added - New queue created
    • +
  • queue_updated - Queue state changed
    • +
  • queue_items_updated - Content changed
    • +
  • queue_time_updated - Playback position updated
    • +
+
+ +
+

📚 Library Events

+
    +
  • media_item_added - New media added
    • +
  • media_item_updated - Metadata updated
    • +
  • media_item_deleted - Media removed
    • +
  • media_item_played - Playback started
    • +
+
+ +
+

⚙️ System Events

+
    +
  • providers_updated - Provider status changed
    • +
  • sync_tasks_updated - Sync progress updated
    • +
  • application_shutdown - Server shutting down
    • +
+
+
+ +
+

Client Libraries

+

+ Don't want to implement the API from scratch? Use our official client libraries: +

+ +

Python Client

+

+ Official Python client library with full type hints and async support: +

+
+# Install +pip install music-assistant-client + +# Usage +from music_assistant_client import MusicAssistantClient + +async with MusicAssistantClient("{SERVER_HOST}") as client: + # Get all players + players = await client.get_players() + + # Play media + await client.play_media( + queue_id="player_123", + media=["library://track/456"] + ) +
+
+ + 📦 GitHub Repository + + + 🐍 PyPI Package + +
+ +

TypeScript/JavaScript

+

+ Reference implementation in the Music Assistant frontend: +

+
+// Example from frontend code +import { MusicAssistantApi } from './api'; + +const api = new MusicAssistantApi('{SERVER_HOST}'); + +// Connect +await api.connect(); + +// Subscribe to events +api.subscribe('player_updated', (event) => { + console.log('Player updated:', event.data); +}); + +// Call commands +const players = await api.getPlayers(); +
+
+ + 📦 Frontend API Code + +
+
+ Coming Soon: A dedicated TypeScript client library is in development! + For now, you can use the frontend's API implementation as a reference. +
+
+ +
+

Best Practices

+

✓ Do:

+
    +
  • Use WebSocket API for real-time applications
    • +
  • Handle connection drops and reconnect automatically
    • +
  • Subscribe to relevant events instead of polling
    • +
  • Use unique message IDs for WebSocket commands
    • +
  • Implement proper error handling
    • +
+

✗ Don't:

+
    +
  • Poll the REST API frequently for updates (use WebSocket events instead)
    • +
  • Send commands without waiting for previous responses
    • +
  • Ignore error responses
    • +
  • Hardcode server URLs (make them configurable)
    • +
+
+ +
+

Authentication

+
+ Note: Authentication is not yet implemented but will be added + in a future release. For now, ensure your Music Assistant server is not directly + exposed to the internet. Use a VPN or reverse proxy for secure access. +
+
+
+ +
+

+ Music Assistant {VERSION} • + music-assistant.io • + GitHub +

+
+
+ + diff --git a/music_assistant/helpers/resources/redoc_ui.html b/music_assistant/helpers/resources/redoc_ui.html new file mode 100644 index 00000000..ed67b298 --- /dev/null +++ b/music_assistant/helpers/resources/redoc_ui.html @@ -0,0 +1,28 @@ + + + + + + Music Assistant API Documentation + + + + + + + + diff --git a/music_assistant/helpers/resources/swagger_ui.html b/music_assistant/helpers/resources/swagger_ui.html new file mode 100644 index 00000000..3eed9dcf --- /dev/null +++ b/music_assistant/helpers/resources/swagger_ui.html @@ -0,0 +1,61 @@ + + + + + + Music Assistant API Explorer + + + + +
+ + + + + -- 2.34.1