fix(mcp): address code review findings for RBAC tool visibility

- Fail closed (return only public tools) when credentials are invalid
  (PermissionError from bad API key, ValueError from unknown dev username);
  fail open only when no auth source is configured at all
- Extract _get_app_context_manager() to module level in auth.py so
  RBACToolVisibilityMiddleware reuses the same context-selection logic as
  mcp_auth_hook, preventing external g.user from being shadowed
- Add RBACToolVisibilityMiddleware to __main__.py stdio entry point via
  build_middleware_list() to keep all transports in sync
- Fix stale patch targets in test_tool_search_transform.py: update
  superset.mcp_service.server.user_can_view_data_model_metadata →
  superset.mcp_service.privacy.user_can_view_data_model_metadata
- Qualify write tool listings in instructions with "(requires write access)"
  and add a permissions preamble so read-only users are not confused by
  tools they cannot call

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

superset/mcp_service/__main__.py

@@ -47,32 +47,27 @@ from superset.mcp_service.app import init_fastmcp_server, mcp
 def _add_default_middlewares() -> None:
     """Add the standard middleware stack to the MCP instance.
 
-    This ensures all entry points (stdio, streamable-http, etc.) get
-    the same protection middlewares that the Flask CLI and server.py add.
-    Order is innermost → outermost (last-added wraps everything).
-    """
-    from superset.mcp_service.middleware import (
-        create_response_size_guard_middleware,
-        GlobalErrorHandlerMiddleware,
-        LoggingMiddleware,
-        StructuredContentStripperMiddleware,
-    )
+    Delegates to ``server.build_middleware_list()`` for the core stack so
+    the stdio entry point stays in sync with the HTTP server without
+    duplicating middleware ordering.  The optional response size guard is
+    appended separately (innermost position, same as in run_server()).
 
-    # Response size guard (innermost among these)
+    FastMCP wraps handlers so that the FIRST-added middleware is outermost.
+    ``build_middleware_list()`` already returns middlewares in the correct
+    outermost-first order.
+    """
+    from superset.mcp_service.middleware import create_response_size_guard_middleware
+    from superset.mcp_service.server import build_middleware_list
+
+    for middleware in build_middleware_list():
+        mcp.add_middleware(middleware)
+
+    # Response size guard is innermost (added last)
     if size_guard := create_response_size_guard_middleware():
         mcp.add_middleware(size_guard)
         limit = size_guard.token_limit
         sys.stderr.write(f"[MCP] Response size guard enabled (token_limit={limit})\n")
 
-    # Logging
-    mcp.add_middleware(LoggingMiddleware())
-
-    # Global error handler
-    mcp.add_middleware(GlobalErrorHandlerMiddleware())
-
-    # Structured content stripper (must be outermost)
-    mcp.add_middleware(StructuredContentStripperMiddleware())
-
 
 def main() -> None:
     """

superset/mcp_service/app.py

@@ -46,13 +46,19 @@ You are connected to the {branding} MCP (Model Context Protocol) service.
 This service provides programmatic access to {branding} dashboards, charts, datasets,
 SQL Lab, and instance metadata via a comprehensive set of tools.
 
-Available tools:
+IMPORTANT - Permission-based tool availability:
+Available tools vary based on your access level. Write operations (generating charts,
+dashboards, or datasets; saving SQL queries; executing SQL) require write permissions.
+If a write tool does not appear in the tool list, the current user lacks the necessary
+access — do NOT attempt to call it. Read-only users will only see read tools.
+
+Tool capabilities (subject to your access level):
 
 Dashboard Management:
 - list_dashboards: List dashboards with advanced filters (1-based pagination)
 - get_dashboard_info: Get detailed dashboard information by ID
-- generate_dashboard: Create a dashboard from chart IDs
-- add_chart_to_existing_dashboard: Add a chart to an existing dashboard
+- generate_dashboard: Create a dashboard from chart IDs (requires write access)
+- add_chart_to_existing_dashboard: Add a chart to an existing dashboard (requires write access)
 
 Database Connections:
 - list_databases: List database connections with advanced filters (1-based pagination)
@@ -61,7 +67,7 @@ Database Connections:
 Dataset Management:
 - list_datasets: List datasets with advanced filters (1-based pagination)
 - get_dataset_info: Get detailed dataset information by ID (includes columns/metrics)
-- create_virtual_dataset: Save a SQL query as a virtual dataset for charting
+- create_virtual_dataset: Save a SQL query as a virtual dataset for charting (requires write access)
 - query_dataset: Query a dataset using its semantic layer (saved metrics, dimensions, filters) without needing a saved chart
 
 Chart Management:
@@ -70,14 +76,14 @@ Chart Management:
 - get_chart_preview: Get a visual preview of a chart as formatted content or URL
 - get_chart_data: Get underlying chart data in text-friendly format
 - get_chart_sql: Get the rendered SQL query for a chart (without executing it)
-- generate_chart: Create and save a new chart permanently
+- generate_chart: Create and save a new chart permanently (requires write access)
 - generate_explore_link: Create an interactive explore URL (preferred for exploration)
-- update_chart: Update existing saved chart configuration
-- update_chart_preview: Update cached chart preview without saving
+- update_chart: Update existing saved chart configuration (requires write access)
+- update_chart_preview: Update cached chart preview without saving (requires write access)
 
 SQL Lab Integration:
-- execute_sql: Execute SQL queries and get results (requires database_id)
-- save_sql_query: Save a SQL query to Saved Queries list
+- execute_sql: Execute SQL queries and get results (requires database_id and SQL access)
+- save_sql_query: Save a SQL query to Saved Queries list (requires write access)
 - open_sql_lab_with_context: Generate SQL Lab URL with pre-filled sql
 
 Schema Discovery:
@@ -295,6 +301,10 @@ Permission Awareness:
 - get_instance_info returns current_user.roles (e.g., ["Admin"], ["Alpha"], ["Viewer"]).
 - ALWAYS check the user's roles BEFORE suggesting write operations (creating datasets,
   charts, dashboards, or running SQL).
+- Write tools (generate_chart, generate_dashboard, update_chart, create_virtual_dataset,
+  save_sql_query, add_chart_to_existing_dashboard, update_chart_preview) require write
+  permissions. These tools are only listed for users who have the necessary access.
+  If a write tool does not appear in the tool list, the current user lacks write access.
 - Do NOT disclose dashboard access lists, dashboard owners, chart owners, dataset
   owners, workspace admins, or other users' names, usernames, email addresses,
   contact details, roles, admin status, ownership, or access-list information.

superset/mcp_service/auth.py

@@ -145,6 +145,56 @@ def check_tool_permission(func: Callable[..., Any]) -> bool:
         return False
 
 
+def is_tool_visible_to_current_user(tool: Any) -> bool:
+    """Return whether the current user can see a tool in tools/list.
+
+    Checks both RBAC permissions and data-model metadata privacy. The caller
+    must set ``g.user`` before calling this function.
+
+    This is the single source of truth for tool visibility — called from both
+    ``RBACToolVisibilityMiddleware`` (``tools/list``) and
+    ``_tool_allowed_for_current_user()`` (tool search).
+
+    Args:
+        tool: A FastMCP Tool object.
+
+    Returns:
+        True if the tool is visible to the current user, False otherwise.
+    """
+    try:
+        from flask import current_app
+
+        if not current_app.config.get("MCP_RBAC_ENABLED", True):
+            return True
+
+        tool_func = getattr(tool, "fn", None)
+        if tool_func is None:
+            return True
+
+        from superset.mcp_service.privacy import (
+            tool_requires_data_model_metadata_access,
+            user_can_view_data_model_metadata,
+        )
+
+        if (
+            tool_requires_data_model_metadata_access(tool_func)
+            and not user_can_view_data_model_metadata()
+        ):
+            return False
+
+        class_permission_name = getattr(tool_func, CLASS_PERMISSION_ATTR, None)
+        if not class_permission_name:
+            return True
+
+        return check_tool_permission(tool_func)
+
+    except (AttributeError, RuntimeError, ValueError):
+        logger.debug(
+            "Could not evaluate tool visibility for current user", exc_info=True
+        )
+        return False
+
+
 def load_user_with_relationships(
     username: str | None = None, email: str | None = None
 ) -> User | None:
@@ -516,6 +566,39 @@ def _cleanup_session_on_error() -> None:
         logger.warning("Error cleaning up session after exception: %s", e)
 
 
+def _get_app_context_manager() -> AbstractContextManager[None]:
+    """Return the right context manager for the current Flask state.
+
+    When a request context is present, external middleware (e.g.
+    Preset's WorkspaceContextMiddleware) has already set ``g.user``
+    on a per-request app context — reuse it via ``nullcontext()``.
+
+    When only a bare app context exists (no request context), push a
+    **new** app context so concurrent tool calls do not share one ``g``
+    namespace (which would cause ``g.user`` races under asyncio).
+
+    When no context exists at all, push a fresh app context from the
+    Flask singleton.
+
+    This is the single source of truth for context selection — called
+    from both ``mcp_auth_hook`` (tool execution) and
+    ``RBACToolVisibilityMiddleware`` (tools/list filtering).
+    """
+    import contextlib
+
+    from flask import current_app, has_app_context, has_request_context
+
+    if has_request_context():
+        return contextlib.nullcontext()
+    if has_app_context():
+        # Push a new context for the CURRENT app (not get_flask_app()
+        # which may return a different instance in test environments).
+        return current_app._get_current_object().app_context()
+    from superset.mcp_service.flask_singleton import get_flask_app
+
+    return get_flask_app().app_context()
+
+
 def mcp_auth_hook(tool_func: F) -> F:  # noqa: C901
     """
     Authentication and authorization decorator for MCP tools.
@@ -530,42 +613,10 @@ def mcp_auth_hook(tool_func: F) -> F:  # noqa: C901
 
     Supports both sync and async tool functions.
     """
-    import contextlib
     import functools
     import inspect
     import types
 
-    from flask import current_app, has_app_context, has_request_context
-
-    def _get_app_context_manager() -> AbstractContextManager[None]:
-        """Push a fresh app context unless a request context is active.
-
-        When a request context is present, external middleware (e.g.
-        Preset's WorkspaceContextMiddleware) has already set ``g.user``
-        on a per-request app context — reuse it via ``nullcontext()``.
-
-        When only a bare app context exists (no request context), we must
-        push a **new** app context. The MCP server typically runs inside
-        a long-lived app context (e.g. ``__main__.py`` wraps
-        ``mcp.run()`` in ``app.app_context()``). When FastMCP dispatches
-        concurrent tool calls via ``asyncio.create_task()``, each task
-        inherits the parent's ``ContextVar`` *value* — a reference to the
-        **same** ``AppContext`` object. Without a fresh push, all tasks
-        share one ``g`` namespace and concurrent ``g.user`` mutations
-        race: one user's identity can overwrite another's before
-        ``get_user_id()`` runs during the SQLAlchemy INSERT flush,
-        attributing the created asset to the wrong user.
-        """
-        if has_request_context():
-            return contextlib.nullcontext()
-        if has_app_context():
-            # Push a new context for the CURRENT app (not get_flask_app()
-            # which may return a different instance in test environments).
-            return current_app._get_current_object().app_context()
-        from superset.mcp_service.flask_singleton import get_flask_app
-
-        return get_flask_app().app_context()
-
     is_async = inspect.iscoroutinefunction(tool_func)
 
     # Detect if the original function expects a ctx: Context parameter.

superset/mcp_service/middleware.py

@@ -37,6 +37,7 @@ from superset.commands.exceptions import (
 )
 from superset.exceptions import SupersetException, SupersetSecurityException
 from superset.extensions import event_logger
+from superset.mcp_service.auth import MCPPermissionDeniedError
 from superset.mcp_service.constants import (
     DEFAULT_TOKEN_LIMIT,
     DEFAULT_WARN_THRESHOLD_PCT,
@@ -123,6 +124,7 @@ _USER_ERROR_TYPES = (
     ToolError,
     ValidationError,
     PermissionError,
+    MCPPermissionDeniedError,
     ValueError,
     FileNotFoundError,
     CommandInvalidError,
@@ -393,6 +395,79 @@ class StructuredContentStripperMiddleware(Middleware):
         return result
 
 
+class RBACToolVisibilityMiddleware(Middleware):
+    """Filter tools/list response based on current user's RBAC permissions.
+
+    Intercepts every ``tools/list`` request and removes tools the calling user
+    is not permitted to execute. Public tools (no ``class_permission_name``) and
+    tools whose permission check passes are included; all others are hidden.
+
+    Fail-open vs fail-closed behaviour:
+    - No auth context at all (no Flask context, no auth header, no dev user
+      configured) → fail open (return all tools). Call-time RBAC enforces.
+    - Auth was attempted but credentials are invalid (bad API key, dev
+      username not in DB, etc.) → fail closed (return only public tools,
+      i.e. those with no ``class_permission_name``).
+    - Unexpected errors → fail open. Call-time RBAC still enforces.
+    """
+
+    @staticmethod
+    def _public_tools_only(tools: list[Tool]) -> list[Tool]:
+        """Return only tools that require no class-level permission."""
+        return [
+            t
+            for t in tools
+            if getattr(getattr(t, "fn", None), "_class_permission_name", None) is None
+        ]
+
+    async def on_list_tools(
+        self,
+        context: MiddlewareContext[mt.ListToolsRequest],
+        call_next: CallNext[mt.ListToolsRequest, list[Tool]],
+    ) -> list[Tool]:
+        tools = await call_next(context)
+
+        try:
+            from flask import g
+
+            from superset.mcp_service.auth import (
+                _get_app_context_manager,
+                _setup_user_context,
+                is_tool_visible_to_current_user,
+            )
+
+            with _get_app_context_manager():
+                try:
+                    user = _setup_user_context()
+                except ValueError as exc:
+                    if "No authenticated user found" in str(exc):
+                        # No auth source configured at all → fail open
+                        return tools
+                    # Auth was attempted (e.g. MCP_DEV_USERNAME set) but the
+                    # user was not found in the DB → fail closed
+                    logger.warning(
+                        "MCP tool list: credential failure, hiding protected tools: %s",
+                        exc,
+                    )
+                    return self._public_tools_only(tools)
+                except PermissionError as exc:
+                    # API key present but invalid/expired → fail closed
+                    logger.warning(
+                        "MCP tool list: credential failure, hiding protected tools: %s",
+                        exc,
+                    )
+                    return self._public_tools_only(tools)
+
+                if user is None:
+                    return tools  # no Flask app context → fail open
+                g.user = user
+                return [t for t in tools if is_tool_visible_to_current_user(t)]
+        except Exception:  # noqa: BLE001
+            # Unexpected setup errors (ImportError, etc.) → fail open.
+            # Call-time RBAC still enforces permissions.
+            return tools
+
+
 class GlobalErrorHandlerMiddleware(Middleware):
     """
     Global error handler middleware that provides consistent error responses
@@ -501,6 +576,9 @@ class GlobalErrorHandlerMiddleware(Middleware):
             raise ToolError(
                 f"Invalid request for {tool_name}: {_sanitize_error_for_logging(error)}"
             ) from error
+        elif isinstance(error, MCPPermissionDeniedError):
+            # MCP RBAC permission denied — convert to structured ToolError
+            raise ToolError(str(error)) from error
         elif isinstance(error, (ForbiddenError, SupersetSecurityException)):
             # Superset access denied — agent tried a tool it can't use
             raise ToolError(

superset/mcp_service/server.py

@@ -41,12 +41,9 @@ from superset.mcp_service.middleware import (
     create_response_size_guard_middleware,
     GlobalErrorHandlerMiddleware,
     LoggingMiddleware,
+    RBACToolVisibilityMiddleware,
     StructuredContentStripperMiddleware,
 )
-from superset.mcp_service.privacy import (
-    tool_requires_data_model_metadata_access,
-    user_can_view_data_model_metadata,
-)
 from superset.mcp_service.storage import _create_redis_store
 from superset.utils import json
 
@@ -403,38 +400,20 @@ def _build_summary_serializer(max_desc: int) -> Any:
 def _tool_allowed_for_current_user(tool: Any) -> bool:
     """Return whether the current Flask user can see this tool in search results."""
     try:
-        from flask import current_app, g
+        from flask import g
 
-        if not current_app.config.get("MCP_RBAC_ENABLED", True):
-            return True
-
-        from superset import security_manager
         from superset.mcp_service.auth import (
-            CLASS_PERMISSION_ATTR,
             get_user_from_request,
-            METHOD_PERMISSION_ATTR,
-            PERMISSION_PREFIX,
+            is_tool_visible_to_current_user,
         )
 
-        tool_func = getattr(tool, "fn", None)
-        if tool_requires_data_model_metadata_access(tool_func) and not (
-            user_can_view_data_model_metadata()
-        ):
-            return False
-
-        class_permission_name = getattr(tool_func, CLASS_PERMISSION_ATTR, None)
-        if not class_permission_name:
-            return True
-
         if not getattr(g, "user", None):
             try:
                 g.user = get_user_from_request()
             except ValueError:
                 return False
 
-        method_permission_name = getattr(tool_func, METHOD_PERMISSION_ATTR, "read")
-        permission_name = f"{PERMISSION_PREFIX}{method_permission_name}"
-        return security_manager.can_access(permission_name, class_permission_name)
+        return is_tool_visible_to_current_user(tool)
     except (AttributeError, RuntimeError, ValueError):
         logger.debug("Could not evaluate tool search permission", exc_info=True)
         return False
@@ -711,11 +690,15 @@ def build_middleware_list() -> list[Middleware]:
 
     1. StructuredContentStripper — safety net, converts exceptions
        to safe ToolResult text for transports that can't encode errors
-    2. LoggingMiddleware — logs tool calls with success/failure status
-    3. GlobalErrorHandler — catches tool exceptions, raises ToolError
+    2. RBACToolVisibilityMiddleware — filters tools/list by RBAC;
+       positioned inside the Stripper so it sees full tool objects
+       (with outputSchema) before stripping occurs
+    3. LoggingMiddleware — logs tool calls with success/failure status
+    4. GlobalErrorHandler — catches tool exceptions, raises ToolError
     """
     return [
         StructuredContentStripperMiddleware(),
+        RBACToolVisibilityMiddleware(),
         LoggingMiddleware(),
         GlobalErrorHandlerMiddleware(),
     ]

tests/unit_tests/mcp_service/test_auth_rbac.py

@@ -25,6 +25,7 @@ from flask import g
 from superset.mcp_service.auth import (
     check_tool_permission,
     CLASS_PERMISSION_ATTR,
+    is_tool_visible_to_current_user,
     MCPPermissionDeniedError,
     METHOD_PERMISSION_ATTR,
     PERMISSION_PREFIX,
@@ -223,3 +224,122 @@ def app_context(app):
     """Provide Flask app context for tests needing g.user."""
     with app.app_context():
         yield
+
+
+# -- is_tool_visible_to_current_user --
+
+
+def _make_mock_tool(
+    class_perm: str | None = None,
+    method_perm: str | None = None,
+    fn: object | None = None,
+) -> MagicMock:
+    """Create a mock FastMCP Tool object for visibility tests."""
+    tool = MagicMock()
+    if fn is not None:
+        tool.fn = fn
+    elif class_perm is not None:
+        func = _make_tool_func(class_perm, method_perm)
+        tool.fn = func
+    else:
+        tool.fn = None
+    return tool
+
+
+def test_visibility_returns_true_when_rbac_disabled(app_context, app) -> None:
+    """is_tool_visible_to_current_user returns True when RBAC is disabled."""
+    app.config["MCP_RBAC_ENABLED"] = False
+    tool = _make_mock_tool(class_perm="Chart", method_perm="write")
+    try:
+        assert is_tool_visible_to_current_user(tool) is True
+    finally:
+        app.config["MCP_RBAC_ENABLED"] = True
+
+
+def test_visibility_returns_true_when_fn_is_none(app_context) -> None:
+    """Tools with fn=None (public/synthetic) are always visible."""
+    tool = _make_mock_tool()
+    assert is_tool_visible_to_current_user(tool) is True
+
+
+def test_visibility_public_tool_no_class_permission(app_context) -> None:
+    """Tools without class_permission_name are visible to all users."""
+    g.user = MagicMock(username="viewer")
+    func = _make_tool_func()  # no class permission
+    tool = MagicMock()
+    tool.fn = func
+    assert is_tool_visible_to_current_user(tool) is True
+
+
+def test_visibility_allowed_tool(app_context) -> None:
+    """Tools where security_manager grants access are visible."""
+    g.user = MagicMock(username="admin")
+    func = _make_tool_func(class_perm="Chart", method_perm="read")
+    tool = MagicMock()
+    tool.fn = func
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with patch("superset.security_manager", mock_sm):
+        result = is_tool_visible_to_current_user(tool)
+
+    assert result is True
+
+
+def test_visibility_denied_tool(app_context) -> None:
+    """Tools where security_manager denies access are hidden."""
+    g.user = MagicMock(username="viewer")
+    func = _make_tool_func(class_perm="Dashboard", method_perm="write")
+    tool = MagicMock()
+    tool.fn = func
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=False)
+    with patch("superset.security_manager", mock_sm):
+        result = is_tool_visible_to_current_user(tool)
+
+    assert result is False
+
+
+def test_visibility_data_model_metadata_denied(app_context) -> None:
+    """Tools requiring data-model metadata access are hidden when user lacks it."""
+    g.user = MagicMock(username="viewer")
+    func = _make_tool_func(class_perm="Dataset", method_perm="read")
+    func._requires_data_model_metadata_access = True  # type: ignore[attr-defined]
+    tool = MagicMock()
+    tool.fn = func
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with (
+        patch("superset.security_manager", mock_sm),
+        patch(
+            "superset.mcp_service.privacy.user_can_view_data_model_metadata",
+            return_value=False,
+        ),
+    ):
+        result = is_tool_visible_to_current_user(tool)
+
+    assert result is False
+
+
+def test_visibility_data_model_metadata_allowed(app_context) -> None:
+    """Tools requiring data-model metadata access are visible when user has it."""
+    g.user = MagicMock(username="alpha")
+    func = _make_tool_func(class_perm="Dataset", method_perm="read")
+    func._requires_data_model_metadata_access = True  # type: ignore[attr-defined]
+    tool = MagicMock()
+    tool.fn = func
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with (
+        patch("superset.security_manager", mock_sm),
+        patch(
+            "superset.mcp_service.privacy.user_can_view_data_model_metadata",
+            return_value=True,
+        ),
+    ):
+        result = is_tool_visible_to_current_user(tool)
+
+    assert result is True

tests/unit_tests/mcp_service/test_middleware.py

@@ -1021,12 +1021,162 @@ class TestGlobalErrorHandlerLogLevels:
         error.status = 500
         call_next = AsyncMock(side_effect=error)
 
+        mock_logger = MagicMock()
         with (
             patch("superset.mcp_service.middleware.get_user_id", return_value=1),
             patch("superset.mcp_service.middleware.event_logger"),
-            patch("superset.mcp_service.middleware.logger") as mock_logger,
+            patch("superset.mcp_service.middleware.logger", mock_logger),
             pytest.raises(ToolError, match="Internal error"),
         ):
             await middleware.on_message(context, call_next)
 
         mock_logger.error.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_mcp_permission_denied_error_becomes_tool_error(self) -> None:
+        """MCPPermissionDeniedError must convert to ToolError, not a generic error."""
+        from superset.mcp_service.auth import MCPPermissionDeniedError
+
+        middleware = GlobalErrorHandlerMiddleware()
+
+        context = MagicMock()
+        context.message.name = "generate_dashboard"
+        context.method = "tools/call"
+
+        error = MCPPermissionDeniedError(
+            permission_name="can_write",
+            view_name="Dashboard",
+            user="viewer",
+            tool_name="generate_dashboard",
+        )
+        call_next = AsyncMock(side_effect=error)
+
+        with (
+            patch("superset.mcp_service.middleware.get_user_id", return_value=42),
+            patch("superset.mcp_service.middleware.event_logger"),
+            pytest.raises(ToolError) as exc_info,
+        ):
+            await middleware.on_message(context, call_next)
+
+        assert "can_write" in str(exc_info.value)
+        assert "Dashboard" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_mcp_permission_denied_error_is_user_error(self) -> None:
+        """MCPPermissionDeniedError must be classified as a user error (WARNING)."""
+        from superset.mcp_service.auth import MCPPermissionDeniedError
+
+        error = MCPPermissionDeniedError(
+            permission_name="can_write",
+            view_name="Chart",
+        )
+        assert _is_user_error(error) is True
+
+    @pytest.mark.asyncio
+    async def test_mcp_permission_denied_error_logs_at_warning(self) -> None:
+        """MCPPermissionDeniedError should log at WARNING, not ERROR."""
+        from superset.mcp_service.auth import MCPPermissionDeniedError
+
+        middleware = GlobalErrorHandlerMiddleware()
+
+        context = MagicMock()
+        context.message.name = "generate_chart"
+        context.method = "tools/call"
+
+        error = MCPPermissionDeniedError(
+            permission_name="can_write",
+            view_name="Chart",
+            user="reader",
+        )
+        call_next = AsyncMock(side_effect=error)
+
+        mock_logger = MagicMock()
+        with (
+            patch("superset.mcp_service.middleware.get_user_id", return_value=5),
+            patch("superset.mcp_service.middleware.event_logger"),
+            patch("superset.mcp_service.middleware.logger", mock_logger),
+            pytest.raises(ToolError),
+        ):
+            await middleware.on_message(context, call_next)
+
+        mock_logger.warning.assert_called()
+        mock_logger.error.assert_not_called()
+
+
+class TestRBACToolVisibilityMiddleware:
+    """Tests for RBACToolVisibilityMiddleware.on_list_tools."""
+
+    def _make_tool(self, name: str = "test_tool") -> Any:
+        """Create a minimal mock tool object."""
+        tool = MagicMock()
+        tool.name = name
+        return tool
+
+    @pytest.mark.asyncio
+    async def test_fails_open_on_exception(self) -> None:
+        """Returns all tools when get_flask_app raises (fail open)."""
+        from superset.mcp_service.middleware import RBACToolVisibilityMiddleware
+
+        tools = [self._make_tool("list_charts"), self._make_tool("generate_chart")]
+        call_next = AsyncMock(return_value=tools)
+        middleware = RBACToolVisibilityMiddleware()
+
+        with patch(
+            "superset.mcp_service.flask_singleton.get_flask_app",
+            side_effect=RuntimeError("no app"),
+        ):
+            result = await middleware.on_list_tools(MagicMock(), call_next)
+
+        assert result == tools
+
+    @pytest.mark.asyncio
+    async def test_fails_open_when_user_is_none(self, app) -> None:
+        """Returns all tools when _setup_user_context returns None."""
+        from superset.mcp_service.middleware import RBACToolVisibilityMiddleware
+
+        tools = [self._make_tool("list_charts"), self._make_tool("generate_chart")]
+        call_next = AsyncMock(return_value=tools)
+        middleware = RBACToolVisibilityMiddleware()
+
+        with (
+            patch(
+                "superset.mcp_service.flask_singleton.get_flask_app", return_value=app
+            ),
+            patch("superset.mcp_service.auth._setup_user_context", return_value=None),
+        ):
+            result = await middleware.on_list_tools(MagicMock(), call_next)
+
+        assert result == tools
+
+    @pytest.mark.asyncio
+    async def test_filters_tools_by_rbac(self, app) -> None:
+        """Tools denied by is_tool_visible_to_current_user are removed."""
+        from superset.mcp_service.middleware import RBACToolVisibilityMiddleware
+
+        read_tool = self._make_tool("list_charts")
+        write_tool = self._make_tool("generate_chart")
+        tools = [read_tool, write_tool]
+        call_next = AsyncMock(return_value=tools)
+        middleware = RBACToolVisibilityMiddleware()
+
+        mock_user = MagicMock()
+
+        def _visible(tool: Any) -> bool:
+            return tool.name == "list_charts"
+
+        with (
+            patch(
+                "superset.mcp_service.flask_singleton.get_flask_app", return_value=app
+            ),
+            patch(
+                "superset.mcp_service.auth._setup_user_context", return_value=mock_user
+            ),
+            patch(
+                "superset.mcp_service.auth.is_tool_visible_to_current_user",
+                side_effect=_visible,
+            ),
+        ):
+            result = await middleware.on_list_tools(MagicMock(), call_next)
+
+        assert read_tool in result
+        assert write_tool not in result

tests/unit_tests/mcp_service/test_tool_search_transform.py

@@ -916,7 +916,7 @@ def test_tool_search_filter_hides_metadata_tools_without_access() -> None:
     with app.app_context():
         g.user = SimpleNamespace(username="viewer")
         with patch(
-            "superset.mcp_service.server.user_can_view_data_model_metadata",
+            "superset.mcp_service.privacy.user_can_view_data_model_metadata",
             return_value=False,
         ):
             result = _filter_tools_by_current_user_permission([metadata, public])
@@ -943,7 +943,7 @@ def test_tool_search_permission_filter_still_applies_rbac_to_metadata_tools() ->
         g.user = SimpleNamespace(username="viewer")
         with (
             patch(
-                "superset.mcp_service.server.user_can_view_data_model_metadata",
+                "superset.mcp_service.privacy.user_can_view_data_model_metadata",
                 return_value=True,
             ),
             patch("superset.security_manager", new_callable=Mock) as security_manager,
@@ -996,7 +996,7 @@ def test_tool_search_permission_filter_keeps_get_schema_visible_without_metadata
         g.user = SimpleNamespace(username="viewer")
         with (
             patch(
-                "superset.mcp_service.server.user_can_view_data_model_metadata",
+                "superset.mcp_service.privacy.user_can_view_data_model_metadata",
                 return_value=False,
             ),
             patch("superset.security_manager", new_callable=Mock) as security_manager,