/
/
/
1"""
2Mount the Connect Wizard endpoints onto MA's webserver.
3
4Five routes are registered under ``<mount_path>/connect``; the returned
5callable removes all of them when invoked (called from
6:meth:`provider.server.MCPServerRuntime.stop`).
7"""
8
9from __future__ import annotations
10
11import contextlib
12from typing import TYPE_CHECKING
13
14from ..origins import compute_origin_allowlist, is_origin_allowed_for_request # noqa: TID252
15from .handlers import (
16 WizardContext,
17 make_exchange,
18 make_info,
19 make_login,
20 make_mint_token,
21 make_serve_page,
22)
23
24if TYPE_CHECKING:
25 from collections.abc import Callable
26
27 from music_assistant.mass import MusicAssistant
28
29
30async def mount_connect_wizard(
31 mass: MusicAssistant,
32 mount_path: str,
33 *,
34 enabled_tags_provider: Callable[[], list[str]],
35 extra_origins_csv: str = "",
36 trust_forwarded_proto: bool = False,
37) -> Callable[[], None]:
38 """
39 Register the wizard routes and return a callable that unregisters them.
40
41 :param mass: MusicAssistant instance.
42 :param mount_path: HTTP path prefix where the MCP server is mounted
43 (e.g. ``/mcp/v1``); wizard routes nest under ``<mount_path>/connect``.
44 :param enabled_tags_provider: Zero-arg callable returning the list of
45 currently-enabled permission tag strings; called per-request so
46 permission hot-swaps surface in the UI without remount.
47 :param extra_origins_csv: Comma-separated additional ``Origin`` values to
48 accept beyond the auto-derived loopback + base_url + publish_ip set.
49 :param trust_forwarded_proto: When True, accept a trusted reverse proxy's
50 ``X-Forwarded-Proto: https`` as proof the public hop was HTTPS, so the
51 credential-bearing endpoints work behind a TLS-terminating proxy.
52 :return: Callable that, when invoked, unregisters every wizard route.
53 """
54 allowlist = compute_origin_allowlist(mass, extra_origins_csv)
55 ctx = WizardContext(
56 mass=mass,
57 mount_path=mount_path,
58 enabled_tags_provider=enabled_tags_provider,
59 origin_check=lambda request: is_origin_allowed_for_request(request, allowlist),
60 trust_forwarded_proto=trust_forwarded_proto,
61 )
62
63 base = "/" + mount_path.strip("/")
64 routes: list[tuple[str, str]] = [
65 (f"{base}/connect", "GET"),
66 (f"{base}/connect/info", "GET"),
67 (f"{base}/connect/exchange", "POST"),
68 (f"{base}/connect/login", "POST"),
69 (f"{base}/connect/token", "POST"),
70 ]
71 handlers = [
72 make_serve_page(ctx),
73 make_info(ctx),
74 make_exchange(ctx),
75 make_login(ctx),
76 make_mint_token(ctx),
77 ]
78
79 unregister_fns: list[Callable[[], None]] = []
80 for (path, method), handler in zip(routes, handlers, strict=True):
81 unregister_fns.append(mass.webserver.register_dynamic_route(path, handler, method=method))
82
83 def _unregister_all() -> None:
84 for fn in unregister_fns:
85 with contextlib.suppress(Exception):
86 fn()
87
88 return _unregister_all
89