feat(mcp): add semantic layer MCP tools (list_metrics, get_table, get_compatible_dimensions, get_compatible_metrics) (#41611)

tests/unit_tests/mcp_service/semantic_layer/tool/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

tests/unit_tests/mcp_service/semantic_layer/tool/test_get_compatible_dimensions.py

@@ -0,0 +1,284 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for the get_compatible_dimensions MCP tool."""
+
+from __future__ import annotations
+
+import importlib
+from collections.abc import Generator
+from types import ModuleType
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from fastmcp import Client, FastMCP
+
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetSecurityException
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+get_compatible_dimensions_module: ModuleType = importlib.import_module(
+    "superset.mcp_service.semantic_layer.tool.get_compatible_dimensions"
+)
+
+
+@pytest.fixture
+def mcp_server() -> FastMCP:
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth() -> Generator[MagicMock, None, None]:
+    with (
+        patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user,
+        patch.object(
+            get_compatible_dimensions_module,
+            "user_can_view_data_model_metadata",
+            return_value=True,
+        ),
+    ):
+        mock_user = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+def _make_column(name: str, groupby: bool = True) -> MagicMock:
+    col: MagicMock = MagicMock()
+    col.column_name = name
+    col.verbose_name = None
+    col.description = None
+    col.type = "VARCHAR"
+    col.is_dttm = False
+    col.groupby = groupby
+    col.filterable = True
+    return col
+
+
+def _make_metric(name: str) -> MagicMock:
+    m = MagicMock()
+    m.metric_name = name
+    return m
+
+
+def _make_dataset(dataset_id: int = 42) -> MagicMock:
+    ds: MagicMock = MagicMock()
+    ds.id = dataset_id
+    ds.table_name = f"table_{dataset_id}"
+    ds.metrics = [_make_metric("revenue")]
+    ds.columns = [
+        _make_column("region"),
+        _make_column("category"),
+        _make_column("internal_only", groupby=False),
+    ]
+    return ds
+
+
+def _make_view(view_id: int = 5) -> MagicMock:
+    view: MagicMock = MagicMock()
+    view.id = view_id
+    view.name = f"view_{view_id}"
+    view.raise_for_access = MagicMock(return_value=None)
+    view.columns = [_make_column("country_name")]
+    view.get_compatible_dimensions = MagicMock(return_value=["country_name"])
+    return view
+
+
+def _access_denied_exc(message: str = "Access denied") -> SupersetSecurityException:
+    return SupersetSecurityException(
+        SupersetError(
+            message=message,
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_builtin_happy_path(
+    mcp_server: FastMCP,
+) -> None:
+    """Builtin datasets return all groupby-enabled columns for a valid selection."""
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions",
+                {"request": {"dataset_id": 42, "selected_metrics": ["revenue"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    assert data["source"] == "builtin"
+    names = {d["name"] for d in data["compatible_dimensions"]}
+    assert names == {"region", "category"}
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_builtin_unknown_selection(
+    mcp_server: FastMCP,
+) -> None:
+    """Builtin datasets reject unknown selected metric/dimension names."""
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions",
+                {
+                    "request": {
+                        "dataset_id": 42,
+                        "selected_metrics": ["bogus_metric"],
+                        "selected_dimensions": ["bogus_dim"],
+                    }
+                },
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+    assert "Unknown metric: 'bogus_metric'" in data["error"]
+    assert "Unknown dimension: 'bogus_dim'" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_external_happy_path(
+    mcp_server: FastMCP,
+) -> None:
+    """External views delegate to view.get_compatible_dimensions()."""
+    mock_view = _make_view(5)
+
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=mock_view,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions",
+                {"request": {"view_id": 5, "selected_metrics": ["bookings"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    assert data["source"] == "external"
+    assert [d["name"] for d in data["compatible_dimensions"]] == ["country_name"]
+    mock_view.get_compatible_dimensions.assert_called_once_with(["bookings"], [])
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_mutual_exclusion_validation(
+    mcp_server: FastMCP,
+) -> None:
+    """Errors when both dataset_id and view_id are provided."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_compatible_dimensions",
+            {"request": {"dataset_id": 1, "view_id": 2}},
+        )
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_requires_one_source(
+    mcp_server: FastMCP,
+) -> None:
+    """Errors when neither dataset_id nor view_id is provided."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("get_compatible_dimensions", {"request": {}})
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_privacy_check(mcp_server: FastMCP) -> None:
+    """Errors when the user lacks data-model metadata access."""
+    with patch.object(
+        get_compatible_dimensions_module,
+        "user_can_view_data_model_metadata",
+        return_value=False,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions", {"request": {"dataset_id": 1}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "DataModelMetadataRestricted"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_external_access_denied(
+    mcp_server: FastMCP,
+) -> None:
+    """Returns AccessDenied when raise_for_access rejects the view."""
+    mock_view = _make_view(5)
+    mock_view.raise_for_access.side_effect = _access_denied_exc()
+
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=mock_view,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions", {"request": {"view_id": 5}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "AccessDenied"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_not_found(mcp_server: FastMCP) -> None:
+    """Returns NotFound when the dataset doesn't exist."""
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=None):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions", {"request": {"dataset_id": 999}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "NotFound"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_dimensions_external_not_found(
+    mcp_server: FastMCP,
+) -> None:
+    """Returns NotFound when the semantic view doesn't exist."""
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=None,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_dimensions", {"request": {"view_id": 999}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "NotFound"

tests/unit_tests/mcp_service/semantic_layer/tool/test_get_compatible_metrics.py

@@ -0,0 +1,280 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for the get_compatible_metrics MCP tool."""
+
+from __future__ import annotations
+
+import importlib
+from collections.abc import Generator
+from types import ModuleType
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from fastmcp import Client, FastMCP
+
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetSecurityException
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+get_compatible_metrics_module: ModuleType = importlib.import_module(
+    "superset.mcp_service.semantic_layer.tool.get_compatible_metrics"
+)
+
+
+@pytest.fixture
+def mcp_server() -> FastMCP:
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth() -> Generator[MagicMock, None, None]:
+    with (
+        patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user,
+        patch.object(
+            get_compatible_metrics_module,
+            "user_can_view_data_model_metadata",
+            return_value=True,
+        ),
+    ):
+        mock_user: Mock = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+def _make_metric(name: str, expression: str = "COUNT(*)") -> MagicMock:
+    m: MagicMock = MagicMock()
+    m.metric_name = name
+    m.verbose_name = None
+    m.expression = expression
+    m.description = None
+    m.d3format = None
+    m.warning_text = None
+    return m
+
+
+def _make_column(name: str) -> MagicMock:
+    col = MagicMock()
+    col.column_name = name
+    return col
+
+
+def _make_dataset(dataset_id: int = 42) -> MagicMock:
+    ds: MagicMock = MagicMock()
+    ds.id = dataset_id
+    ds.table_name = f"table_{dataset_id}"
+    ds.columns = [_make_column("region")]
+    ds.metrics = [_make_metric("count"), _make_metric("revenue", "SUM(revenue)")]
+    return ds
+
+
+def _make_view(view_id: int = 5) -> MagicMock:
+    view: MagicMock = MagicMock()
+    view.id = view_id
+    view.name = f"view_{view_id}"
+    view.raise_for_access = MagicMock(return_value=None)
+    view.metrics = [_make_metric("bookings")]
+    view.get_compatible_metrics = MagicMock(return_value=["bookings"])
+    return view
+
+
+def _access_denied_exc(message: str = "Access denied") -> SupersetSecurityException:
+    return SupersetSecurityException(
+        SupersetError(
+            message=message,
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_builtin_happy_path(mcp_server: FastMCP) -> None:
+    """Builtin datasets return all metrics for a valid dimension selection."""
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics",
+                {"request": {"dataset_id": 42, "selected_dimensions": ["region"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    assert data["source"] == "builtin"
+    names = {m["name"] for m in data["compatible_metrics"]}
+    assert names == {"count", "revenue"}
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_builtin_excludes_selected_metrics(
+    mcp_server: FastMCP,
+) -> None:
+    """Builtin datasets exclude metrics already in selected_metrics.
+
+    Regression test: previously all dataset metrics were returned unfiltered,
+    so already-selected metrics were suggested again as "compatible".
+    """
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics",
+                {"request": {"dataset_id": 42, "selected_metrics": ["count"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    names = {m["name"] for m in data["compatible_metrics"]}
+    assert names == {"revenue"}
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_builtin_unknown_selection(
+    mcp_server: FastMCP,
+) -> None:
+    """Builtin datasets reject unknown selected metric/dimension names."""
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics",
+                {
+                    "request": {
+                        "dataset_id": 42,
+                        "selected_metrics": ["bogus_metric"],
+                        "selected_dimensions": ["bogus_dim"],
+                    }
+                },
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+    assert "Unknown metric: 'bogus_metric'" in data["error"]
+    assert "Unknown dimension: 'bogus_dim'" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_external_happy_path(mcp_server: FastMCP) -> None:
+    """External views delegate to view.get_compatible_metrics()."""
+    mock_view = _make_view(5)
+
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=mock_view,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics",
+                {"request": {"view_id": 5, "selected_dimensions": ["country_name"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    assert data["source"] == "external"
+    assert [m["name"] for m in data["compatible_metrics"]] == ["bookings"]
+    mock_view.get_compatible_metrics.assert_called_once_with([], ["country_name"])
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_mutual_exclusion_validation(
+    mcp_server: FastMCP,
+) -> None:
+    """Errors when both dataset_id and view_id are provided."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_compatible_metrics",
+            {"request": {"dataset_id": 1, "view_id": 2}},
+        )
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_requires_one_source(
+    mcp_server: FastMCP,
+) -> None:
+    """Errors when neither dataset_id nor view_id is provided."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("get_compatible_metrics", {"request": {}})
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_privacy_check(mcp_server: FastMCP) -> None:
+    """Errors when the user lacks data-model metadata access."""
+    with patch.object(
+        get_compatible_metrics_module,
+        "user_can_view_data_model_metadata",
+        return_value=False,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics", {"request": {"dataset_id": 1}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "DataModelMetadataRestricted"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_external_access_denied(
+    mcp_server: FastMCP,
+) -> None:
+    """Returns AccessDenied when raise_for_access rejects the view."""
+    mock_view = _make_view(5)
+    mock_view.raise_for_access.side_effect = _access_denied_exc()
+
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=mock_view,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics", {"request": {"view_id": 5}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "AccessDenied"
+
+
+@pytest.mark.asyncio
+async def test_get_compatible_metrics_not_found(mcp_server: FastMCP) -> None:
+    """Returns NotFound when the dataset doesn't exist."""
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=None):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_compatible_metrics", {"request": {"dataset_id": 999}}
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "NotFound"

tests/unit_tests/mcp_service/semantic_layer/tool/test_get_table.py

@@ -0,0 +1,299 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for the get_table MCP tool."""
+
+from __future__ import annotations
+
+import importlib
+from collections.abc import Generator
+from types import ModuleType
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from fastmcp import Client, FastMCP
+
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetSecurityException
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+get_table_module: ModuleType = importlib.import_module(
+    "superset.mcp_service.semantic_layer.tool.get_table"
+)
+
+
+@pytest.fixture
+def mcp_server() -> FastMCP:
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth() -> Generator[MagicMock, None, None]:
+    with (
+        patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user,
+        patch.object(
+            get_table_module,
+            "user_can_view_data_model_metadata",
+            return_value=True,
+        ),
+    ):
+        mock_user = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+def _make_metric(name: str, expression: str = "COUNT(*)") -> MagicMock:
+    m = MagicMock()
+    m.metric_name = name
+    m.verbose_name = None
+    m.expression = expression
+    m.description = None
+    m.d3format = None
+    m.warning_text = None
+    return m
+
+
+def _make_column(name: str, is_dttm: bool = False) -> MagicMock:
+    col = MagicMock()
+    col.column_name = name
+    col.verbose_name = None
+    col.description = None
+    col.type = "VARCHAR"
+    col.is_dttm = is_dttm
+    col.groupby = True
+    col.filterable = True
+    return col
+
+
+def _make_dataset(dataset_id: int = 42) -> MagicMock:
+    ds = MagicMock()
+    ds.id = dataset_id
+    ds.table_name = f"table_{dataset_id}"
+    ds.main_dttm_col = "created_at"
+    ds.metrics = [_make_metric("revenue", "SUM(revenue)")]
+    ds.columns = [
+        _make_column("region"),
+        _make_column("created_at", is_dttm=True),
+    ]
+    return ds
+
+
+def _make_view(view_id: int = 5) -> MagicMock:
+    view = MagicMock()
+    view.id = view_id
+    view.name = f"view_{view_id}"
+    view.raise_for_access = MagicMock(return_value=None)
+    view.metrics = [_make_metric("bookings")]
+    view.columns = [_make_column("country_name")]
+    return view
+
+
+def _access_denied_exc(message: str = "Access denied") -> SupersetSecurityException:
+    return SupersetSecurityException(
+        SupersetError(
+            message=message,
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_table_builtin_happy_path(mcp_server: FastMCP) -> None:
+    """get_table returns tabular data for a built-in dataset."""
+    mock_ds = _make_dataset(42)
+    query_result = {
+        "queries": [
+            {
+                "data": [{"region": "west", "revenue": 100}],
+                "colnames": ["region", "revenue"],
+                "rowcount": 1,
+            }
+        ]
+    }
+
+    with (
+        patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds),
+        patch(
+            "superset.commands.chart.data.get_data_command.ChartDataCommand"
+        ) as mock_command_cls,
+        patch(
+            "superset.common.query_context_factory.QueryContextFactory"
+        ) as mock_factory_cls,
+    ):
+        mock_command_cls.return_value.run.return_value = query_result
+        mock_factory_cls.return_value.create.return_value = MagicMock()
+
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_table",
+                {
+                    "request": {
+                        "dataset_id": 42,
+                        "metrics": ["revenue"],
+                        "dimensions": ["region"],
+                    }
+                },
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    assert data["row_count"] == 1
+    assert data["source"] == "builtin"
+    assert data["dataset_id"] == 42
+
+
+@pytest.mark.asyncio
+async def test_get_table_requires_one_source(mcp_server: FastMCP) -> None:
+    """get_table errors when neither dataset_id nor view_id is provided."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("get_table", {"request": {}})
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_get_table_mutual_exclusion_validation(mcp_server: FastMCP) -> None:
+    """get_table errors when both dataset_id and view_id are provided."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_table", {"request": {"dataset_id": 1, "view_id": 2}}
+        )
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_get_table_privacy_check(mcp_server: FastMCP) -> None:
+    """get_table errors when the user lacks data-model metadata access."""
+    with patch.object(
+        get_table_module,
+        "user_can_view_data_model_metadata",
+        return_value=False,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool("get_table", {"request": {"dataset_id": 1}})
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "DataModelMetadataRestricted"
+
+
+@pytest.mark.asyncio
+async def test_get_table_unknown_metric_validation_error(mcp_server: FastMCP) -> None:
+    """get_table errors when a requested metric doesn't exist on the dataset."""
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_table",
+                {"request": {"dataset_id": 42, "metrics": ["does_not_exist"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+    assert "Valid metrics: revenue" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_get_table_time_column_not_dttm_validation_error(
+    mcp_server: FastMCP,
+) -> None:
+    """get_table rejects a time_column that isn't marked as a datetime column."""
+    mock_ds = _make_dataset(42)
+
+    with patch("superset.daos.dataset.DatasetDAO.find_by_id", return_value=mock_ds):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_table",
+                {
+                    "request": {
+                        "dataset_id": 42,
+                        "metrics": ["revenue"],
+                        "time_column": "region",
+                    }
+                },
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+    assert "not marked as a datetime column" in data["message"]
+
+
+@pytest.mark.asyncio
+async def test_get_table_external_view_access_denied(mcp_server: FastMCP) -> None:
+    """get_table returns AccessDenied when raise_for_access rejects the view."""
+    mock_view = _make_view(5)
+    mock_view.raise_for_access.side_effect = _access_denied_exc()
+
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=mock_view,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_table",
+                {"request": {"view_id": 5, "metrics": ["bookings"]}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "AccessDenied"
+
+
+@pytest.mark.asyncio
+async def test_get_table_external_time_range_without_dttm_validation_error(
+    mcp_server: FastMCP,
+) -> None:
+    """get_table rejects time_range on a view with no datetime dimension.
+
+    Regression test: previously this silently dropped the time filter and
+    ran an unfiltered query instead of erroring, which could return
+    incorrect data for a time-bounded request.
+    """
+    mock_view = _make_view(5)  # columns have no is_dttm=True column
+
+    with patch(
+        "superset.daos.semantic_layer.SemanticViewDAO.find_by_id",
+        return_value=mock_view,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_table",
+                {
+                    "request": {
+                        "view_id": 5,
+                        "metrics": ["bookings"],
+                        "time_range": "Last 30 days",
+                    }
+                },
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+    assert "no datetime dimension" in data["message"]

tests/unit_tests/mcp_service/semantic_layer/tool/test_list_metrics.py

@@ -0,0 +1,329 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for the list_metrics MCP tool."""
+
+from __future__ import annotations
+
+import importlib
+from collections.abc import Generator
+from types import ModuleType
+from unittest.mock import call, MagicMock, Mock, patch
+
+import pytest
+from fastmcp import Client, FastMCP
+
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetSecurityException
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+list_metrics_module: ModuleType = importlib.import_module(
+    "superset.mcp_service.semantic_layer.tool.list_metrics"
+)
+
+
+@pytest.fixture
+def mcp_server() -> FastMCP:
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth() -> Generator[MagicMock, None, None]:
+    with (
+        patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user,
+        patch.object(
+            list_metrics_module,
+            "user_can_view_data_model_metadata",
+            return_value=True,
+        ),
+    ):
+        mock_user: Mock = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+def _make_metric(name: str, expression: str = "COUNT(*)") -> MagicMock:
+    m = MagicMock()
+    m.metric_name = name
+    m.verbose_name = None
+    m.expression = expression
+    m.description = None
+    m.d3format = None
+    m.warning_text = None
+    return m
+
+
+def _make_column(name: str) -> MagicMock:
+    col = MagicMock()
+    col.column_name = name
+    col.verbose_name = None
+    col.description = None
+    col.type = "VARCHAR"
+    col.is_dttm = False
+    col.groupby = True
+    col.filterable = True
+    return col
+
+
+def _make_dataset(dataset_id: int = 1) -> MagicMock:
+    ds = MagicMock()
+    ds.id = dataset_id
+    ds.table_name = f"table_{dataset_id}"
+    ds.metrics = [_make_metric("count"), _make_metric("revenue", "SUM(revenue)")]
+    ds.columns = [_make_column("region"), _make_column("category")]
+    return ds
+
+
+def _make_view(view_id: int = 5) -> MagicMock:
+    view = MagicMock()
+    view.id = view_id
+    view.name = f"view_{view_id}"
+    view.raise_for_access = MagicMock(return_value=None)
+    view.metrics = [_make_metric("bookings"), _make_metric("revenue", "SUM(revenue)")]
+    view.columns = [_make_column("listing__country_name"), _make_column("channel")]
+    view.get_compatible_dimensions = MagicMock(return_value=["listing__country_name"])
+    return view
+
+
+def _access_denied_exc(message: str = "Access denied") -> SupersetSecurityException:
+    return SupersetSecurityException(
+        SupersetError(
+            message=message,
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_builtin_happy_path(mcp_server: FastMCP) -> None:
+    """list_metrics returns builtin metrics when only datasets exist."""
+    mock_ds = _make_dataset(42)
+
+    with (
+        patch(
+            "superset.mcp_service.semantic_layer.tool.list_metrics.DatasetDAO"
+        ) as mock_dao,
+        patch(
+            "superset.mcp_service.semantic_layer.tool.list_metrics.SemanticViewDAO"
+        ) as mock_view_dao,
+    ):
+        mock_dao.find_by_id.return_value = mock_ds
+        mock_view_dao.find_accessible.return_value = []
+
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "list_metrics",
+                {"request": {"dataset_id": 42, "include_compatible_dimensions": False}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    assert data["total_count"] == 2
+    metrics = data["metrics"]
+    assert {m["name"] for m in metrics} == {"count", "revenue"}
+    assert all(m["source"] == "builtin" for m in metrics)
+    assert all(m["dataset_id"] == 42 for m in metrics)
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_mutual_exclusion_validation(mcp_server: FastMCP) -> None:
+    """list_metrics returns a validation error when dataset_id and view_id coexist."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "list_metrics",
+            {"request": {"dataset_id": 1, "view_id": 2}},
+        )
+    data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "ValidationError"
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_privacy_check(mcp_server: FastMCP) -> None:
+    """list_metrics returns an error when the user lacks data-model metadata access."""
+    with patch.object(
+        list_metrics_module,
+        "user_can_view_data_model_metadata",
+        return_value=False,
+    ):
+        async with Client(mcp_server) as client:
+            result = await client.call_tool("list_metrics", {})
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "DataModelMetadataRestricted"
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_search_filter(mcp_server: FastMCP) -> None:
+    """list_metrics filters metrics by search term."""
+    mock_ds: MagicMock = _make_dataset(1)
+
+    with (
+        patch(
+            "superset.mcp_service.semantic_layer.tool.list_metrics.DatasetDAO"
+        ) as mock_dao,
+        patch(
+            "superset.mcp_service.semantic_layer.tool.list_metrics.SemanticViewDAO"
+        ) as mock_view_dao,
+        patch("superset.mcp_service.semantic_layer.tool.list_metrics.db") as mock_db,
+    ):
+        mock_view_dao.find_accessible.return_value = []
+        mock_query: MagicMock = MagicMock()
+        mock_db.session.query.return_value.options.return_value = mock_query
+        mock_dao._apply_base_filter.return_value = mock_query
+        mock_query.all.return_value = [mock_ds]
+
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "list_metrics",
+                {"request": {"search": "revenue"}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    # Only the "revenue" metric should match the search
+    metrics = data["metrics"]
+    assert len(metrics) == 1
+    assert metrics[0]["name"] == "revenue"
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_external_includes_verbose_name(
+    mcp_server: FastMCP,
+) -> None:
+    """External metrics include verbose_name, matching the builtin path."""
+    mock_view = _make_view(5)
+    mock_view.metrics[0].verbose_name = "Bookings Count"
+
+    with patch(
+        "superset.mcp_service.semantic_layer.tool.list_metrics.SemanticViewDAO"
+    ) as mock_view_dao:
+        mock_view_dao.find_by_id.return_value = mock_view
+
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "list_metrics",
+                {"request": {"view_id": 5, "include_compatible_dimensions": False}},
+            )
+        data = json.loads(result.content[0].text)
+
+    metrics = {m["name"]: m for m in data["metrics"]}
+    assert metrics["bookings"]["verbose_name"] == "Bookings Count"
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_external_access_denied(mcp_server: FastMCP) -> None:
+    """An explicit view_id lookup surfaces AccessDenied instead of InternalError."""
+    mock_view = _make_view(5)
+    mock_view.raise_for_access.side_effect = _access_denied_exc()
+
+    with patch(
+        "superset.mcp_service.semantic_layer.tool.list_metrics.SemanticViewDAO"
+    ) as mock_view_dao:
+        mock_view_dao.find_by_id.return_value = mock_view
+
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "list_metrics",
+                {"request": {"view_id": 5}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is False
+    assert data["error_type"] == "AccessDenied"
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_external_per_metric_compatible_dimensions(
+    mcp_server: FastMCP,
+) -> None:
+    """External metrics resolve compatible_dimensions per metric, not view-wide."""
+    mock_view = _make_view(5)
+
+    def _compatible_dimensions(
+        selected_metrics: list[str], selected_dimensions: list[str]
+    ) -> list[str]:
+        return ["listing__country_name"] if selected_metrics == ["bookings"] else []
+
+    mock_view.get_compatible_dimensions.side_effect = _compatible_dimensions
+
+    with patch(
+        "superset.mcp_service.semantic_layer.tool.list_metrics.SemanticViewDAO"
+    ) as mock_view_dao:
+        mock_view_dao.find_by_id.return_value = mock_view
+
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "list_metrics",
+                {"request": {"view_id": 5, "include_compatible_dimensions": True}},
+            )
+        data = json.loads(result.content[0].text)
+
+    assert data["success"] is True
+    metrics = {m["name"]: m for m in data["metrics"]}
+    assert [d["name"] for d in metrics["bookings"]["compatible_dimensions"]] == [
+        "listing__country_name"
+    ]
+    assert metrics["revenue"]["compatible_dimensions"] == []
+    assert mock_view.get_compatible_dimensions.call_args_list == [
+        call(["bookings"], []),
+        call(["revenue"], []),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_list_metrics_pagination_is_stable(mcp_server: FastMCP) -> None:
+    """Metrics are sorted deterministically before pagination is applied."""
+    mock_ds = MagicMock()
+    mock_ds.id = 1
+    mock_ds.table_name = "table_1"
+    mock_ds.metrics = [_make_metric("zzz_metric"), _make_metric("aaa_metric")]
+    mock_ds.columns = []
+
+    with (
+        patch(
+            "superset.mcp_service.semantic_layer.tool.list_metrics.DatasetDAO"
+        ) as mock_dao,
+        patch(
+            "superset.mcp_service.semantic_layer.tool.list_metrics.SemanticViewDAO"
+        ) as mock_view_dao,
+        patch("superset.mcp_service.semantic_layer.tool.list_metrics.db") as mock_db,
+    ):
+        mock_view_dao.find_accessible.return_value = []
+        mock_query = MagicMock()
+        mock_db.session.query.return_value.options.return_value = mock_query
+        mock_dao._apply_base_filter.return_value = mock_query
+        mock_query.all.return_value = [mock_ds]
+
+        async with Client(mcp_server) as client:
+            page_1 = await client.call_tool(
+                "list_metrics", {"request": {"page": 1, "page_size": 1}}
+            )
+            page_2 = await client.call_tool(
+                "list_metrics", {"request": {"page": 2, "page_size": 1}}
+            )
+        data_1 = json.loads(page_1.content[0].text)
+        data_2 = json.loads(page_2.content[0].text)
+
+    assert data_1["metrics"][0]["name"] == "aaa_metric"
+    assert data_2["metrics"][0]["name"] == "zzz_metric"