feat(mcp): add list and get tools for CSS templates and themes

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 @@ def get_default_instructions(
 - 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 @@ def get_default_instructions(
 - 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 @@ def create_mcp_app(
     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 @@ def create_mcp_app(
     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 @@ def get_all_column_names(columns: list[ColumnMetadata]) -> 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),
+    )