feat(mcp): add basic_auth to the service contract and a discovery lookup

Add basic_auth to the MCP auth schema so Jenkins can declare its
HTTP-Basic endpoint through the contract instead of around it.

Introduce the client-side renderers: mcp_authorization builds the
Authorization header per scheme, mcp_tool_server_connections builds the
Open WebUI tool-server entries and drops servers whose scheme it cannot
present, and mcp_renderable_servers filters a server list, since Jinja
select() needs a test and the repo ships no test plugins.

Replace the inline MCP_DISCOVERED_SERVERS Jinja with a mcp_servers
lookup, and carry auth and username per entry so basic_auth consumers can
build their credential.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/contributing/design/role/services/mcp.md

@@ -12,7 +12,7 @@ mcp:
   direction: server        # server | client | both
   transport: streamable_http  # streamable_http | sse
   exposure: internal       # internal | public
-  auth: bearer_token       # oidc | app_password | bearer_token | upstream_session | none
+  auth: bearer_token       # oidc | app_password | basic_auth | bearer_token | upstream_session | none
   auth_subject: user       # user | service_account | administrator | none
   implementation: native   # native | plugin | sidecar | external
   endpoint:                # required for direction: server | both

group_vars/all/16_ai.yml

@@ -11,19 +11,4 @@ LMSTUDIO_BASE_LOCAL_URL: "http://{{ lookup('config','svc-ai-lmstudio', 'services
 
 OPENROUTER_API_KEY: ""
 
-MCP_DISCOVERED_SERVERS: >-
-  {%- set out = [] -%}
-  {%- set tokens = lookup('users', 'administrator').tokens -%}
-  {%- for server in lookup('roles_with_service', 'mcp', direction='server', scope='all') -%}
-    {%- set token = tokens[server.id] | default('') | trim -%}
-    {%- set endpoint = server.endpoint -%}
-    {%- if token | length > 0 and endpoint.port and endpoint.path -%}
-      {%- set _ = out.append({
-        'id': server.id,
-        'url': 'http://' ~ endpoint.service_key ~ ':' ~ endpoint.port ~ endpoint.path,
-        'token': token,
-        'transport': server.transport | replace('_', '-'),
-      }) -%}
-    {%- endif -%}
-  {%- endfor -%}
-  {{ out }}
+MCP_DISCOVERED_SERVERS: "{{ lookup('mcp_servers') }}"

plugins/filter/mcp_authorization.py

@@ -0,0 +1,72 @@
+"""Filter ``mcp_authorization``: the Authorization header an MCP server wants.
+
+    {{ server | mcp_authorization }}          -> "Bearer <token>"
+    {{ server | mcp_authorization(env='X') }} -> "Bearer ${env:X}"
+
+``bearer_token``, ``app_password`` and ``upstream_session`` present the
+credential as a bearer. ``basic_auth`` presents ``username:token`` base64
+encoded, which cannot reference an environment variable because the encoding
+happens before the value is known.
+"""
+
+from __future__ import annotations
+
+import base64
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+BEARER_AUTHS = frozenset({"bearer_token", "app_password", "upstream_session", "oidc"})
+BASIC_AUTHS = frozenset({"basic_auth"})
+
+SUPPORTED_AUTHS = BEARER_AUTHS | BASIC_AUTHS
+
+
+def mcp_authorization(server: Mapping[str, Any], env: str | None = None) -> str:
+    """Return the Authorization header value for one discovered MCP server.
+
+    Args:
+        server: a MCP_DISCOVERED_SERVERS entry carrying auth, token, username.
+        env: name of the environment variable holding the token. Bearer schemes
+            reference it instead of inlining the secret; basic auth ignores it
+            because the credential is encoded, not substituted.
+    """
+    auth = str(server.get("auth") or "")
+    token = str(server.get("token") or "")
+
+    if auth in BASIC_AUTHS:
+        username = str(server.get("username") or "")
+        encoded = base64.b64encode(f"{username}:{token}".encode()).decode()
+        return f"Basic {encoded}"
+
+    if auth in BEARER_AUTHS:
+        return f"Bearer ${{env:{env}}}" if env else f"Bearer {token}"
+
+    raise ValueError(
+        f"MCP server {server.get('id')!r} declares auth {auth!r}, which no "
+        f"client renderer can present. Supported: {sorted(SUPPORTED_AUTHS)}."
+    )
+
+
+def mcp_authorization_is_renderable(server: Mapping[str, Any]) -> bool:
+    """Return whether a client can present this server's auth scheme at all."""
+    return str(server.get("auth") or "") in SUPPORTED_AUTHS
+
+
+def mcp_renderable_servers(servers: Any) -> list:
+    """Return only the servers whose auth a client can present.
+
+    Args:
+        servers: a MCP_DISCOVERED_SERVERS-shaped sequence, or None.
+    """
+    return [s for s in servers or [] if mcp_authorization_is_renderable(s)]
+
+
+class FilterModule:
+    def filters(self):
+        return {
+            "mcp_authorization": mcp_authorization,
+            "mcp_authorization_is_renderable": mcp_authorization_is_renderable,
+            "mcp_renderable_servers": mcp_renderable_servers,
+        }

plugins/filter/mcp_tool_server_connections.py

@@ -0,0 +1,57 @@
+"""Filter ``mcp_tool_server_connections``: Open WebUI tool-server entries.
+
+    {{ MCP_DISCOVERED_SERVERS | mcp_tool_server_connections }}
+
+Open WebUI reads ``TOOL_SERVER_CONNECTIONS`` as a JSON list validated by its
+``ToolServerConnection`` model. For ``type: mcp`` it connects to ``url``
+directly, so the discovered endpoint URL goes there whole and ``path`` stays
+empty.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+
+TOOL_SERVER_TYPE = "mcp"
+AUTH_TYPE_BEARER = "bearer"
+
+BEARER_PRESENTABLE_AUTHS = frozenset(
+    {"bearer_token", "app_password", "upstream_session", "oidc"}
+)
+
+
+def mcp_tool_server_connections(servers: Sequence[Mapping[str, Any]]) -> list[dict]:
+    """Return one Open WebUI tool-server connection per discovered MCP server.
+
+    Servers whose auth scheme Open WebUI cannot present are skipped: its
+    ToolServerConnection accepts bearer, session, system_oauth and oauth_2.1,
+    so a basic-auth endpoint would be registered with an unusable credential.
+    """
+    connections = []
+    for server in servers or []:
+        server_id = str(server.get("id", "")).strip()
+        url = str(server.get("url", "")).strip()
+        if not server_id or not url:
+            continue
+        if str(server.get("auth") or "") not in BEARER_PRESENTABLE_AUTHS:
+            continue
+        connections.append(
+            {
+                "url": url,
+                "path": "",
+                "type": TOOL_SERVER_TYPE,
+                "auth_type": AUTH_TYPE_BEARER,
+                "key": str(server.get("token", "")),
+                "config": {"enable": True, "access_control": None},
+                "info": {"id": server_id, "name": server_id},
+            }
+        )
+    return connections
+
+
+class FilterModule:
+    def filters(self):
+        return {"mcp_tool_server_connections": mcp_tool_server_connections}

plugins/lookup/mcp_servers.py

@@ -0,0 +1,78 @@
+"""Lookup ``mcp_servers``: the MCP servers a client role can connect to.
+
+    {{ lookup('mcp_servers') }}
+
+Returns one entry per deployed MCP server role the administrator holds a
+token for, shaped for the client renderers ``mcp_authorization`` and
+``mcp_tool_server_connections``.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+
+
+def build_mcp_servers(
+    servers: Sequence[Mapping[str, Any]] | None,
+    administrator: Mapping[str, Any],
+) -> list[dict[str, Any]]:
+    """Return the connectable MCP servers among the discovered ones.
+
+    Args:
+        servers: ``roles_with_service('mcp', direction='server')`` entries.
+        administrator: the administrator user, carrying tokens and username.
+    """
+    tokens = administrator.get("tokens") or {}
+    username = administrator.get("username")
+    connectable = []
+    for server in servers or []:
+        server_id = str(server.get("id") or "")
+        token = str(tokens.get(server_id) or "").strip()
+        endpoint = server.get("endpoint") or {}
+        port = endpoint.get("port")
+        path = endpoint.get("path")
+        if not server_id or not token or not port or not path:
+            continue
+        connectable.append(
+            {
+                "id": server_id,
+                "url": f"http://{endpoint.get('service_key')}:{port}{path}",
+                "token": token,
+                "auth": server.get("auth"),
+                "username": username,
+                "transport": str(server.get("transport") or "").replace("_", "-"),
+            }
+        )
+    return connectable
+
+
+class LookupModule(LookupBase):
+    def run(
+        self,
+        terms: Sequence[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[Any]:
+        if terms:
+            raise AnsibleError(
+                "mcp_servers: expected no terms — lookup('mcp_servers')"
+            )
+
+        vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
+        templar = getattr(self, "_templar", None)
+
+        servers = lookup_loader.get(
+            "roles_with_service", loader=self._loader, templar=templar
+        ).run(["mcp"], variables=vars_, direction="server", scope="all")[0]
+        administrator = lookup_loader.get(
+            "users", loader=self._loader, templar=templar
+        ).run(["administrator"], variables=vars_)[0]
+
+        return [build_mcp_servers(servers, administrator)]

tests/unit/plugins/filter/test_mcp_authorization.py

@@ -0,0 +1,66 @@
+import base64
+import importlib
+import unittest
+
+plugin_module = importlib.import_module("plugins.filter.mcp_authorization")
+mcp_authorization = plugin_module.mcp_authorization
+mcp_authorization_is_renderable = plugin_module.mcp_authorization_is_renderable
+mcp_renderable_servers = plugin_module.mcp_renderable_servers
+
+
+class TestMcpAuthorization(unittest.TestCase):
+    def test_bearer_token_inlines_the_token(self):
+        header = mcp_authorization({"auth": "bearer_token", "token": "secret"})
+        self.assertEqual(header, "Bearer secret")
+
+    def test_bearer_token_references_the_env_var_when_asked(self):
+        header = mcp_authorization(
+            {"auth": "bearer_token", "token": "secret"}, env="HA_MCP_TOKEN"
+        )
+        self.assertEqual(header, "Bearer ${env:HA_MCP_TOKEN}")
+        self.assertNotIn("secret", header)
+
+    def test_app_password_is_presented_as_a_bearer(self):
+        self.assertEqual(
+            mcp_authorization({"auth": "app_password", "token": "pw"}), "Bearer pw"
+        )
+
+    def test_basic_auth_encodes_username_and_token(self):
+        header = mcp_authorization(
+            {"auth": "basic_auth", "username": "admin", "token": "t0ken"}
+        )
+        expected = base64.b64encode(b"admin:t0ken").decode()
+        self.assertEqual(header, f"Basic {expected}")
+
+    def test_basic_auth_ignores_the_env_reference(self):
+        header = mcp_authorization(
+            {"auth": "basic_auth", "username": "admin", "token": "t0ken"},
+            env="SOME_VAR",
+        )
+        self.assertNotIn("SOME_VAR", header)
+        self.assertTrue(header.startswith("Basic "))
+
+    def test_unpresentable_scheme_raises(self):
+        with self.assertRaises(ValueError):
+            mcp_authorization({"id": "web-app-x", "auth": "none", "token": ""})
+
+    def test_renderable_reports_supported_schemes(self):
+        self.assertTrue(mcp_authorization_is_renderable({"auth": "bearer_token"}))
+        self.assertTrue(mcp_authorization_is_renderable({"auth": "basic_auth"}))
+        self.assertFalse(mcp_authorization_is_renderable({"auth": "none"}))
+        self.assertFalse(mcp_authorization_is_renderable({}))
+
+
+class TestMcpRenderableServers(unittest.TestCase):
+    def test_none_yields_empty_list(self):
+        self.assertEqual(mcp_renderable_servers(None), [])
+
+    def test_unpresentable_servers_are_dropped(self):
+        bearer = {"id": "a", "auth": "bearer_token"}
+        basic = {"id": "b", "auth": "basic_auth"}
+        servers = [bearer, {"id": "c", "auth": "none"}, basic, {"id": "d"}]
+        self.assertEqual(mcp_renderable_servers(servers), [bearer, basic])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/plugins/filter/test_mcp_tool_server_connections.py

@@ -0,0 +1,90 @@
+import importlib
+import unittest
+
+plugin_module = importlib.import_module("plugins.filter.mcp_tool_server_connections")
+mcp_tool_server_connections = plugin_module.mcp_tool_server_connections
+
+
+class TestMcpToolServerConnections(unittest.TestCase):
+    def test_empty_input_yields_no_connections(self):
+        self.assertEqual(mcp_tool_server_connections([]), [])
+
+    def test_none_input_yields_no_connections(self):
+        self.assertEqual(mcp_tool_server_connections(None), [])
+
+    def test_server_becomes_an_mcp_bearer_connection(self):
+        result = mcp_tool_server_connections(
+            [
+                {
+                    "id": "web-app-homeassistant",
+                    "url": "http://homeassistant:8123/api/mcp",
+                    "token": "secret",
+                    "auth": "bearer_token",
+                    "transport": "streamable-http",
+                }
+            ]
+        )
+
+        self.assertEqual(len(result), 1)
+        entry = result[0]
+        self.assertEqual(entry["url"], "http://homeassistant:8123/api/mcp")
+        self.assertEqual(entry["path"], "")
+        self.assertEqual(entry["type"], "mcp")
+        self.assertEqual(entry["auth_type"], "bearer")
+        self.assertEqual(entry["key"], "secret")
+        self.assertTrue(entry["config"]["enable"])
+        self.assertEqual(entry["info"]["id"], "web-app-homeassistant")
+
+    def test_entry_without_url_is_dropped(self):
+        self.assertEqual(
+            mcp_tool_server_connections(
+                [{"id": "x", "url": "", "token": "t", "auth": "bearer_token"}]
+            ),
+            [],
+        )
+
+    def test_entry_without_id_is_dropped(self):
+        self.assertEqual(
+            mcp_tool_server_connections(
+                [{"id": "", "url": "http://a/mcp", "auth": "bearer_token"}]
+            ),
+            [],
+        )
+
+    def test_basic_auth_server_is_skipped(self):
+        self.assertEqual(
+            mcp_tool_server_connections(
+                [
+                    {
+                        "id": "web-app-jenkins",
+                        "url": "http://jenkins:8080/mcp-server/mcp",
+                        "token": "t",
+                        "auth": "basic_auth",
+                    }
+                ]
+            ),
+            [],
+        )
+
+    def test_every_discovered_server_is_kept(self):
+        result = mcp_tool_server_connections(
+            [
+                {
+                    "id": "a",
+                    "url": "http://a/mcp",
+                    "token": "1",
+                    "auth": "bearer_token",
+                },
+                {
+                    "id": "b",
+                    "url": "http://b/mcp",
+                    "token": "2",
+                    "auth": "app_password",
+                },
+            ]
+        )
+        self.assertEqual([e["info"]["id"] for e in result], ["a", "b"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/plugins/lookup/test_mcp_servers.py

@@ -0,0 +1,71 @@
+"""Unit tests for ``plugins/lookup/mcp_servers.py``.
+
+Only the pure builder is exercised; the lookup wrapper just forwards the
+``roles_with_service`` and ``users`` results into it.
+"""
+
+import importlib
+import unittest
+
+plugin_module = importlib.import_module("plugins.lookup.mcp_servers")
+build_mcp_servers = plugin_module.build_mcp_servers
+
+ADMINISTRATOR = {
+    "username": "administrator",
+    "tokens": {"web-app-homeassistant": "ha-token", "web-app-jenkins": "jk-token"},
+}
+
+HOMEASSISTANT = {
+    "id": "web-app-homeassistant",
+    "auth": "bearer_token",
+    "transport": "streamable_http",
+    "endpoint": {"service_key": "homeassistant", "port": 8123, "path": "/api/mcp"},
+}
+
+
+class TestBuildMcpServers(unittest.TestCase):
+    def test_no_servers_yields_no_entries(self):
+        self.assertEqual(build_mcp_servers(None, ADMINISTRATOR), [])
+        self.assertEqual(build_mcp_servers([], ADMINISTRATOR), [])
+
+    def test_server_becomes_a_container_network_entry(self):
+        entry = build_mcp_servers([HOMEASSISTANT], ADMINISTRATOR)[0]
+        self.assertEqual(entry["id"], "web-app-homeassistant")
+        self.assertEqual(entry["url"], "http://homeassistant:8123/api/mcp")
+        self.assertEqual(entry["token"], "ha-token")
+        self.assertEqual(entry["auth"], "bearer_token")
+        self.assertEqual(entry["username"], "administrator")
+
+    def test_transport_is_rendered_with_dashes(self):
+        entry = build_mcp_servers([HOMEASSISTANT], ADMINISTRATOR)[0]
+        self.assertEqual(entry["transport"], "streamable-http")
+
+    def test_server_without_a_token_is_dropped(self):
+        untokened = dict(HOMEASSISTANT, id="web-app-baserow")
+        self.assertEqual(build_mcp_servers([untokened], ADMINISTRATOR), [])
+
+    def test_blank_token_is_dropped(self):
+        administrator = dict(ADMINISTRATOR, tokens={"web-app-homeassistant": "  "})
+        self.assertEqual(build_mcp_servers([HOMEASSISTANT], administrator), [])
+
+    def test_incomplete_endpoint_is_dropped(self):
+        no_port = dict(HOMEASSISTANT, endpoint={"service_key": "x", "path": "/mcp"})
+        no_path = dict(HOMEASSISTANT, endpoint={"service_key": "x", "port": 80})
+        self.assertEqual(build_mcp_servers([no_port, no_path], ADMINISTRATOR), [])
+
+    def test_order_of_the_discovered_servers_is_kept(self):
+        jenkins = {
+            "id": "web-app-jenkins",
+            "auth": "basic_auth",
+            "transport": "streamable_http",
+            "endpoint": {"service_key": "jenkins", "port": 8080, "path": "/mcp"},
+        }
+        result = build_mcp_servers([jenkins, HOMEASSISTANT], ADMINISTRATOR)
+        self.assertEqual(
+            [e["id"] for e in result],
+            ["web-app-jenkins", "web-app-homeassistant"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/roles/applications/services/mcp.py

@@ -24,7 +24,7 @@ MCP_DIRECTIONS = frozenset({"server", "client", "both"})
 MCP_TRANSPORTS = frozenset({"streamable_http", "sse"})
 MCP_EXPOSURES = frozenset({"internal", "public"})
 MCP_AUTHS = frozenset(
-    {"oidc", "app_password", "bearer_token", "upstream_session", "none"}
+    {"oidc", "app_password", "basic_auth", "bearer_token", "upstream_session", "none"}
 )
 MCP_AUTH_SUBJECTS = frozenset({"user", "service_account", "administrator", "none"})
 MCP_IMPLEMENTATIONS = frozenset({"native", "plugin", "sidecar", "external"})