fix(docker): replace pip install uv with COPY from official uv image

pip install --upgrade uv downloads uv as a source distribution when no
pre-built wheel matches the platform (python:3.11-slim-trixie), then
fails to compile because cc is absent in the slim image. Using
COPY --from=ghcr.io/astral-sh/uv:latest ships a statically-linked
binary that works on any Linux regardless of glibc version, which is
the approach uv's own documentation recommends for Dockerfiles.

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

superset/mcp_service/app.py

@@ -123,6 +123,14 @@ 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)
+
+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)
+
 Dataset Management:
 - list_datasets: List datasets with advanced filters (1-based pagination)
 - get_dataset_info: Get detailed dataset information by ID (includes columns/metrics)
@@ -146,7 +154,7 @@ SQL Lab Integration:
 - open_sql_lab_with_context: Generate SQL Lab URL with pre-filled sql
 
 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)
 
 System Information:
 - get_instance_info: Get instance-wide statistics, metadata, and current user identity
@@ -617,6 +625,10 @@ 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
+    get_css_template_info,
+    list_css_templates,
+)
 from superset.mcp_service.dashboard.tool import (  # noqa: F401, E402
     add_chart_to_existing_dashboard,
     generate_dashboard,
@@ -652,6 +664,10 @@ from superset.mcp_service.system.tool import (  # noqa: F401, E402
     get_schema,
     health_check,
 )
+from superset.mcp_service.theme.tool import (  # noqa: F401, E402
+    get_theme_info,
+    list_themes,
+)
 
 
 def _remove_disabled_tools(disabled_tools: set[str]) -> None:

superset/mcp_service/common/schema_discovery.py

@@ -635,3 +635,55 @@ 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"]
+
+
+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,
+        exclude_columns=set(USER_DIRECTORY_FIELDS),
+    )
+
+
+# Theme configuration
+THEME_DEFAULT_COLUMNS = [
+    "id",
+    "theme_name",
+    "uuid",
+]
+THEME_SORTABLE_COLUMNS = [
+    "id",
+    "theme_name",
+    "changed_on",
+    "created_on",
+]
+THEME_SEARCH_COLUMNS = ["theme_name"]
+
+
+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,249 @@
+# 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.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"] = Field(
+        ...,
+        description="Column to filter on.",
+    )
+    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")
+
+    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")
+
+    @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 serialize_css_template_object(obj: Any) -> CssTemplateInfo | None:
+    if not obj:
+        return None
+
+    return 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),
+    )

superset/mcp_service/css_template/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_css_template_info import get_css_template_info
+from .list_css_templates import list_css_templates
+
+__all__ = [
+    "list_css_templates",
+    "get_css_template_info",
+]

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/system/tool/get_schema.py

@@ -34,6 +34,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,
@@ -44,12 +47,17 @@ 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_SEARCH_COLUMNS,
+    THEME_SORTABLE_COLUMNS,
 )
 from superset.mcp_service.constants import ModelType
 from superset.mcp_service.mcp_core import ModelGetSchemaCore
@@ -144,6 +152,42 @@ 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",
+        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",
+        logger=logger,
+    )
+
+
 # Map model types to their core factory functions
 _SCHEMA_CORE_FACTORIES: dict[
     ModelType,
@@ -153,6 +197,8 @@ _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,
 }
 
 
@@ -182,7 +228,8 @@ 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

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,259 @@
+# 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.schema_utils import (
+    parse_json_or_list,
+    parse_json_or_model_list,
+)
+
+
+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"] = Field(
+        ...,
+        description="Column to filter on.",
+    )
+    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: str | None = Field(
+        None,
+        description="Theme JSON configuration data",
+    )
+    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")
+
+    @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 serialize_theme_object(obj: Any) -> ThemeInfo | None:
+    if not obj:
+        return None
+
+    return 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=getattr(obj, "json_data", None),
+        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_css_template_tools.py

@@ -0,0 +1,237 @@
+# 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.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+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 create_mock_css_template(
+    template_id: int = 1,
+    template_name: str = "my_template",
+    css: str = "body { color: red; }",
+) -> 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 = f"test-css-template-uuid-{template_id}"
+    template.changed_on = None
+    template.created_on = None
+    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"] == "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"] == "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"] == "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"] == "my_template"
+        assert data["uuid"] == "test-css-template-uuid-1"
+        assert data["css"] == "body { color: red; }"
+
+
+@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 result.data["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/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,126 @@ 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."""
+
+    @patch("superset.daos.css.CssTemplateDAO.get_filterable_columns_and_operators")
+    @pytest.mark.asyncio
+    async def test_get_schema_css_template(self, mock_filters, mcp_server):
+        """Test get_schema for css_template model type."""
+        mock_filters.return_value = {
+            "template_name": ["eq", "sw", "ilike"],
+        }
+
+        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"]
+
+    @patch("superset.daos.theme.ThemeDAO.get_filterable_columns_and_operators")
+    @pytest.mark.asyncio
+    async def test_get_schema_theme(self, mock_filters, mcp_server):
+        """Test get_schema for theme model type."""
+        mock_filters.return_value = {
+            "theme_name": ["eq", "sw", "ilike"],
+        }
+
+        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"]
+
+    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"

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,235 @@
+# 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.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+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 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"] == "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"] == "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"] == "light_theme"
+        assert data["uuid"] == "test-theme-uuid-1"
+        assert data["json_data"] == '{"primaryColor": "#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"
+
+
+@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 result.data["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"}],
+        )