feat(mcp): add create_css_template and update_css_template tools

- Add CreateCssTemplateCommand, UpdateCssTemplateCommand with exceptions
- Add create_css_template and update_css_template MCP tools with Pydantic
  schemas, event logging, and structured error responses
- Add sanitize_error_for_llm_context validator on error fields (CWE-79)
- Add unit tests for both tools including schema validation and exception
  re-raise coverage

.github/workflows/superset-docs-deploy.yml

@@ -45,7 +45,13 @@ jobs:
           SUPERSET_SITE_BUILD: ${{ (secrets.SUPERSET_SITE_BUILD != '' && secrets.SUPERSET_SITE_BUILD != '') || '' }}
   build-deploy:
     needs: config
-    if: needs.config.outputs.has-secrets
+    # For workflow_run triggers, only deploy when the triggering run originated
+    # from this repository (not a fork), ensuring the checked-out code and any
+    # local actions executed with deploy credentials are trusted.
+    if: >-
+      needs.config.outputs.has-secrets &&
+      (github.event_name != 'workflow_run' ||
+       github.event.workflow_run.head_repository.full_name == github.repository)
     name: Build & Deploy
     runs-on: ubuntu-24.04
     steps:

.github/workflows/superset-docs-verify.yml

@@ -97,7 +97,8 @@ jobs:
     # Only runs if integration tests succeeded
     if: >
       github.event_name == 'workflow_run' &&
-      github.event.workflow_run.conclusion == 'success'
+      github.event.workflow_run.conclusion == 'success' &&
+      github.event.workflow_run.head_repository.full_name == github.repository
     name: Build (after integration tests)
     runs-on: ubuntu-24.04
     defaults:

.pre-commit-config.yaml

@@ -158,3 +158,14 @@ repos:
         language: system
         files: ^superset/config\.py$
         pass_filenames: false
+      - id: zizmor
+        name: zizmor (GHA security audit)
+        entry: zizmor
+        language: python
+        additional_dependencies: [zizmor==1.25.2]
+        files: ^\.github/
+        types: [yaml]
+        pass_filenames: false
+        # Advisory until pre-existing findings are resolved; remove
+        # --no-exit-codes to make this hook blocking.
+        args: [--no-exit-codes, .github/]

Dockerfile

@@ -113,7 +113,7 @@ RUN useradd --user-group -d ${SUPERSET_HOME} -m --no-log-init --shell /bin/bash
 # Some bash scripts needed throughout the layers
 COPY --chmod=755 docker/*.sh /app/docker/
 
-RUN pip install --no-cache-dir --upgrade uv
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
 
 # Using uv as it's faster/simpler than pip
 RUN uv venv /app/.venv

docs/package.json

@@ -128,7 +128,10 @@
     "react-redux": "^9.2.0",
     "@reduxjs/toolkit": "^2.5.0",
     "baseline-browser-mapping": "^2.9.19",
-    "swagger-client": "3.37.3"
+    "swagger-client": "3.37.3",
+    "lodash": "4.18.1",
+    "lodash-es": "4.18.1",
+    "yaml": "1.10.3"
   },
   "packageManager": "yarn@1.22.22+sha1.ac34549e6aa8e7ead463a7407e1c7390f61a6610"
 }

docs/yarn.lock

@@ -9676,10 +9676,10 @@ locate-path@^7.1.0:
   dependencies:
     p-locate "^6.0.0"
 
-lodash-es@^4.17.21:
-  version "4.17.21"
-  resolved "https://registry.npmjs.org/lodash-es/-/lodash-es-4.17.21.tgz"
-  integrity sha512-mKnC+QJ9pWVzv+C4/U3rRsHapFfHvQFoFB92e52xeyGMcX6/OlIl78je1u8vePzYZSkkogMPJ2yjxxsb89cxyw==
+lodash-es@4.18.1, lodash-es@^4.17.21:
+  version "4.18.1"
+  resolved "https://registry.yarnpkg.com/lodash-es/-/lodash-es-4.18.1.tgz#b962eeb80d9d983a900bf342961fb7418ca10b1d"
+  integrity sha512-J8xewKD/Gk22OZbhpOVSwcs60zhd95ESDwezOFuA3/099925PdHJ7OFHNTGtajL3AlZkykD32HykiMo+BIBI8A==
 
 lodash.debounce@^4, lodash.debounce@^4.0.8:
   version "4.0.8"
@@ -9701,12 +9701,7 @@ lodash.uniq@^4.5.0:
   resolved "https://registry.npmjs.org/lodash.uniq/-/lodash.uniq-4.5.0.tgz"
   integrity sha512-xfBaXQd9ryd9dlSDvnvI0lvxfLJlYAZzXomUYzLKtUeOQvOP5piqAWuGtrhWeqaXK9hhoM/iyJc5AV+XfsX3HQ==
 
-lodash@4.17.21:
-  version "4.17.21"
-  resolved "https://registry.yarnpkg.com/lodash/-/lodash-4.17.21.tgz#679591c564c3bffaae8454cf0b3df370c3d6911c"
-  integrity sha512-v2kDEe57lecTulaDIuNTPy3Ry4gLGJ6Z1O3vE1krgXZNrsQ+LFTGHVxVjcXPs17LhbZVGedAJv8XZ1tvj5FvSg==
-
-lodash@^4.15.0, lodash@^4.17.10, lodash@^4.17.15, lodash@^4.17.20, lodash@^4.17.21, lodash@^4.17.4, lodash@^4.18.1:
+lodash@4.17.21, lodash@4.18.1, lodash@^4.15.0, lodash@^4.17.10, lodash@^4.17.15, lodash@^4.17.20, lodash@^4.17.21, lodash@^4.17.4, lodash@^4.18.1:
   version "4.18.1"
   resolved "https://registry.yarnpkg.com/lodash/-/lodash-4.18.1.tgz#ff2b66c1f6326d59513de2407bf881439812771c"
   integrity sha512-dMInicTPVE8d1e5otfwmmjlxkZoUpiVLwyeTdUsi/Caj/gfzzblBcCE5sRHV/AsjuCmxWrte2TNGSYuCeCq+0Q==
@@ -15209,12 +15204,7 @@ yaml-ast-parser@0.0.43:
   resolved "https://registry.npmjs.org/yaml-ast-parser/-/yaml-ast-parser-0.0.43.tgz"
   integrity sha512-2PTINUwsRqSd+s8XxKaJWQlUuEMHJQyEuh2edBbW8KNJz0SJPwUSD2zRWqezFEdN7IzAgeuYHFUCF7o8zRdZ0A==
 
-yaml@1.10.2:
-  version "1.10.2"
-  resolved "https://registry.yarnpkg.com/yaml/-/yaml-1.10.2.tgz#2301c5ffbf12b467de8da2333a459e29e7920e4b"
-  integrity sha512-r3vXyErRCYJ7wg28yvBY5VSoAF8ZvlcW9/BwUzEtUsjvX/DKs24dIkuwjtuprwJJHsbyUbLApepYTR1BN4uHrg==
-
-yaml@^1.10.0:
+yaml@1.10.2, yaml@1.10.3, yaml@^1.10.0:
   version "1.10.3"
   resolved "https://registry.yarnpkg.com/yaml/-/yaml-1.10.3.tgz#76e407ed95c42684fb8e14641e5de62fe65bbcb3"
   integrity sha512-vIYeF1u3CjlhAFekPPAk2h/Kv4T3mAkMox5OymRiJQB0spDP10LHvt+K7G9Ny6NuuMAb25/6n1qyUjAcGNf/AA==

scripts/change_detector.py

@@ -33,6 +33,7 @@ PATTERNS = {
         r"^setup\.py",
         r"^pyproject\.toml$",
         r"^requirements/.+\.txt",
+        r"^pyproject\.toml",
         r"^.pylintrc",
     ],
     "frontend": [
@@ -156,7 +157,7 @@ def main(event_type: str, sha: str, repo: str) -> None:
 
 def get_git_sha() -> str:
     return os.getenv("GITHUB_SHA") or subprocess.check_output(  # noqa: S603
-        ["git", "rev-parse", "HEAD"]  # noqa: S607
+        ["git", "rev-parse", "HEAD"]  # noqa: S603, S607
     ).strip().decode("utf-8")
 
 

superset-frontend/src/hooks/useLocale.ts

@@ -31,6 +31,7 @@ import 'dayjs/locale/pt';
 import 'dayjs/locale/pt-br';
 import 'dayjs/locale/ru';
 import 'dayjs/locale/ko';
+import 'dayjs/locale/cs';
 import 'dayjs/locale/sk';
 import 'dayjs/locale/sl';
 import 'dayjs/locale/nl';
@@ -50,6 +51,7 @@ export const LOCALE_MAPPING = {
   pt_BR: () => import('antd/locale/pt_BR'),
   ru: () => import('antd/locale/ru_RU'),
   ko: () => import('antd/locale/ko_KR'),
+  cs: () => import('antd/locale/cs_CZ'),
   sk: () => import('antd/locale/sk_SK'),
   sl: () => import('antd/locale/sl_SI'),
   nl: () => import('antd/locale/nl_NL'),

superset/commands/css/create.py

@@ -0,0 +1,46 @@
+# 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.
+import logging
+from functools import partial
+from typing import Any
+
+from superset.commands.base import BaseCommand
+from superset.commands.css.exceptions import (
+    CssTemplateCreateFailedError,
+    CssTemplateInvalidError,
+)
+from superset.daos.css import CssTemplateDAO
+from superset.models.core import CssTemplate
+from superset.utils.decorators import on_error, transaction
+
+logger = logging.getLogger(__name__)
+
+
+class CreateCssTemplateCommand(BaseCommand):
+    def __init__(self, properties: dict[str, Any]):
+        self._properties = properties
+
+    @transaction(on_error=partial(on_error, reraise=CssTemplateCreateFailedError))
+    def run(self) -> CssTemplate:
+        self.validate()
+        return CssTemplateDAO.create(attributes=self._properties)
+
+    def validate(self) -> None:
+        if not self._properties.get("template_name", "").strip():
+            raise CssTemplateInvalidError()
+        if "css" not in self._properties:
+            raise CssTemplateInvalidError()

superset/commands/css/exceptions.py

@@ -16,7 +16,13 @@
 # under the License.
 from flask_babel import lazy_gettext as _
 
-from superset.commands.exceptions import CommandException, DeleteFailedError
+from superset.commands.exceptions import (
+    CommandException,
+    CommandInvalidError,
+    CreateFailedError,
+    DeleteFailedError,
+    UpdateFailedError,
+)
 
 
 class CssTemplateDeleteFailedError(DeleteFailedError):
@@ -25,3 +31,15 @@ class CssTemplateDeleteFailedError(DeleteFailedError):
 
 class CssTemplateNotFoundError(CommandException):
     message = _("CSS template not found.")
+
+
+class CssTemplateCreateFailedError(CreateFailedError):
+    message = _("CSS template could not be created.")
+
+
+class CssTemplateInvalidError(CommandInvalidError):
+    message = _("CSS template parameters are invalid.")
+
+
+class CssTemplateUpdateFailedError(UpdateFailedError):
+    message = _("CSS template could not be updated.")

superset/commands/css/update.py

@@ -0,0 +1,53 @@
+# 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.
+
+import logging
+from functools import partial
+from typing import Any
+
+from superset.commands.base import BaseCommand
+from superset.commands.css.exceptions import (
+    CssTemplateInvalidError,
+    CssTemplateNotFoundError,
+    CssTemplateUpdateFailedError,
+)
+from superset.daos.css import CssTemplateDAO
+from superset.models.core import CssTemplate
+from superset.utils.decorators import on_error, transaction
+
+logger = logging.getLogger(__name__)
+
+
+class UpdateCssTemplateCommand(BaseCommand):
+    def __init__(self, model_id: int, properties: dict[str, Any]):
+        self._model_id = model_id
+        self._properties = properties
+        self._model: CssTemplate | None = None
+
+    @transaction(on_error=partial(on_error, reraise=CssTemplateUpdateFailedError))
+    def run(self) -> CssTemplate:
+        self.validate()
+        assert self._model
+        return CssTemplateDAO.update(self._model, attributes=self._properties)
+
+    def validate(self) -> None:
+        self._model = CssTemplateDAO.find_by_id(self._model_id)
+        if not self._model:
+            raise CssTemplateNotFoundError()
+        template_name = self._properties.get("template_name")
+        if template_name is not None and not template_name.strip():
+            raise CssTemplateInvalidError()

superset/mcp_service/app.py

@@ -145,6 +145,16 @@ Database Connections:
 - list_databases: List database connections with advanced filters (1-based pagination)
 - get_database_info: Get detailed database connection info by ID (backend, capabilities)
 
+CSS Templates:
+- list_css_templates: List CSS templates with advanced filters (1-based pagination)
+- get_css_template_info: Get CSS template details by ID (includes full css content)
+- create_css_template: Create a new named CSS template for dashboard styling
+- update_css_template: Update an existing CSS template's name or CSS content
+
+Themes:
+- list_themes: List themes with advanced filters (1-based pagination)
+- get_theme_info: Get theme details by ID or UUID (includes json_data configuration)
+
 User and Role Management:
 - list_users: List users with filtering (1-based pagination, admin only)
 - get_user_info: Get user details by ID (admin only)
@@ -178,7 +188,7 @@ SQL Lab Integration:
 - get_query_info: Get SQL query history details by ID
 
 Schema Discovery:
-- get_schema: Get schema metadata for chart/dataset/dashboard (columns, filters)
+- get_schema: Get schema metadata for chart/dataset/dashboard/database/css_template/theme (columns, filters)
 
 Action Logs (requires SUPERSET_LOG_VIEW and FAB_ADD_SECURITY_VIEWS):
 - list_action_logs: List user action logs with filtering and pagination (defaults to last 7 days)
@@ -676,6 +686,12 @@ from superset.mcp_service.chart.tool import (  # noqa: F401, E402
     update_chart,
     update_chart_preview,
 )
+from superset.mcp_service.css_template.tool import (  # noqa: F401, E402
+    create_css_template,
+    get_css_template_info,
+    list_css_templates,
+    update_css_template,
+)
 from superset.mcp_service.dashboard.tool import (  # noqa: F401, E402
     add_chart_to_existing_dashboard,
     generate_dashboard,
@@ -732,6 +748,10 @@ from superset.mcp_service.task.tool import (  # noqa: F401, E402
     get_task_info,
     list_tasks,
 )
+from superset.mcp_service.theme.tool import (  # noqa: F401, E402
+    get_theme_info,
+    list_themes,
+)
 from superset.mcp_service.user.tool import (  # noqa: F401, E402
     get_user_info,
     list_users,

superset/mcp_service/common/schema_discovery.py

@@ -635,3 +635,82 @@ CHART_ALL_COLUMNS: list[str] = []
 DATASET_ALL_COLUMNS: list[str] = []
 DASHBOARD_ALL_COLUMNS: list[str] = []
 DATABASE_ALL_COLUMNS: list[str] = []
+
+
+# CSS Template configuration
+CSS_TEMPLATE_DEFAULT_COLUMNS = [
+    "id",
+    "uuid",
+    "template_name",
+]
+CSS_TEMPLATE_SORTABLE_COLUMNS = [
+    "id",
+    "template_name",
+    "changed_on",
+    "created_on",
+]
+CSS_TEMPLATE_SEARCH_COLUMNS = ["template_name"]
+CSS_TEMPLATE_FILTER_COLUMNS: dict[str, list[str]] = {
+    "template_name": ["eq", "sw", "ilike"],
+    "created_by_fk": ["eq"],
+}
+CSS_TEMPLATE_EXTRA_COLUMNS: dict[str, ColumnMetadata] = {
+    "created_by_name": ColumnMetadata(
+        name="created_by_name",
+        description="Username of the creator",
+        type="str",
+        is_default=False,
+    ),
+    "changed_by_name": ColumnMetadata(
+        name="changed_by_name",
+        description="Username of the last modifier",
+        type="str",
+        is_default=False,
+    ),
+}
+
+
+def get_css_template_columns() -> list[ColumnMetadata]:
+    """Get column metadata for CssTemplate model dynamically."""
+    from superset.models.core import CssTemplate
+
+    return get_columns_from_model(
+        CssTemplate,
+        CSS_TEMPLATE_DEFAULT_COLUMNS,
+        CSS_TEMPLATE_EXTRA_COLUMNS,
+        exclude_columns=set(USER_DIRECTORY_FIELDS)
+        - {"created_by_name", "changed_by_name"},
+    )
+
+
+# Theme configuration
+THEME_DEFAULT_COLUMNS = [
+    "id",
+    "theme_name",
+    "uuid",
+]
+THEME_SORTABLE_COLUMNS = [
+    "id",
+    "theme_name",
+    "changed_on",
+    "created_on",
+]
+THEME_SEARCH_COLUMNS = ["theme_name"]
+THEME_FILTER_COLUMNS: dict[str, list[str]] = {
+    "theme_name": ["eq", "sw", "ilike"],
+    "is_system": ["eq"],
+    "is_system_default": ["eq"],
+    "is_system_dark": ["eq"],
+    "created_by_fk": ["eq"],
+}
+
+
+def get_theme_columns() -> list[ColumnMetadata]:
+    """Get column metadata for Theme model dynamically."""
+    from superset.models.core import Theme
+
+    return get_columns_from_model(
+        Theme,
+        THEME_DEFAULT_COLUMNS,
+        exclude_columns=set(USER_DIRECTORY_FIELDS),
+    )

superset/mcp_service/constants.py

@@ -19,7 +19,9 @@
 from typing import Literal
 
 # Supported model types for schema discovery and MCP tools
-ModelType = Literal["chart", "dataset", "dashboard", "database"]
+ModelType = Literal[
+    "chart", "dataset", "dashboard", "database", "css_template", "theme"
+]
 
 # Pagination defaults
 DEFAULT_PAGE_SIZE = 10  # Default number of items per page

superset/mcp_service/css_template/__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.

superset/mcp_service/css_template/schemas.py

@@ -0,0 +1,384 @@
+# 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.
+
+"""
+Pydantic schemas for CSS template-related responses
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Annotated, Any, Dict, List, Literal
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+    model_serializer,
+    model_validator,
+    PositiveInt,
+)
+
+from superset.daos.base import ColumnOperator, ColumnOperatorEnum
+from superset.mcp_service.common.cache_schemas import MetadataCacheControl
+from superset.mcp_service.constants import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
+from superset.mcp_service.system.schemas import PaginationInfo
+from superset.mcp_service.utils.sanitization import sanitize_for_llm_context
+from superset.mcp_service.utils.schema_utils import (
+    parse_json_or_list,
+    parse_json_or_model_list,
+)
+
+
+class CssTemplateFilter(ColumnOperator):
+    """
+    Filter object for CSS template listing.
+    col: The column to filter on. Must be one of the allowed filter fields.
+    opr: The operator to use. Must be one of the supported operators.
+    value: The value to filter by (type depends on col and opr).
+    """
+
+    col: Literal["template_name", "created_by_fk"] = Field(
+        ...,
+        description="Column to filter on. Use get_schema(model_type='css_template') "
+        "for available filter columns. To filter by creator, first call find_users "
+        "to resolve a name to a user ID, then filter by created_by_fk with "
+        "that integer ID.",
+    )
+    opr: ColumnOperatorEnum = Field(
+        ...,
+        description="Operator to use.",
+    )
+    value: str | int | float | bool | List[str | int | float | bool] = Field(
+        ..., description="Value to filter by (type depends on col and opr)"
+    )
+
+
+class CssTemplateInfo(BaseModel):
+    id: int | None = Field(None, description="CSS template ID")
+    uuid: str | None = Field(None, description="CSS template UUID")
+    template_name: str | None = Field(None, description="CSS template name")
+    css: str | None = Field(
+        None,
+        description="CSS content (can be large; request via select_columns=['css'])",
+    )
+    changed_on: str | datetime | None = Field(
+        None, description="Last modification timestamp"
+    )
+    created_on: str | datetime | None = Field(None, description="Creation timestamp")
+    created_by_name: str | None = Field(None, description="Username of the creator")
+    changed_by_name: str | None = Field(
+        None, description="Username of the last modifier"
+    )
+
+    model_config = ConfigDict(
+        from_attributes=True,
+        ser_json_timedelta="iso8601",
+        populate_by_name=True,
+    )
+
+    @model_serializer(mode="wrap")
+    def _filter_fields_by_context(self, serializer: Any, info: Any) -> Dict[str, Any]:
+        """Filter fields based on serialization context.
+
+        If context contains 'select_columns', only include those fields.
+        Otherwise, include all fields (default behavior).
+        """
+        data = serializer(self)
+
+        if info.context and isinstance(info.context, dict):
+            select_columns = info.context.get("select_columns")
+            if select_columns:
+                requested_fields = set(select_columns)
+                return {k: v for k, v in data.items() if k in requested_fields}
+
+        return data
+
+
+class CssTemplateList(BaseModel):
+    css_templates: List[CssTemplateInfo]
+    count: int
+    total_count: int
+    page: int
+    page_size: int
+    total_pages: int
+    has_previous: bool
+    has_next: bool
+    columns_requested: List[str] = Field(
+        default_factory=list,
+        description="Requested columns for the response",
+    )
+    columns_loaded: List[str] = Field(
+        default_factory=list,
+        description="Columns that were actually loaded for each CSS template",
+    )
+    columns_available: List[str] = Field(
+        default_factory=list,
+        description="All columns available for selection via select_columns parameter",
+    )
+    sortable_columns: List[str] = Field(
+        default_factory=list,
+        description="Columns that can be used with order_column parameter",
+    )
+    filters_applied: List[CssTemplateFilter] = Field(
+        default_factory=list,
+        description="List of advanced filter dicts applied to the query.",
+    )
+    pagination: PaginationInfo | None = None
+    timestamp: datetime | None = None
+    model_config = ConfigDict(ser_json_timedelta="iso8601")
+
+
+class ListCssTemplatesRequest(MetadataCacheControl):
+    """Request schema for list_css_templates."""
+
+    filters: Annotated[
+        List[CssTemplateFilter],
+        Field(
+            default_factory=list,
+            description="List of filter objects (column, operator, value). Cannot be "
+            "used together with 'search'.",
+        ),
+    ]
+    select_columns: Annotated[
+        List[str],
+        Field(
+            default_factory=list,
+            description="List of columns to select. Defaults to common columns if not "
+            "specified. Use select_columns=['css'] to include the CSS content.",
+        ),
+    ]
+    search: Annotated[
+        str | None,
+        Field(
+            default=None,
+            description="Text search string to match against CSS template fields. "
+            "Cannot be used together with 'filters'.",
+        ),
+    ]
+    order_column: Annotated[
+        str | None, Field(default=None, description="Column to order results by")
+    ]
+    order_direction: Annotated[
+        Literal["asc", "desc"],
+        Field(
+            default="desc", description="Direction to order results ('asc' or 'desc')"
+        ),
+    ]
+    page: Annotated[
+        PositiveInt,
+        Field(default=1, description="Page number for pagination (1-based)"),
+    ]
+    page_size: Annotated[
+        int,
+        Field(
+            default=DEFAULT_PAGE_SIZE,
+            gt=0,
+            le=MAX_PAGE_SIZE,
+            description=f"Number of items per page (max {MAX_PAGE_SIZE})",
+        ),
+    ]
+
+    @field_validator("filters", mode="before")
+    @classmethod
+    def parse_filters(cls, v: Any) -> List[CssTemplateFilter]:
+        """Accept both JSON string and list of objects."""
+        return parse_json_or_model_list(v, CssTemplateFilter, "filters")
+
+    @field_validator("select_columns", mode="before")
+    @classmethod
+    def parse_columns(cls, v: Any) -> List[str]:
+        """Accept JSON array, list, or comma-separated string."""
+        return parse_json_or_list(v, "select_columns")
+
+    @model_validator(mode="after")
+    def validate_search_and_filters(self) -> "ListCssTemplatesRequest":
+        """Prevent using both search and filters simultaneously."""
+        if self.search and self.filters:
+            raise ValueError(
+                "Cannot use both 'search' and 'filters' parameters simultaneously. "
+                "Use either 'search' for text-based searching across multiple fields, "
+                "or 'filters' for precise column-based filtering, but not both."
+            )
+        return self
+
+
+class CssTemplateError(BaseModel):
+    error: str = Field(..., description="Error message")
+    error_type: str = Field(..., description="Type of error")
+    timestamp: str | datetime | None = Field(None, description="Error timestamp")
+    model_config = ConfigDict(ser_json_timedelta="iso8601")
+
+    @field_validator("error")
+    @classmethod
+    def sanitize_error_for_llm_context(cls, value: str) -> str:
+        """Wrap error text before it is exposed to LLM context."""
+        return sanitize_for_llm_context(value, field_path=("error",))
+
+    @classmethod
+    def create(cls, error: str, error_type: str) -> "CssTemplateError":
+        """Create a standardized CssTemplateError with timestamp."""
+        from datetime import datetime, timezone
+
+        return cls(
+            error=error, error_type=error_type, timestamp=datetime.now(timezone.utc)
+        )
+
+
+class GetCssTemplateInfoRequest(MetadataCacheControl):
+    """Request schema for get_css_template_info with support for ID or UUID."""
+
+    identifier: Annotated[
+        int | str,
+        Field(description="CSS template identifier - can be numeric ID or UUID string"),
+    ]
+
+
+def _sanitize_css_template_info_for_llm_context(
+    css_template_info: CssTemplateInfo,
+) -> CssTemplateInfo:
+    """Wrap CSS template user-controlled fields before LLM exposure."""
+    payload = css_template_info.model_dump(mode="python")
+    for field_name in ("template_name", "css", "created_by_name", "changed_by_name"):
+        payload[field_name] = sanitize_for_llm_context(
+            payload.get(field_name),
+            field_path=(field_name,),
+        )
+    return CssTemplateInfo.model_validate(payload)
+
+
+def serialize_css_template_object(obj: Any) -> CssTemplateInfo | None:
+    if not obj:
+        return None
+
+    return _sanitize_css_template_info_for_llm_context(
+        CssTemplateInfo(
+            id=getattr(obj, "id", None),
+            uuid=str(getattr(obj, "uuid", "")) if getattr(obj, "uuid", None) else None,
+            template_name=getattr(obj, "template_name", None),
+            css=getattr(obj, "css", None),
+            changed_on=getattr(obj, "changed_on", None),
+            created_on=getattr(obj, "created_on", None),
+            created_by_name=getattr(obj, "created_by_name", None) or None,
+            changed_by_name=getattr(obj, "changed_by_name", None) or None,
+        )
+    )
+
+
+class CreateCssTemplateRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    template_name: str = Field(
+        ...,
+        min_length=1,
+        max_length=250,
+        description="Name for the CSS template.",
+    )
+    css: str = Field(
+        ...,
+        description="CSS content for the template.",
+    )
+
+    @field_validator("template_name")
+    @classmethod
+    def template_name_must_not_be_empty(cls, v: str) -> str:
+        if not v.strip():
+            raise ValueError("template_name must not be empty")
+        return v.strip()
+
+
+class CreateCssTemplateResponse(BaseModel):
+    """Response schema for create_css_template."""
+
+    id: int | None = Field(
+        None,
+        description="ID of the created CSS template. None if creation failed.",
+    )
+    template_name: str | None = Field(
+        None,
+        description="Name of the created CSS template.",
+    )
+    css: str | None = Field(
+        None,
+        description="CSS content of the created template.",
+    )
+    error: str | None = Field(
+        None,
+        description="Error message if creation failed, otherwise null.",
+    )
+
+    @field_validator("error")
+    @classmethod
+    def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
+        """Sanitize error text before it is exposed to LLM context."""
+        if value is None:
+            return value
+        return sanitize_for_llm_context(value, field_path=("error",))
+
+
+class UpdateCssTemplateRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: int = Field(..., description="ID of the CSS template to update.")
+    template_name: str | None = Field(
+        None,
+        max_length=250,
+        description="New name for the CSS template.",
+    )
+    css: str | None = Field(
+        None,
+        description="New CSS content for the template.",
+    )
+
+    @field_validator("template_name")
+    @classmethod
+    def template_name_must_not_be_empty(cls, v: str | None) -> str | None:
+        if v is not None:
+            if not v.strip():
+                raise ValueError("template_name must not be empty")
+            return v.strip()
+        return v
+
+
+class UpdateCssTemplateResponse(BaseModel):
+    """Response schema for update_css_template."""
+
+    id: int | None = Field(
+        None,
+        description="ID of the updated CSS template. None if update failed.",
+    )
+    template_name: str | None = Field(
+        None,
+        description="Name of the updated CSS template.",
+    )
+    css: str | None = Field(
+        None,
+        description="CSS content of the updated template.",
+    )
+    error: str | None = Field(
+        None,
+        description="Error message if update failed, otherwise null.",
+    )
+
+    @field_validator("error")
+    @classmethod
+    def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
+        """Sanitize error text before it is exposed to LLM context."""
+        if value is None:
+            return value
+        return sanitize_for_llm_context(value, field_path=("error",))

superset/mcp_service/css_template/tool/__init__.py

@@ -0,0 +1,28 @@
+# 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.
+
+from .create_css_template import create_css_template
+from .get_css_template_info import get_css_template_info
+from .list_css_templates import list_css_templates
+from .update_css_template import update_css_template
+
+__all__ = [
+    "list_css_templates",
+    "get_css_template_info",
+    "create_css_template",
+    "update_css_template",
+]

superset/mcp_service/css_template/tool/create_css_template.py

@@ -0,0 +1,95 @@
+# 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.
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.css_template.schemas import (
+    CreateCssTemplateRequest,
+    CreateCssTemplateResponse,
+)
+
+
+@tool(
+    tags=["mutate"],
+    class_permission_name="CssTemplate",
+    method_permission_name="write",
+    annotations=ToolAnnotations(
+        title="Create CSS template",
+        readOnlyHint=False,
+        destructiveHint=False,
+    ),
+)
+async def create_css_template(
+    request: CreateCssTemplateRequest, ctx: Context
+) -> CreateCssTemplateResponse:
+    """Create a new CSS template that can be applied to dashboards.
+
+    Use this tool when a user wants to save a CSS stylesheet as a named
+    template for reuse across multiple dashboards.
+
+    The returned ``id`` can be used when configuring dashboard appearance.
+    """
+    await ctx.info("Creating CSS template: template_name=%r" % (request.template_name,))
+
+    try:
+        from superset.commands.css.create import CreateCssTemplateCommand
+        from superset.commands.css.exceptions import (
+            CssTemplateCreateFailedError,
+            CssTemplateInvalidError,
+        )
+
+        with event_logger.log_context(action="mcp.create_css_template.create"):
+            template = CreateCssTemplateCommand(
+                {
+                    "template_name": request.template_name,
+                    "css": request.css,
+                }
+            ).run()
+
+        await ctx.info(
+            "CSS template created: id=%s, template_name=%r"
+            % (template.id, template.template_name)
+        )
+
+        return CreateCssTemplateResponse(
+            id=template.id,
+            template_name=template.template_name,
+            css=template.css,
+        )
+
+    except CssTemplateInvalidError as exc:
+        await ctx.warning("CSS template validation failed: %s" % (str(exc),))
+        return CreateCssTemplateResponse(
+            template_name=request.template_name,
+            css=request.css,
+            error=str(exc),
+        )
+    except CssTemplateCreateFailedError as exc:
+        await ctx.error("CSS template creation failed: %s" % (str(exc),))
+        return CreateCssTemplateResponse(
+            template_name=request.template_name,
+            css=request.css,
+            error=f"Failed to create CSS template: {exc}",
+        )
+    except Exception as exc:
+        await ctx.error(
+            "Unexpected error creating CSS template: %s: %s"
+            % (type(exc).__name__, str(exc))
+        )
+        raise

superset/mcp_service/css_template/tool/get_css_template_info.py

@@ -0,0 +1,108 @@
+# 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.
+
+"""
+Get CSS template info FastMCP tool
+"""
+
+import logging
+from datetime import datetime, timezone
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.css_template.schemas import (
+    CssTemplateError,
+    CssTemplateInfo,
+    GetCssTemplateInfoRequest,
+    serialize_css_template_object,
+)
+from superset.mcp_service.mcp_core import ModelGetInfoCore
+
+logger = logging.getLogger(__name__)
+
+
+@tool(
+    tags=["discovery"],
+    class_permission_name="CssTemplate",
+    annotations=ToolAnnotations(
+        title="Get CSS template info",
+        readOnlyHint=True,
+        destructiveHint=False,
+    ),
+)
+async def get_css_template_info(
+    request: GetCssTemplateInfoRequest, ctx: Context
+) -> CssTemplateInfo | CssTemplateError:
+    """Get CSS template details by ID or UUID.
+
+    Returns the full CSS template including the css content.
+
+    IMPORTANT FOR LLM CLIENTS:
+    - Use numeric ID (e.g., 123) or UUID string (e.g., "a1b2c3d4-...")
+    - To find a CSS template ID, use the list_css_templates tool first
+
+    Example usage:
+    ```json
+    {
+        "identifier": 1
+    }
+    ```
+    """
+    await ctx.info(
+        "Retrieving CSS template information: identifier=%s" % (request.identifier,)
+    )
+
+    try:
+        from superset.daos.css import CssTemplateDAO
+
+        with event_logger.log_context(action="mcp.get_css_template_info.lookup"):
+            get_tool = ModelGetInfoCore(
+                dao_class=CssTemplateDAO,
+                output_schema=CssTemplateInfo,
+                error_schema=CssTemplateError,
+                serializer=serialize_css_template_object,
+                supports_slug=False,
+                logger=logger,
+            )
+
+            result = get_tool.run_tool(request.identifier)
+
+        if isinstance(result, CssTemplateInfo):
+            await ctx.info(
+                "CSS template information retrieved successfully: "
+                "id=%s, template_name=%s" % (result.id, result.template_name)
+            )
+        else:
+            await ctx.warning(
+                "CSS template retrieval failed: error_type=%s, error=%s"
+                % (result.error_type, result.error)
+            )
+
+        return result
+
+    except Exception as e:
+        await ctx.error(
+            "CSS template information retrieval failed: identifier=%s, error=%s, "
+            "error_type=%s" % (request.identifier, str(e), type(e).__name__)
+        )
+        return CssTemplateError(
+            error=f"Failed to get CSS template info: {str(e)}",
+            error_type="InternalError",
+            timestamp=datetime.now(timezone.utc),
+        )

superset/mcp_service/css_template/tool/list_css_templates.py

@@ -0,0 +1,150 @@
+# 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.
+
+"""
+List CSS templates FastMCP tool
+"""
+
+import logging
+from typing import TYPE_CHECKING
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+if TYPE_CHECKING:
+    from superset.models.core import CssTemplate
+
+from superset.extensions import event_logger
+from superset.mcp_service.css_template.schemas import (
+    CssTemplateError,
+    CssTemplateFilter,
+    CssTemplateInfo,
+    CssTemplateList,
+    ListCssTemplatesRequest,
+    serialize_css_template_object,
+)
+from superset.mcp_service.mcp_core import ModelListCore
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_LIST_CSS_TEMPLATES_REQUEST = ListCssTemplatesRequest()
+
+
+@tool(
+    tags=["core"],
+    class_permission_name="CssTemplate",
+    annotations=ToolAnnotations(
+        title="List CSS templates",
+        readOnlyHint=True,
+        destructiveHint=False,
+    ),
+)
+async def list_css_templates(
+    request: ListCssTemplatesRequest | None = None,
+    ctx: Context | None = None,
+) -> CssTemplateList | CssTemplateError:
+    """List CSS templates with filtering and search.
+
+    Returns CSS template metadata including name. Use select_columns=['css']
+    to include the CSS content (omitted by default due to size).
+
+    Sortable columns for order_column: id, template_name, changed_on,
+    created_on
+    """
+    if ctx is None:
+        raise RuntimeError("FastMCP context is required for list_css_templates")
+
+    request = request or _DEFAULT_LIST_CSS_TEMPLATES_REQUEST.model_copy(deep=True)
+
+    await ctx.info(
+        "Listing CSS templates: page=%s, page_size=%s, search=%s"
+        % (request.page, request.page_size, request.search)
+    )
+    await ctx.debug(
+        "CSS template listing parameters: filters=%s, order_column=%s, "
+        "order_direction=%s, select_columns=%s"
+        % (
+            request.filters,
+            request.order_column,
+            request.order_direction,
+            request.select_columns,
+        )
+    )
+
+    try:
+        from superset.daos.css import CssTemplateDAO
+        from superset.mcp_service.common.schema_discovery import (
+            CSS_TEMPLATE_DEFAULT_COLUMNS,
+            CSS_TEMPLATE_SORTABLE_COLUMNS,
+            get_all_column_names,
+            get_css_template_columns,
+        )
+
+        all_columns = get_all_column_names(get_css_template_columns())
+
+        def _serialize_css_template(
+            obj: "CssTemplate | None", cols: list[str] | None
+        ) -> CssTemplateInfo | None:
+            return serialize_css_template_object(obj)
+
+        list_tool = ModelListCore(
+            dao_class=CssTemplateDAO,
+            output_schema=CssTemplateInfo,
+            item_serializer=_serialize_css_template,
+            filter_type=CssTemplateFilter,
+            default_columns=CSS_TEMPLATE_DEFAULT_COLUMNS,
+            search_columns=["template_name"],
+            list_field_name="css_templates",
+            output_list_schema=CssTemplateList,
+            all_columns=all_columns,
+            sortable_columns=CSS_TEMPLATE_SORTABLE_COLUMNS,
+            logger=logger,
+        )
+
+        with event_logger.log_context(action="mcp.list_css_templates.query"):
+            result = list_tool.run_tool(
+                filters=request.filters,
+                search=request.search,
+                select_columns=request.select_columns,
+                order_column=request.order_column,
+                order_direction=request.order_direction,
+                page=max(request.page - 1, 0),
+                page_size=request.page_size,
+            )
+
+        await ctx.info(
+            "CSS templates listed successfully: count=%s, total_count=%s"
+            % (
+                len(result.css_templates) if hasattr(result, "css_templates") else 0,
+                getattr(result, "total_count", None),
+            )
+        )
+
+        columns_to_filter = result.columns_requested
+        with event_logger.log_context(action="mcp.list_css_templates.serialization"):
+            return result.model_dump(
+                mode="json",
+                context={"select_columns": columns_to_filter},
+            )
+
+    except Exception as e:
+        await ctx.error(
+            "CSS template listing failed: page=%s, page_size=%s, error=%s, "
+            "error_type=%s"
+            % (request.page, request.page_size, str(e), type(e).__name__)
+        )
+        raise

superset/mcp_service/css_template/tool/update_css_template.py

@@ -0,0 +1,111 @@
+# 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.
+
+from typing import Any
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.css_template.schemas import (
+    UpdateCssTemplateRequest,
+    UpdateCssTemplateResponse,
+)
+
+
+@tool(
+    tags=["mutate"],
+    class_permission_name="CssTemplate",
+    method_permission_name="write",
+    annotations=ToolAnnotations(
+        title="Update CSS template",
+        readOnlyHint=False,
+        destructiveHint=False,
+    ),
+)
+async def update_css_template(
+    request: UpdateCssTemplateRequest, ctx: Context
+) -> UpdateCssTemplateResponse:
+    """Update an existing CSS template's name or CSS content.
+
+    Use this tool when a user wants to rename a CSS template or replace its
+    CSS content. At least one of ``template_name`` or ``css`` must be provided.
+
+    The template is identified by its ``id``.
+    """
+    await ctx.info(
+        "Updating CSS template: id=%s, fields=%r"
+        % (
+            request.id,
+            [f for f in ("template_name", "css") if getattr(request, f) is not None],
+        )
+    )
+
+    try:
+        from superset.commands.css.exceptions import (
+            CssTemplateInvalidError,
+            CssTemplateNotFoundError,
+            CssTemplateUpdateFailedError,
+        )
+        from superset.commands.css.update import UpdateCssTemplateCommand
+
+        properties: dict[str, Any] = {}
+        if request.template_name is not None:
+            properties["template_name"] = request.template_name
+        if request.css is not None:
+            properties["css"] = request.css
+
+        if not properties:
+            return UpdateCssTemplateResponse(
+                error="At least one of template_name or css must be provided.",
+            )
+
+        with event_logger.log_context(action="mcp.update_css_template.update"):
+            template = UpdateCssTemplateCommand(request.id, properties).run()
+
+        await ctx.info(
+            "CSS template updated: id=%s, template_name=%r"
+            % (template.id, template.template_name)
+        )
+
+        return UpdateCssTemplateResponse(
+            id=template.id,
+            template_name=template.template_name,
+            css=template.css,
+        )
+
+    except CssTemplateNotFoundError:
+        await ctx.warning("CSS template not found: id=%s" % (request.id,))
+        return UpdateCssTemplateResponse(
+            error="CSS template not found: %s" % (request.id,),
+        )
+    except CssTemplateInvalidError as exc:
+        await ctx.warning("CSS template validation failed: %s" % (str(exc),))
+        return UpdateCssTemplateResponse(
+            error=str(exc),
+        )
+    except CssTemplateUpdateFailedError as exc:
+        await ctx.error("CSS template update failed: %s" % (str(exc),))
+        return UpdateCssTemplateResponse(
+            error="Failed to update CSS template: %s" % (exc,),
+        )
+    except Exception as exc:
+        await ctx.error(
+            "Unexpected error updating CSS template: %s: %s"
+            % (type(exc).__name__, str(exc))
+        )
+        raise

superset/mcp_service/mcp_core.py

@@ -733,6 +733,7 @@ class ModelGetSchemaCore(BaseCore, Generic[S]):
         default_sort: str = "changed_on",
         default_sort_direction: Literal["asc", "desc"] = "desc",
         exclude_filter_columns: set[str] | None = None,
+        filter_columns_override: dict[str, list[str]] | None = None,
         logger: logging.Logger | None = None,
     ) -> None:
         """
@@ -750,6 +751,9 @@ class ModelGetSchemaCore(BaseCore, Generic[S]):
             default_sort_direction: Default sort direction
             exclude_filter_columns: Column names to omit from filter discovery
                 (e.g., sensitive fields like passwords or connection URIs)
+            filter_columns_override: When set, use this mapping directly as the
+                filter_columns output instead of querying the DAO. Use this to
+                restrict advertised filters to the exact set the list tool accepts.
             logger: Optional logger instance
         """
         super().__init__(logger)
@@ -770,9 +774,12 @@ class ModelGetSchemaCore(BaseCore, Generic[S]):
         # Hide user-directory columns from filter discovery, except the small
         # set callers may legitimately filter by ID (resolved via find_users).
         self.exclude_filter_columns.update(USER_DIRECTORY_FIELDS - USER_FILTER_FIELDS)
+        self.filter_columns_override = filter_columns_override
 
     def _get_filter_columns(self) -> Dict[str, List[str]]:
         """Get filterable columns and operators from the DAO."""
+        if self.filter_columns_override is not None:
+            return self.filter_columns_override
         try:
             filterable = self.dao_class.get_filterable_columns_and_operators()
             # Defensive handling: ensure we have a valid mapping

superset/mcp_service/system/tool/get_schema.py

@@ -30,10 +30,15 @@ from fastmcp import Context
 from superset_core.mcp.decorators import tool, ToolAnnotations
 
 from superset.extensions import event_logger
+from superset.mcp_service.auth import MCPPermissionDeniedError
 from superset.mcp_service.common.schema_discovery import (
     CHART_DEFAULT_COLUMNS,
     CHART_SEARCH_COLUMNS,
     CHART_SORTABLE_COLUMNS,
+    CSS_TEMPLATE_DEFAULT_COLUMNS,
+    CSS_TEMPLATE_FILTER_COLUMNS,
+    CSS_TEMPLATE_SEARCH_COLUMNS,
+    CSS_TEMPLATE_SORTABLE_COLUMNS,
     DASHBOARD_DEFAULT_COLUMNS,
     DASHBOARD_SEARCH_COLUMNS,
     DASHBOARD_SORTABLE_COLUMNS,
@@ -44,12 +49,18 @@ from superset.mcp_service.common.schema_discovery import (
     DATASET_SEARCH_COLUMNS,
     DATASET_SORTABLE_COLUMNS,
     get_chart_columns,
+    get_css_template_columns,
     get_dashboard_columns,
     get_database_columns,
     get_dataset_columns,
+    get_theme_columns,
     GetSchemaRequest,
     GetSchemaResponse,
     ModelSchemaInfo,
+    THEME_DEFAULT_COLUMNS,
+    THEME_FILTER_COLUMNS,
+    THEME_SEARCH_COLUMNS,
+    THEME_SORTABLE_COLUMNS,
 )
 from superset.mcp_service.constants import ModelType
 from superset.mcp_service.mcp_core import ModelGetSchemaCore
@@ -144,6 +155,44 @@ def _get_database_schema_core() -> ModelGetSchemaCore[ModelSchemaInfo]:
     )
 
 
+def _get_css_template_schema_core() -> ModelGetSchemaCore[ModelSchemaInfo]:
+    """Create CSS template schema core with dynamically extracted columns."""
+    from superset.daos.css import CssTemplateDAO
+
+    return ModelGetSchemaCore(
+        model_type="css_template",
+        dao_class=CssTemplateDAO,
+        output_schema=ModelSchemaInfo,
+        select_columns=get_css_template_columns(),
+        sortable_columns=CSS_TEMPLATE_SORTABLE_COLUMNS,
+        default_columns=CSS_TEMPLATE_DEFAULT_COLUMNS,
+        search_columns=CSS_TEMPLATE_SEARCH_COLUMNS,
+        default_sort="changed_on",
+        default_sort_direction="desc",
+        filter_columns_override=CSS_TEMPLATE_FILTER_COLUMNS,
+        logger=logger,
+    )
+
+
+def _get_theme_schema_core() -> ModelGetSchemaCore[ModelSchemaInfo]:
+    """Create theme schema core with dynamically extracted columns."""
+    from superset.daos.theme import ThemeDAO
+
+    return ModelGetSchemaCore(
+        model_type="theme",
+        dao_class=ThemeDAO,
+        output_schema=ModelSchemaInfo,
+        select_columns=get_theme_columns(),
+        sortable_columns=THEME_SORTABLE_COLUMNS,
+        default_columns=THEME_DEFAULT_COLUMNS,
+        search_columns=THEME_SEARCH_COLUMNS,
+        default_sort="changed_on",
+        default_sort_direction="desc",
+        filter_columns_override=THEME_FILTER_COLUMNS,
+        logger=logger,
+    )
+
+
 # Map model types to their core factory functions
 _SCHEMA_CORE_FACTORIES: dict[
     ModelType,
@@ -153,12 +202,25 @@ _SCHEMA_CORE_FACTORIES: dict[
     "dataset": _get_dataset_schema_core,
     "dashboard": _get_dashboard_schema_core,
     "database": _get_database_schema_core,
+    "css_template": _get_css_template_schema_core,
+    "theme": _get_theme_schema_core,
+}
+
+# Maps each model type to the FAB class permission name used by its tools.
+# Used for per-model-type inline RBAC checks instead of a single static
+# class_permission_name on the @tool decorator.
+_MODEL_TYPE_CLASS_PERMISSION: dict[ModelType, str] = {
+    "chart": "Chart",
+    "dataset": "Dataset",
+    "dashboard": "Dashboard",
+    "database": "Database",
+    "css_template": "CssTemplate",
+    "theme": "Theme",
 }
 
 
 @tool(
     tags=["discovery"],
-    class_permission_name="Dataset",
     annotations=ToolAnnotations(
         title="Get schema",
         readOnlyHint=True,
@@ -182,13 +244,39 @@ async def get_schema(
     Column metadata is extracted dynamically from SQLAlchemy models.
 
     Args:
-        model_type: One of "chart", "dataset", "dashboard", or "database"
+        model_type: One of "chart", "dataset", "dashboard", "database",
+            "css_template", or "theme"
 
     Returns:
         Comprehensive schema information for the requested model type
     """
     await ctx.info(f"Getting schema for model_type={request.model_type}")
 
+    # Per-model-type RBAC check (replaces the static class_permission_name on @tool,
+    # which wrongly gated all schema types behind Dataset permission).
+    class_permission = _MODEL_TYPE_CLASS_PERMISSION.get(request.model_type)
+    if class_permission:
+        from flask import current_app, g
+
+        from superset import security_manager
+
+        if current_app.config.get("MCP_RBAC_ENABLED", True) and not (
+            security_manager.can_access("can_read", class_permission)
+        ):
+            user_str = getattr(getattr(g, "user", None), "username", None)
+            logger.warning(
+                "get_schema RBAC denied: user=%s type=%s view=%s",
+                user_str,
+                request.model_type,
+                class_permission,
+            )
+            raise MCPPermissionDeniedError(
+                permission_name="can_read",
+                view_name=class_permission,
+                user=user_str,
+                tool_name="get_schema",
+            )
+
     can_view_data_model_metadata = user_can_view_data_model_metadata()
     if not can_view_data_model_metadata and request.model_type in {
         "dataset",

superset/mcp_service/theme/__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.

superset/mcp_service/theme/schemas.py

@@ -0,0 +1,305 @@
+# 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.
+
+"""
+Pydantic schemas for theme-related responses
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Annotated, Any, Dict, List, Literal
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+    model_serializer,
+    model_validator,
+    PositiveInt,
+)
+
+from superset.daos.base import ColumnOperator, ColumnOperatorEnum
+from superset.mcp_service.common.cache_schemas import MetadataCacheControl
+from superset.mcp_service.constants import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
+from superset.mcp_service.system.schemas import PaginationInfo
+from superset.mcp_service.utils.sanitization import sanitize_for_llm_context
+from superset.mcp_service.utils.schema_utils import (
+    parse_json_or_list,
+    parse_json_or_model_list,
+)
+from superset.utils import json as superset_json
+
+
+class ThemeFilter(ColumnOperator):
+    """
+    Filter object for theme listing.
+    col: The column to filter on. Must be one of the allowed filter fields.
+    opr: The operator to use. Must be one of the supported operators.
+    value: The value to filter by (type depends on col and opr).
+    """
+
+    col: Literal[
+        "theme_name",
+        "is_system",
+        "is_system_default",
+        "is_system_dark",
+        "created_by_fk",
+    ] = Field(
+        ...,
+        description="Column to filter on. Use get_schema(model_type='theme') for "
+        "available filter columns. To filter by creator, first call find_users "
+        "to resolve a name to a user ID, then filter by created_by_fk with "
+        "that integer ID.",
+    )
+    opr: ColumnOperatorEnum = Field(
+        ...,
+        description="Operator to use.",
+    )
+    value: str | int | float | bool | List[str | int | float | bool] = Field(
+        ..., description="Value to filter by (type depends on col and opr)"
+    )
+
+
+class ThemeInfo(BaseModel):
+    id: int | None = Field(None, description="Theme ID")
+    uuid: str | None = Field(None, description="Theme UUID")
+    theme_name: str | None = Field(None, description="Theme display name")
+    json_data: Dict[str, Any] | None = Field(
+        None,
+        description="Theme token configuration as a parsed dictionary",
+    )
+    is_system: bool | None = Field(None, description="Whether this is a system theme")
+    is_system_default: bool | None = Field(
+        None, description="Whether this is the default system theme"
+    )
+    is_system_dark: bool | None = Field(
+        None, description="Whether this is the dark system theme"
+    )
+    changed_on: str | datetime | None = Field(
+        None, description="Last modification timestamp"
+    )
+    created_on: str | datetime | None = Field(None, description="Creation timestamp")
+
+    model_config = ConfigDict(
+        from_attributes=True,
+        ser_json_timedelta="iso8601",
+        populate_by_name=True,
+    )
+
+    @model_serializer(mode="wrap")
+    def _filter_fields_by_context(self, serializer: Any, info: Any) -> Dict[str, Any]:
+        """Filter fields based on serialization context.
+
+        If context contains 'select_columns', only include those fields.
+        Otherwise, include all fields (default behavior).
+        """
+        data = serializer(self)
+
+        if info.context and isinstance(info.context, dict):
+            select_columns = info.context.get("select_columns")
+            if select_columns:
+                requested_fields = set(select_columns)
+                return {k: v for k, v in data.items() if k in requested_fields}
+
+        return data
+
+
+class ThemeList(BaseModel):
+    themes: List[ThemeInfo]
+    count: int
+    total_count: int
+    page: int
+    page_size: int
+    total_pages: int
+    has_previous: bool
+    has_next: bool
+    columns_requested: List[str] = Field(
+        default_factory=list,
+        description="Requested columns for the response",
+    )
+    columns_loaded: List[str] = Field(
+        default_factory=list,
+        description="Columns that were actually loaded for each theme",
+    )
+    columns_available: List[str] = Field(
+        default_factory=list,
+        description="All columns available for selection via select_columns parameter",
+    )
+    sortable_columns: List[str] = Field(
+        default_factory=list,
+        description="Columns that can be used with order_column parameter",
+    )
+    filters_applied: List[ThemeFilter] = Field(
+        default_factory=list,
+        description="List of advanced filter dicts applied to the query.",
+    )
+    pagination: PaginationInfo | None = None
+    timestamp: datetime | None = None
+    model_config = ConfigDict(ser_json_timedelta="iso8601")
+
+
+class ListThemesRequest(MetadataCacheControl):
+    """Request schema for list_themes."""
+
+    filters: Annotated[
+        List[ThemeFilter],
+        Field(
+            default_factory=list,
+            description="List of filter objects (column, operator, value). Cannot be "
+            "used together with 'search'.",
+        ),
+    ]
+    select_columns: Annotated[
+        List[str],
+        Field(
+            default_factory=list,
+            description="List of columns to select. Defaults to common columns if not "
+            "specified.",
+        ),
+    ]
+    search: Annotated[
+        str | None,
+        Field(
+            default=None,
+            description="Text search string to match against theme fields. "
+            "Cannot be used together with 'filters'.",
+        ),
+    ]
+    order_column: Annotated[
+        str | None, Field(default=None, description="Column to order results by")
+    ]
+    order_direction: Annotated[
+        Literal["asc", "desc"],
+        Field(
+            default="desc", description="Direction to order results ('asc' or 'desc')"
+        ),
+    ]
+    page: Annotated[
+        PositiveInt,
+        Field(default=1, description="Page number for pagination (1-based)"),
+    ]
+    page_size: Annotated[
+        int,
+        Field(
+            default=DEFAULT_PAGE_SIZE,
+            gt=0,
+            le=MAX_PAGE_SIZE,
+            description=f"Number of items per page (max {MAX_PAGE_SIZE})",
+        ),
+    ]
+
+    @field_validator("filters", mode="before")
+    @classmethod
+    def parse_filters(cls, v: Any) -> List[ThemeFilter]:
+        """Accept both JSON string and list of objects."""
+        return parse_json_or_model_list(v, ThemeFilter, "filters")
+
+    @field_validator("select_columns", mode="before")
+    @classmethod
+    def parse_columns(cls, v: Any) -> List[str]:
+        """Accept JSON array, list, or comma-separated string."""
+        return parse_json_or_list(v, "select_columns")
+
+    @model_validator(mode="after")
+    def validate_search_and_filters(self) -> "ListThemesRequest":
+        """Prevent using both search and filters simultaneously."""
+        if self.search and self.filters:
+            raise ValueError(
+                "Cannot use both 'search' and 'filters' parameters simultaneously. "
+                "Use either 'search' for text-based searching across multiple fields, "
+                "or 'filters' for precise column-based filtering, but not both."
+            )
+        return self
+
+
+class ThemeError(BaseModel):
+    error: str = Field(..., description="Error message")
+    error_type: str = Field(..., description="Type of error")
+    timestamp: str | datetime | None = Field(None, description="Error timestamp")
+    model_config = ConfigDict(ser_json_timedelta="iso8601")
+
+    @field_validator("error")
+    @classmethod
+    def sanitize_error_for_llm_context(cls, value: str) -> str:
+        """Wrap error text before it is exposed to LLM context."""
+        return sanitize_for_llm_context(value, field_path=("error",))
+
+    @classmethod
+    def create(cls, error: str, error_type: str) -> "ThemeError":
+        """Create a standardized ThemeError with timestamp."""
+        from datetime import datetime, timezone
+
+        return cls(
+            error=error, error_type=error_type, timestamp=datetime.now(timezone.utc)
+        )
+
+
+class GetThemeInfoRequest(MetadataCacheControl):
+    """Request schema for get_theme_info with support for ID or UUID."""
+
+    identifier: Annotated[
+        int | str,
+        Field(description="Theme identifier - can be numeric ID or UUID string"),
+    ]
+
+
+def _sanitize_theme_info_for_llm_context(
+    theme_info: ThemeInfo,
+) -> ThemeInfo:
+    """Wrap theme user-controlled fields before LLM exposure."""
+    payload = theme_info.model_dump(mode="python")
+    payload["theme_name"] = sanitize_for_llm_context(
+        payload.get("theme_name"),
+        field_path=("theme_name",),
+    )
+    if payload.get("json_data") is not None:
+        payload["json_data"] = sanitize_for_llm_context(
+            payload["json_data"],
+            field_path=("json_data",),
+        )
+    return ThemeInfo.model_validate(payload)
+
+
+def serialize_theme_object(obj: Any) -> ThemeInfo | None:
+    if not obj:
+        return None
+
+    parsed_json_data: Dict[str, Any] | None = None
+    if (raw_json_data := getattr(obj, "json_data", None)) is not None:
+        if isinstance(raw_json_data, dict):
+            parsed_json_data = raw_json_data
+        elif isinstance(raw_json_data, str):
+            try:
+                parsed_json_data = superset_json.loads(raw_json_data)
+            except (TypeError, ValueError):
+                pass
+
+    return _sanitize_theme_info_for_llm_context(
+        ThemeInfo(
+            id=getattr(obj, "id", None),
+            uuid=str(getattr(obj, "uuid", "")) if getattr(obj, "uuid", None) else None,
+            theme_name=getattr(obj, "theme_name", None),
+            json_data=parsed_json_data,
+            is_system=getattr(obj, "is_system", None),
+            is_system_default=getattr(obj, "is_system_default", None),
+            is_system_dark=getattr(obj, "is_system_dark", None),
+            changed_on=getattr(obj, "changed_on", None),
+            created_on=getattr(obj, "created_on", None),
+        )
+    )

superset/mcp_service/theme/tool/__init__.py

@@ -0,0 +1,24 @@
+# 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.
+
+from .get_theme_info import get_theme_info
+from .list_themes import list_themes
+
+__all__ = [
+    "list_themes",
+    "get_theme_info",
+]

superset/mcp_service/theme/tool/get_theme_info.py

@@ -0,0 +1,116 @@
+# 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.
+
+"""
+Get theme info FastMCP tool
+"""
+
+import logging
+from datetime import datetime, timezone
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.mcp_core import ModelGetInfoCore
+from superset.mcp_service.theme.schemas import (
+    GetThemeInfoRequest,
+    serialize_theme_object,
+    ThemeError,
+    ThemeInfo,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@tool(
+    tags=["discovery"],
+    class_permission_name="Theme",
+    annotations=ToolAnnotations(
+        title="Get theme info",
+        readOnlyHint=True,
+        destructiveHint=False,
+    ),
+)
+async def get_theme_info(
+    request: GetThemeInfoRequest, ctx: Context
+) -> ThemeInfo | ThemeError:
+    """Get theme details by ID or UUID.
+
+    Returns theme configuration including json_data.
+
+    IMPORTANT FOR LLM CLIENTS:
+    - Use numeric ID (e.g., 123) or UUID string (e.g., "a1b2c3d4-...")
+    - To find a theme ID, use the list_themes tool first
+
+    Example usage:
+    ```json
+    {
+        "identifier": 1
+    }
+    ```
+
+    Or with UUID:
+    ```json
+    {
+        "identifier": "a1b2c3d4-5678-90ab-cdef-1234567890ab"
+    }
+    ```
+    """
+    await ctx.info(
+        "Retrieving theme information: identifier=%s" % (request.identifier,)
+    )
+
+    try:
+        from superset.daos.theme import ThemeDAO
+
+        with event_logger.log_context(action="mcp.get_theme_info.lookup"):
+            get_tool = ModelGetInfoCore(
+                dao_class=ThemeDAO,
+                output_schema=ThemeInfo,
+                error_schema=ThemeError,
+                serializer=serialize_theme_object,
+                supports_slug=False,
+                logger=logger,
+            )
+
+            result = get_tool.run_tool(request.identifier)
+
+        if isinstance(result, ThemeInfo):
+            await ctx.info(
+                "Theme information retrieved successfully: "
+                "id=%s, theme_name=%s, uuid=%s"
+                % (result.id, result.theme_name, result.uuid)
+            )
+        else:
+            await ctx.warning(
+                "Theme retrieval failed: error_type=%s, error=%s"
+                % (result.error_type, result.error)
+            )
+
+        return result
+
+    except Exception as e:
+        await ctx.error(
+            "Theme information retrieval failed: identifier=%s, error=%s, "
+            "error_type=%s" % (request.identifier, str(e), type(e).__name__)
+        )
+        return ThemeError(
+            error=f"Failed to get theme info: {str(e)}",
+            error_type="InternalError",
+            timestamp=datetime.now(timezone.utc),
+        )

superset/mcp_service/theme/tool/list_themes.py

@@ -0,0 +1,147 @@
+# 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.
+
+"""
+List themes FastMCP tool
+"""
+
+import logging
+from typing import TYPE_CHECKING
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+if TYPE_CHECKING:
+    from superset.models.core import Theme
+
+from superset.extensions import event_logger
+from superset.mcp_service.mcp_core import ModelListCore
+from superset.mcp_service.theme.schemas import (
+    ListThemesRequest,
+    serialize_theme_object,
+    ThemeError,
+    ThemeFilter,
+    ThemeInfo,
+    ThemeList,
+)
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_LIST_THEMES_REQUEST = ListThemesRequest()
+
+
+@tool(
+    tags=["core"],
+    class_permission_name="Theme",
+    annotations=ToolAnnotations(
+        title="List themes",
+        readOnlyHint=True,
+        destructiveHint=False,
+    ),
+)
+async def list_themes(
+    request: ListThemesRequest | None = None,
+    ctx: Context | None = None,
+) -> ThemeList | ThemeError:
+    """List themes with filtering and search.
+
+    Returns theme metadata including name and UUID.
+
+    Sortable columns for order_column: id, theme_name, changed_on, created_on
+    """
+    if ctx is None:
+        raise RuntimeError("FastMCP context is required for list_themes")
+
+    request = request or _DEFAULT_LIST_THEMES_REQUEST.model_copy(deep=True)
+
+    await ctx.info(
+        "Listing themes: page=%s, page_size=%s, search=%s"
+        % (request.page, request.page_size, request.search)
+    )
+    await ctx.debug(
+        "Theme listing parameters: filters=%s, order_column=%s, "
+        "order_direction=%s, select_columns=%s"
+        % (
+            request.filters,
+            request.order_column,
+            request.order_direction,
+            request.select_columns,
+        )
+    )
+
+    try:
+        from superset.daos.theme import ThemeDAO
+        from superset.mcp_service.common.schema_discovery import (
+            get_all_column_names,
+            get_theme_columns,
+            THEME_DEFAULT_COLUMNS,
+            THEME_SORTABLE_COLUMNS,
+        )
+
+        all_columns = get_all_column_names(get_theme_columns())
+
+        def _serialize_theme(
+            obj: "Theme | None", cols: list[str] | None
+        ) -> ThemeInfo | None:
+            return serialize_theme_object(obj)
+
+        list_tool = ModelListCore(
+            dao_class=ThemeDAO,
+            output_schema=ThemeInfo,
+            item_serializer=_serialize_theme,
+            filter_type=ThemeFilter,
+            default_columns=THEME_DEFAULT_COLUMNS,
+            search_columns=["theme_name"],
+            list_field_name="themes",
+            output_list_schema=ThemeList,
+            all_columns=all_columns,
+            sortable_columns=THEME_SORTABLE_COLUMNS,
+            logger=logger,
+        )
+
+        with event_logger.log_context(action="mcp.list_themes.query"):
+            result = list_tool.run_tool(
+                filters=request.filters,
+                search=request.search,
+                select_columns=request.select_columns,
+                order_column=request.order_column,
+                order_direction=request.order_direction,
+                page=max(request.page - 1, 0),
+                page_size=request.page_size,
+            )
+
+        await ctx.info(
+            "Themes listed successfully: count=%s, total_count=%s"
+            % (
+                len(result.themes) if hasattr(result, "themes") else 0,
+                getattr(result, "total_count", None),
+            )
+        )
+
+        columns_to_filter = result.columns_requested
+        with event_logger.log_context(action="mcp.list_themes.serialization"):
+            return result.model_dump(
+                mode="json",
+                context={"select_columns": columns_to_filter},
+            )
+
+    except Exception as e:
+        await ctx.error(
+            "Theme listing failed: page=%s, page_size=%s, error=%s, error_type=%s"
+            % (request.page, request.page_size, str(e), type(e).__name__)
+        )
+        raise

tests/unit_tests/mcp_service/css_template/__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/css_template/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/css_template/tool/test_create_css_template.py

@@ -0,0 +1,192 @@
+# 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.
+
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from fastmcp import Client
+
+from superset.mcp_service.app import mcp
+from superset.mcp_service.css_template.schemas import CreateCssTemplateRequest
+from superset.utils import json
+
+
+@pytest.fixture
+def mcp_server():
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    """Mock authentication for all tests."""
+    with patch(
+        "superset.mcp_service.auth.get_user_from_request",
+        return_value=Mock(is_authenticated=True),
+    ):
+        yield
+
+
+# ---------------------------------------------------------------------------
+# Schema tests
+# ---------------------------------------------------------------------------
+
+
+def test_create_css_template_request_valid() -> None:
+    req = CreateCssTemplateRequest(
+        template_name="My Theme",
+        css=".header { color: red; }",
+    )
+    assert req.template_name == "My Theme"
+    assert req.css == ".header { color: red; }"
+
+
+def test_create_css_template_request_strips_name_whitespace() -> None:
+    req = CreateCssTemplateRequest(
+        template_name="  My Theme  ",
+        css=".header { color: red; }",
+    )
+    assert req.template_name == "My Theme"
+
+
+def test_create_css_template_request_empty_name_fails() -> None:
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError, match="template_name must not be empty"):
+        CreateCssTemplateRequest(template_name="   ", css=".header { color: red; }")
+
+
+def test_create_css_template_request_name_too_long() -> None:
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        CreateCssTemplateRequest(template_name="a" * 251, css="")
+
+
+# ---------------------------------------------------------------------------
+# Tool logic tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_create_css_template_success(mcp_server: object) -> None:
+    """Happy path: template created, id and fields returned."""
+    mock_template = MagicMock()
+    mock_template.id = 7
+    mock_template.template_name = "Dark Theme"
+    mock_template.css = "body { background: #000; }"
+
+    mock_command = MagicMock()
+    mock_command.run.return_value = mock_template
+
+    with patch(
+        "superset.commands.css.create.CreateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = CreateCssTemplateRequest(
+                template_name="Dark Theme",
+                css="body { background: #000; }",
+            )
+            result = await client.call_tool(
+                "create_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] == 7
+    assert data["template_name"] == "Dark Theme"
+    assert data["css"] == "body { background: #000; }"
+    assert data["error"] is None
+
+
+@pytest.mark.asyncio
+async def test_create_css_template_invalid_error(mcp_server: object) -> None:
+    """CssTemplateInvalidError is caught and returned as an error response."""
+    from superset.commands.css.exceptions import CssTemplateInvalidError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = CssTemplateInvalidError()
+
+    with patch(
+        "superset.commands.css.create.CreateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = CreateCssTemplateRequest(
+                template_name="Bad Template",
+                css="",
+            )
+            result = await client.call_tool(
+                "create_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert data["error"] is not None
+
+
+@pytest.mark.asyncio
+async def test_create_css_template_create_failed(mcp_server: object) -> None:
+    """CssTemplateCreateFailedError is caught and returned as an error response."""
+    from superset.commands.css.exceptions import CssTemplateCreateFailedError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = CssTemplateCreateFailedError()
+
+    with patch(
+        "superset.commands.css.create.CreateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = CreateCssTemplateRequest(
+                template_name="Failing Template",
+                css=".x { color: blue; }",
+            )
+            result = await client.call_tool(
+                "create_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert data["error"] is not None
+    assert "Failed to create CSS template" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_create_css_template_unexpected_exception_is_reraised(
+    mcp_server: object,
+) -> None:
+    """Unexpected exceptions are re-raised (not swallowed as error responses)."""
+    from fastmcp.exceptions import ToolError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = RuntimeError("unexpected database error")
+
+    with patch(
+        "superset.commands.css.create.CreateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            with pytest.raises((RuntimeError, ToolError)):
+                await client.call_tool(
+                    "create_css_template",
+                    {
+                        "request": {
+                            "template_name": "Theme",
+                            "css": ".x { color: red; }",
+                        }
+                    },
+                )

tests/unit_tests/mcp_service/css_template/tool/test_css_template_tools.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.
+
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastmcp import Client
+from fastmcp.exceptions import ToolError
+from pydantic import ValidationError
+
+from superset.mcp_service.app import mcp
+from superset.mcp_service.css_template.schemas import (
+    CssTemplateFilter,
+    ListCssTemplatesRequest,
+)
+from superset.mcp_service.utils.sanitization import (
+    LLM_CONTEXT_CLOSE_DELIMITER,
+    LLM_CONTEXT_OPEN_DELIMITER,
+)
+from superset.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+def _wrapped(value: str) -> str:
+    return f"{LLM_CONTEXT_OPEN_DELIMITER}\n{value}\n{LLM_CONTEXT_CLOSE_DELIMITER}"
+
+
+class TestCssTemplateFilterSchema:
+    """Tests for CssTemplateFilter schema — filterable columns."""
+
+    def test_invalid_filter_column_rejected(self):
+        """Columns not in the Literal set must be rejected."""
+        with pytest.raises(ValidationError):
+            CssTemplateFilter(col="not_a_real_column", opr="eq", value="test")
+
+    def test_valid_filter_column_accepted(self):
+        """template_name is a valid filter column."""
+        f = CssTemplateFilter(col="template_name", opr="eq", value="my_template")
+        assert f.col == "template_name"
+
+    def test_css_column_not_filterable(self):
+        """css is not a public filter column (large field)."""
+        with pytest.raises(ValidationError):
+            CssTemplateFilter(col="css", opr="eq", value="body {}")
+
+    def test_created_by_fk_filter_accepted(self):
+        """created_by_fk is a valid filter column for filtering by creator."""
+        f = CssTemplateFilter(col="created_by_fk", opr="eq", value=1)
+        assert f.col == "created_by_fk"
+
+
+def create_mock_css_template(
+    template_id: int = 1,
+    template_name: str = "my_template",
+    css: str = "body { color: red; }",
+    uuid: str | None = None,
+    created_by_name: str | None = "admin",
+    changed_by_name: str | None = "admin",
+) -> MagicMock:
+    """Factory function to create mock CSS template objects."""
+    template = MagicMock()
+    template.id = template_id
+    template.template_name = template_name
+    template.css = css
+    template.uuid = (
+        uuid if uuid is not None else f"test-css-template-uuid-{template_id}"
+    )
+    template.changed_on = None
+    template.created_on = None
+    template.created_by_name = created_by_name
+    template.changed_by_name = changed_by_name
+    return template
+
+
+@pytest.fixture
+def mcp_server():
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    """Mock authentication for all tests."""
+    from unittest.mock import Mock, patch
+
+    with patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user:
+        mock_user = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+@patch("superset.daos.css.CssTemplateDAO.list")
+@pytest.mark.asyncio
+async def test_list_css_templates_basic(mock_list, mcp_server):
+    """Test basic CSS template listing functionality."""
+    template = create_mock_css_template()
+    mock_list.return_value = ([template], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListCssTemplatesRequest(page=1, page_size=10)
+        result = await client.call_tool(
+            "list_css_templates", {"request": request.model_dump()}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["css_templates"] is not None
+        assert len(data["css_templates"]) == 1
+        assert data["css_templates"][0]["id"] == 1
+        assert data["css_templates"][0]["template_name"] == _wrapped("my_template")
+
+
+@patch("superset.daos.css.CssTemplateDAO.list")
+@pytest.mark.asyncio
+async def test_list_css_templates_css_not_in_default_columns(mock_list, mcp_server):
+    """Test that css field is excluded from default columns (it's large)."""
+    template = create_mock_css_template()
+    mock_list.return_value = ([template], 1)
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("list_css_templates", {})
+        data = json.loads(result.content[0].text)
+
+    assert data["css_templates"] is not None
+    assert len(data["css_templates"]) == 1
+    # css not in default columns
+    assert "css" not in data["css_templates"][0]
+    assert data["columns_requested"] == ["id", "uuid", "template_name"]
+    assert data["css_templates"][0]["uuid"] == "test-css-template-uuid-1"
+
+
+@patch("superset.daos.css.CssTemplateDAO.list")
+@pytest.mark.asyncio
+async def test_list_css_templates_with_search(mock_list, mcp_server):
+    """Test CSS template listing with search functionality."""
+    template = create_mock_css_template(template_name="dark_theme_css")
+    mock_list.return_value = ([template], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListCssTemplatesRequest(page=1, page_size=10, search="dark")
+        result = await client.call_tool(
+            "list_css_templates", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        assert len(data["css_templates"]) == 1
+        assert data["css_templates"][0]["template_name"] == _wrapped("dark_theme_css")
+
+
+@patch("superset.daos.css.CssTemplateDAO.list")
+@pytest.mark.asyncio
+async def test_list_css_templates_with_select_columns_css(mock_list, mcp_server):
+    """Test that css can be requested via select_columns."""
+    template = create_mock_css_template(css="body { margin: 0; }")
+    mock_list.return_value = ([template], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListCssTemplatesRequest(
+            page=1, page_size=10, select_columns=["id", "template_name", "css"]
+        )
+        result = await client.call_tool(
+            "list_css_templates", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        assert data["css_templates"][0]["css"] == _wrapped("body { margin: 0; }")
+
+
+@patch("superset.daos.css.CssTemplateDAO.list")
+@pytest.mark.asyncio
+async def test_list_css_templates_with_filters(mock_list, mcp_server):
+    """Test CSS template listing with filters."""
+    template = create_mock_css_template(template_name="bootstrap_css")
+    mock_list.return_value = ([template], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListCssTemplatesRequest(
+            page=1,
+            page_size=10,
+            filters=[{"col": "template_name", "opr": "eq", "value": "bootstrap_css"}],
+        )
+        result = await client.call_tool(
+            "list_css_templates", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        assert len(data["css_templates"]) == 1
+
+
+@patch("superset.daos.css.CssTemplateDAO.list")
+@pytest.mark.asyncio
+async def test_list_css_templates_api_error(mock_list, mcp_server):
+    """Test error handling when DAO raises an exception."""
+    mock_list.side_effect = ToolError("CSS template error")
+
+    async with Client(mcp_server) as client:
+        request = ListCssTemplatesRequest(page=1, page_size=10)
+        with pytest.raises(ToolError) as excinfo:  # noqa: PT012
+            await client.call_tool(
+                "list_css_templates", {"request": request.model_dump()}
+            )
+        assert "CSS template error" in str(excinfo.value)
+
+
+@patch("superset.daos.css.CssTemplateDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_css_template_info_basic(mock_find, mcp_server):
+    """Test basic get CSS template info functionality."""
+    template = create_mock_css_template()
+    mock_find.return_value = template
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_css_template_info", {"request": {"identifier": 1}}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["id"] == 1
+        assert data["template_name"] == _wrapped("my_template")
+        assert data["uuid"] == "test-css-template-uuid-1"
+        assert data["css"] == _wrapped("body { color: red; }")
+        assert data["created_by_name"] == _wrapped("admin")
+        assert data["changed_by_name"] == _wrapped("admin")
+
+
+@patch("superset.daos.css.CssTemplateDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_css_template_info_by_uuid(mock_find, mcp_server):
+    """Test get CSS template info by UUID."""
+    template = create_mock_css_template(uuid="a1b2c3d4-5678-90ab-cdef-1234567890ab")
+    mock_find.return_value = template
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_css_template_info",
+            {"request": {"identifier": "a1b2c3d4-5678-90ab-cdef-1234567890ab"}},
+        )
+        data = json.loads(result.content[0].text)
+        assert data["id"] == 1
+        assert data["uuid"] == "a1b2c3d4-5678-90ab-cdef-1234567890ab"
+
+    mock_find.assert_called_once_with(
+        "a1b2c3d4-5678-90ab-cdef-1234567890ab", id_column="uuid", query_options=None
+    )
+
+
+@patch("superset.daos.css.CssTemplateDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_css_template_info_not_found(mock_find, mcp_server):
+    """Test get CSS template info when template does not exist."""
+    mock_find.return_value = None
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_css_template_info", {"request": {"identifier": 999}}
+        )
+        assert json.loads(result.content[0].text)["error_type"] == "not_found"
+
+
+def test_list_css_templates_request_search_and_filters_conflict():
+    """Cannot use search and filters simultaneously."""
+    with pytest.raises(ValidationError):
+        ListCssTemplatesRequest(
+            search="something",
+            filters=[{"col": "template_name", "opr": "eq", "value": "test"}],
+        )

tests/unit_tests/mcp_service/css_template/tool/test_update_css_template.py

@@ -0,0 +1,219 @@
+# 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.
+
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from fastmcp import Client
+
+from superset.mcp_service.app import mcp
+from superset.mcp_service.css_template.schemas import UpdateCssTemplateRequest
+from superset.utils import json
+
+
+@pytest.fixture
+def mcp_server():
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    """Mock authentication for all tests."""
+    with patch(
+        "superset.mcp_service.auth.get_user_from_request",
+        return_value=Mock(is_authenticated=True),
+    ):
+        yield
+
+
+# ---------------------------------------------------------------------------
+# Schema tests
+# ---------------------------------------------------------------------------
+
+
+def test_update_css_template_request_valid_name_only() -> None:
+    req = UpdateCssTemplateRequest(id=1, template_name="New Name")
+    assert req.id == 1
+    assert req.template_name == "New Name"
+    assert req.css is None
+
+
+def test_update_css_template_request_valid_css_only() -> None:
+    req = UpdateCssTemplateRequest(id=5, css=".body { color: blue; }")
+    assert req.id == 5
+    assert req.css == ".body { color: blue; }"
+    assert req.template_name is None
+
+
+def test_update_css_template_request_strips_name_whitespace() -> None:
+    req = UpdateCssTemplateRequest(id=1, template_name="  Padded  ")
+    assert req.template_name == "Padded"
+
+
+def test_update_css_template_request_empty_name_fails() -> None:
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError, match="template_name must not be empty"):
+        UpdateCssTemplateRequest(id=1, template_name="   ")
+
+
+def test_update_css_template_request_name_too_long() -> None:
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        UpdateCssTemplateRequest(id=1, template_name="x" * 251)
+
+
+# ---------------------------------------------------------------------------
+# Tool logic tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_update_css_template_success(mcp_server: object) -> None:
+    """Happy path: template updated, all fields returned."""
+    mock_template = MagicMock()
+    mock_template.id = 3
+    mock_template.template_name = "Updated Theme"
+    mock_template.css = "body { background: #fff; }"
+
+    mock_command = MagicMock()
+    mock_command.run.return_value = mock_template
+
+    with patch(
+        "superset.commands.css.update.UpdateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = UpdateCssTemplateRequest(
+                id=3,
+                template_name="Updated Theme",
+                css="body { background: #fff; }",
+            )
+            result = await client.call_tool(
+                "update_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] == 3
+    assert data["template_name"] == "Updated Theme"
+    assert data["css"] == "body { background: #fff; }"
+    assert data["error"] is None
+
+
+@pytest.mark.asyncio
+async def test_update_css_template_no_fields_returns_error(
+    mcp_server: object,
+) -> None:
+    """Calling with neither template_name nor css returns a structured error."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("update_css_template", {"request": {"id": 1}})
+        data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert "At least one" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_update_css_template_not_found(mcp_server: object) -> None:
+    """CssTemplateNotFoundError is caught and returned as an error response."""
+    from superset.commands.css.exceptions import CssTemplateNotFoundError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = CssTemplateNotFoundError()
+
+    with patch(
+        "superset.commands.css.update.UpdateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = UpdateCssTemplateRequest(id=999, template_name="Ghost")
+            result = await client.call_tool(
+                "update_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert "not found" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_update_css_template_invalid_error(mcp_server: object) -> None:
+    """CssTemplateInvalidError is caught and returned as an error response."""
+    from superset.commands.css.exceptions import CssTemplateInvalidError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = CssTemplateInvalidError()
+
+    with patch(
+        "superset.commands.css.update.UpdateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = UpdateCssTemplateRequest(id=1, template_name="Valid Name")
+            result = await client.call_tool(
+                "update_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert data["error"] is not None
+
+
+@pytest.mark.asyncio
+async def test_update_css_template_update_failed(mcp_server: object) -> None:
+    """CssTemplateUpdateFailedError is caught and returned as an error response."""
+    from superset.commands.css.exceptions import CssTemplateUpdateFailedError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = CssTemplateUpdateFailedError()
+
+    with patch(
+        "superset.commands.css.update.UpdateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            request = UpdateCssTemplateRequest(id=1, css=".x { color: red; }")
+            result = await client.call_tool(
+                "update_css_template", {"request": request.model_dump()}
+            )
+            data = json.loads(result.content[0].text)
+
+    assert data["id"] is None
+    assert "Failed to update CSS template" in data["error"]
+
+
+@pytest.mark.asyncio
+async def test_update_css_template_unexpected_exception_is_reraised(
+    mcp_server: object,
+) -> None:
+    """Unexpected exceptions are re-raised (not swallowed as error responses)."""
+    from fastmcp.exceptions import ToolError
+
+    mock_command = MagicMock()
+    mock_command.run.side_effect = RuntimeError("unexpected database error")
+
+    with patch(
+        "superset.commands.css.update.UpdateCssTemplateCommand",
+        return_value=mock_command,
+    ):
+        async with Client(mcp_server) as client:
+            with pytest.raises((RuntimeError, ToolError)):
+                await client.call_tool(
+                    "update_css_template",
+                    {"request": {"id": 1, "css": ".x { color: red; }"}},
+                )

tests/unit_tests/mcp_service/system/tool/test_get_schema.py

@@ -30,6 +30,9 @@ from superset.mcp_service.common.schema_discovery import (
     CHART_DEFAULT_COLUMNS,
     CHART_SEARCH_COLUMNS,
     CHART_SORTABLE_COLUMNS,
+    CSS_TEMPLATE_DEFAULT_COLUMNS,
+    CSS_TEMPLATE_SEARCH_COLUMNS,
+    CSS_TEMPLATE_SORTABLE_COLUMNS,
     DASHBOARD_DEFAULT_COLUMNS,
     DASHBOARD_SEARCH_COLUMNS,
     DASHBOARD_SORTABLE_COLUMNS,
@@ -38,6 +41,9 @@ from superset.mcp_service.common.schema_discovery import (
     DATASET_SORTABLE_COLUMNS,
     GetSchemaRequest,
     ModelSchemaInfo,
+    THEME_DEFAULT_COLUMNS,
+    THEME_SEARCH_COLUMNS,
+    THEME_SORTABLE_COLUMNS,
 )
 from superset.utils import json
 
@@ -545,3 +551,154 @@ class TestSchemaDiscoveryConstants:
         assert "slice_name" in CHART_SEARCH_COLUMNS
         assert "table_name" in DATASET_SEARCH_COLUMNS
         assert "dashboard_title" in DASHBOARD_SEARCH_COLUMNS
+
+    def test_css_template_default_columns(self):
+        """Test CSS template default columns include id and uuid but not css."""
+        assert "id" in CSS_TEMPLATE_DEFAULT_COLUMNS
+        assert "uuid" in CSS_TEMPLATE_DEFAULT_COLUMNS
+        assert "template_name" in CSS_TEMPLATE_DEFAULT_COLUMNS
+        assert "css" not in CSS_TEMPLATE_DEFAULT_COLUMNS
+
+    def test_css_template_sortable_columns(self):
+        """Test CSS template sortable columns are defined correctly."""
+        assert "id" in CSS_TEMPLATE_SORTABLE_COLUMNS
+        assert "template_name" in CSS_TEMPLATE_SORTABLE_COLUMNS
+        assert "changed_on" in CSS_TEMPLATE_SORTABLE_COLUMNS
+        assert "created_on" in CSS_TEMPLATE_SORTABLE_COLUMNS
+
+    def test_theme_default_columns(self):
+        """Test theme default columns include uuid."""
+        assert "id" in THEME_DEFAULT_COLUMNS
+        assert "uuid" in THEME_DEFAULT_COLUMNS
+        assert "theme_name" in THEME_DEFAULT_COLUMNS
+        assert "json_data" not in THEME_DEFAULT_COLUMNS
+
+    def test_theme_sortable_columns(self):
+        """Test theme sortable columns are defined correctly."""
+        assert "id" in THEME_SORTABLE_COLUMNS
+        assert "theme_name" in THEME_SORTABLE_COLUMNS
+        assert "changed_on" in THEME_SORTABLE_COLUMNS
+        assert "created_on" in THEME_SORTABLE_COLUMNS
+
+    def test_css_template_search_columns_defined(self):
+        """Test CSS template search columns are defined."""
+        assert "template_name" in CSS_TEMPLATE_SEARCH_COLUMNS
+
+    def test_theme_search_columns_defined(self):
+        """Test theme search columns are defined."""
+        assert "theme_name" in THEME_SEARCH_COLUMNS
+
+
+class TestGetSchemaCssTemplateAndTheme:
+    """Test get_schema tool for css_template and theme model types."""
+
+    @pytest.mark.asyncio
+    async def test_get_schema_css_template(self, mcp_server):
+        """Test get_schema for css_template model type."""
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_schema", {"request": {"model_type": "css_template"}}
+            )
+
+            assert result.content is not None
+            data = json.loads(result.content[0].text)
+
+            assert "schema_info" in data
+            info = data["schema_info"]
+            assert info["model_type"] == "css_template"
+            assert "select_columns" in info
+            assert "filter_columns" in info
+            assert "sortable_columns" in info
+            assert "default_select" in info
+            assert "search_columns" in info
+
+            # uuid and template_name are in defaults; css is not
+            assert "id" in info["default_select"]
+            assert "uuid" in info["default_select"]
+            assert "template_name" in info["default_select"]
+            assert "css" not in info["default_select"]
+
+            # template_name is searchable
+            assert "template_name" in info["search_columns"]
+
+            # sortable columns
+            assert "template_name" in info["sortable_columns"]
+            assert "changed_on" in info["sortable_columns"]
+
+            # filter_columns matches list_css_templates accepted filters
+            assert "template_name" in info["filter_columns"]
+            assert "created_by_fk" in info["filter_columns"]
+            assert len(info["filter_columns"]) == 2, (
+                "filter_columns must match list_css_templates accepted filters"
+            )
+
+    @pytest.mark.asyncio
+    async def test_get_schema_theme(self, mcp_server):
+        """Test get_schema for theme model type."""
+        async with Client(mcp_server) as client:
+            result = await client.call_tool(
+                "get_schema", {"request": {"model_type": "theme"}}
+            )
+
+            assert result.content is not None
+            data = json.loads(result.content[0].text)
+
+            assert "schema_info" in data
+            info = data["schema_info"]
+            assert info["model_type"] == "theme"
+            assert "select_columns" in info
+            assert "filter_columns" in info
+
+            # uuid and theme_name are in defaults; json_data is not
+            assert "id" in info["default_select"]
+            assert "uuid" in info["default_select"]
+            assert "theme_name" in info["default_select"]
+            assert "json_data" not in info["default_select"]
+
+            # theme_name is searchable
+            assert "theme_name" in info["search_columns"]
+
+            # sortable columns
+            assert "theme_name" in info["sortable_columns"]
+            assert "changed_on" in info["sortable_columns"]
+
+            # filter_columns matches list_themes accepted filters
+            assert "theme_name" in info["filter_columns"]
+            assert "is_system" in info["filter_columns"]
+            assert "is_system_default" in info["filter_columns"]
+            assert "is_system_dark" in info["filter_columns"]
+            assert "created_by_fk" in info["filter_columns"]
+            assert len(info["filter_columns"]) == 5, (
+                "get_schema filter_columns must match list_themes accepted filters"
+            )
+
+    def test_css_template_model_type_accepted(self):
+        """css_template is a valid GetSchemaRequest model_type."""
+        request = GetSchemaRequest(model_type="css_template")
+        assert request.model_type == "css_template"
+
+    def test_theme_model_type_accepted(self):
+        """theme is a valid GetSchemaRequest model_type."""
+        request = GetSchemaRequest(model_type="theme")
+        assert request.model_type == "theme"
+
+
+class TestGetSchemaPermissionMap:
+    """Verify that _MODEL_TYPE_CLASS_PERMISSION covers css_template and theme."""
+
+    def test_css_template_permission_entry(self):
+        """css_template must map to CssTemplate FAB class."""
+        assert (
+            get_schema_module._MODEL_TYPE_CLASS_PERMISSION["css_template"]
+            == "CssTemplate"
+        )
+
+    def test_theme_permission_entry(self):
+        """theme must map to Theme FAB class."""
+        assert get_schema_module._MODEL_TYPE_CLASS_PERMISSION["theme"] == "Theme"
+
+    def test_all_schema_factory_types_covered(self):
+        """Every key in _SCHEMA_CORE_FACTORIES must have a permission entry."""
+        factories = set(get_schema_module._SCHEMA_CORE_FACTORIES.keys())
+        perms = set(get_schema_module._MODEL_TYPE_CLASS_PERMISSION.keys())
+        assert factories == perms

tests/unit_tests/mcp_service/theme/__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/theme/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/theme/tool/test_theme_tools.py

@@ -0,0 +1,267 @@
+# 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.
+
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastmcp import Client
+from fastmcp.exceptions import ToolError
+from pydantic import ValidationError
+
+from superset.mcp_service.app import mcp
+from superset.mcp_service.theme.schemas import (
+    ListThemesRequest,
+    ThemeFilter,
+)
+from superset.mcp_service.utils.sanitization import (
+    LLM_CONTEXT_CLOSE_DELIMITER,
+    LLM_CONTEXT_OPEN_DELIMITER,
+)
+from superset.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+def _wrapped(value: str) -> str:
+    return f"{LLM_CONTEXT_OPEN_DELIMITER}\n{value}\n{LLM_CONTEXT_CLOSE_DELIMITER}"
+
+
+class TestThemeFilterSchema:
+    """Tests for ThemeFilter schema — filterable columns."""
+
+    def test_invalid_filter_column_rejected(self):
+        """Columns not in the Literal set must be rejected."""
+        with pytest.raises(ValidationError):
+            ThemeFilter(col="not_a_real_column", opr="eq", value="test")
+
+    def test_valid_filter_column_accepted(self):
+        """theme_name is a valid filter column."""
+        f = ThemeFilter(col="theme_name", opr="eq", value="my_theme")
+        assert f.col == "theme_name"
+
+    def test_json_data_column_not_filterable(self):
+        """json_data is not a public filter column."""
+        with pytest.raises(ValidationError):
+            ThemeFilter(col="json_data", opr="eq", value="{}")
+
+    def test_is_system_filter_accepted(self):
+        """is_system is a valid filter column for distinguishing system themes."""
+        f = ThemeFilter(col="is_system", opr="eq", value=True)
+        assert f.col == "is_system"
+
+    def test_is_system_default_filter_accepted(self):
+        """is_system_default is a valid filter column."""
+        f = ThemeFilter(col="is_system_default", opr="eq", value=True)
+        assert f.col == "is_system_default"
+
+    def test_is_system_dark_filter_accepted(self):
+        """is_system_dark is a valid filter column."""
+        f = ThemeFilter(col="is_system_dark", opr="eq", value=False)
+        assert f.col == "is_system_dark"
+
+    def test_created_by_fk_filter_accepted(self):
+        """created_by_fk is a valid filter column for filtering by creator."""
+        f = ThemeFilter(col="created_by_fk", opr="eq", value=1)
+        assert f.col == "created_by_fk"
+
+
+def create_mock_theme(
+    theme_id: int = 1,
+    theme_name: str = "light_theme",
+    json_data: str = '{"primaryColor": "#1890ff"}',
+    uuid: str = "test-theme-uuid-1",
+) -> MagicMock:
+    """Factory function to create mock theme objects."""
+    theme = MagicMock()
+    theme.id = theme_id
+    theme.theme_name = theme_name
+    theme.json_data = json_data
+    theme.uuid = uuid
+    theme.is_system = False
+    theme.is_system_default = False
+    theme.is_system_dark = False
+    theme.changed_on = None
+    theme.created_on = None
+    return theme
+
+
+@pytest.fixture
+def mcp_server():
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    """Mock authentication for all tests."""
+    from unittest.mock import Mock, patch
+
+    with patch("superset.mcp_service.auth.get_user_from_request") as mock_get_user:
+        mock_user = Mock()
+        mock_user.id = 1
+        mock_user.username = "admin"
+        mock_get_user.return_value = mock_user
+        yield mock_get_user
+
+
+@patch("superset.daos.theme.ThemeDAO.list")
+@pytest.mark.asyncio
+async def test_list_themes_basic(mock_list, mcp_server):
+    """Test basic theme listing functionality."""
+    theme = create_mock_theme()
+    mock_list.return_value = ([theme], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListThemesRequest(page=1, page_size=10)
+        result = await client.call_tool(
+            "list_themes", {"request": request.model_dump()}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["themes"] is not None
+        assert len(data["themes"]) == 1
+        assert data["themes"][0]["id"] == 1
+        assert data["themes"][0]["theme_name"] == _wrapped("light_theme")
+        assert data["themes"][0]["uuid"] == "test-theme-uuid-1"
+
+
+@patch("superset.daos.theme.ThemeDAO.list")
+@pytest.mark.asyncio
+async def test_list_themes_default_columns_include_uuid(mock_list, mcp_server):
+    """Test that uuid is included in default columns for themes."""
+    theme = create_mock_theme()
+    mock_list.return_value = ([theme], 1)
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("list_themes", {})
+        data = json.loads(result.content[0].text)
+
+    assert data["columns_requested"] == ["id", "theme_name", "uuid"]
+    assert data["themes"][0]["uuid"] == "test-theme-uuid-1"
+
+
+@patch("superset.daos.theme.ThemeDAO.list")
+@pytest.mark.asyncio
+async def test_list_themes_with_search(mock_list, mcp_server):
+    """Test theme listing with search functionality."""
+    theme = create_mock_theme(theme_name="dark_theme")
+    mock_list.return_value = ([theme], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListThemesRequest(page=1, page_size=10, search="dark")
+        result = await client.call_tool(
+            "list_themes", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        assert len(data["themes"]) == 1
+        assert data["themes"][0]["theme_name"] == _wrapped("dark_theme")
+
+
+@patch("superset.daos.theme.ThemeDAO.list")
+@pytest.mark.asyncio
+async def test_list_themes_with_filters(mock_list, mcp_server):
+    """Test theme listing with filters."""
+    theme = create_mock_theme(theme_name="custom_theme")
+    mock_list.return_value = ([theme], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListThemesRequest(
+            page=1,
+            page_size=10,
+            filters=[{"col": "theme_name", "opr": "eq", "value": "custom_theme"}],
+        )
+        result = await client.call_tool(
+            "list_themes", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        assert len(data["themes"]) == 1
+
+
+@patch("superset.daos.theme.ThemeDAO.list")
+@pytest.mark.asyncio
+async def test_list_themes_api_error(mock_list, mcp_server):
+    """Test error handling when DAO raises an exception."""
+    mock_list.side_effect = ToolError("Theme error")
+
+    async with Client(mcp_server) as client:
+        request = ListThemesRequest(page=1, page_size=10)
+        with pytest.raises(ToolError) as excinfo:  # noqa: PT012
+            await client.call_tool("list_themes", {"request": request.model_dump()})
+        assert "Theme error" in str(excinfo.value)
+
+
+@patch("superset.daos.theme.ThemeDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_theme_info_basic(mock_find, mcp_server):
+    """Test basic get theme info functionality."""
+    theme = create_mock_theme()
+    mock_find.return_value = theme
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_theme_info", {"request": {"identifier": 1}}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["id"] == 1
+        assert data["theme_name"] == _wrapped("light_theme")
+        assert data["uuid"] == "test-theme-uuid-1"
+        assert data["json_data"]["primaryColor"] == _wrapped("#1890ff")
+
+
+@patch("superset.daos.theme.ThemeDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_theme_info_by_uuid(mock_find, mcp_server):
+    """Test get theme info by UUID."""
+    theme = create_mock_theme(uuid="a1b2c3d4-5678-90ab-cdef-1234567890ab")
+    mock_find.return_value = theme
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_theme_info",
+            {"request": {"identifier": "a1b2c3d4-5678-90ab-cdef-1234567890ab"}},
+        )
+        data = json.loads(result.content[0].text)
+        assert data["id"] == 1
+        assert data["uuid"] == "a1b2c3d4-5678-90ab-cdef-1234567890ab"
+
+    mock_find.assert_called_once_with(
+        "a1b2c3d4-5678-90ab-cdef-1234567890ab", id_column="uuid", query_options=None
+    )
+
+
+@patch("superset.daos.theme.ThemeDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_theme_info_not_found(mock_find, mcp_server):
+    """Test get theme info when theme does not exist."""
+    mock_find.return_value = None
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_theme_info", {"request": {"identifier": 999}}
+        )
+        assert json.loads(result.content[0].text)["error_type"] == "not_found"
+
+
+def test_list_themes_request_search_and_filters_conflict():
+    """Cannot use search and filters simultaneously."""
+    with pytest.raises(ValidationError):
+        ListThemesRequest(
+            search="something",
+            filters=[{"col": "theme_name", "opr": "eq", "value": "test"}],
+        )