fix(mcp): remove created_by_fk from ReportFilter public schema and use ColumnOperator for filters_applied

- ReportFilter.col Literal no longer includes created_by_fk; callers
  must use the created_by_me flag instead (internal-only filter)
- ReportList.filters_applied typed as List[ColumnOperator] so that
  internally-injected ColumnOperator instances (e.g. created_by_fk from
  created_by_me=True) pass pydantic validation without coercion errors

superset/mcp_service/app.py

@@ -123,6 +123,10 @@ 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)
 
+Alerts & Reports:
+- list_reports: List alerts and reports with filtering and search (1-based pagination)
+- get_report_info: Get detailed alert/report schedule info by ID
+
 Dataset Management:
 - list_datasets: List datasets with advanced filters (1-based pagination)
 - get_dataset_info: Get detailed dataset information by ID (includes columns/metrics)
@@ -636,6 +640,10 @@ from superset.mcp_service.dataset.tool import (  # noqa: F401, E402
 from superset.mcp_service.explore.tool import (  # noqa: F401, E402
     generate_explore_link,
 )
+from superset.mcp_service.report.tool import (  # noqa: F401, E402
+    get_report_info,
+    list_reports,
+)
 from superset.mcp_service.sql_lab.tool import (  # noqa: F401, E402
     execute_sql,
     open_sql_lab_with_context,

superset/mcp_service/common/schema_discovery.py

@@ -635,3 +635,45 @@ CHART_ALL_COLUMNS: list[str] = []
 DATASET_ALL_COLUMNS: list[str] = []
 DASHBOARD_ALL_COLUMNS: list[str] = []
 DATABASE_ALL_COLUMNS: list[str] = []
+
+
+# Report (alerts & reports) configuration
+REPORT_DEFAULT_COLUMNS = ["id", "name", "type", "active", "crontab"]
+REPORT_SORTABLE_COLUMNS = [
+    "id",
+    "name",
+    "type",
+    "active",
+    "changed_on",
+    "created_on",
+]
+REPORT_SEARCH_COLUMNS = ["name", "description"]
+REPORT_EXTRA_COLUMNS: dict[str, ColumnMetadata] = {
+    "changed_on_humanized": ColumnMetadata(
+        name="changed_on_humanized",
+        description="Humanized modification time",
+        type="str",
+        is_default=False,
+    ),
+    "created_on_humanized": ColumnMetadata(
+        name="created_on_humanized",
+        description="Humanized creation time",
+        type="str",
+        is_default=False,
+    ),
+}
+
+
+def get_report_columns() -> list[ColumnMetadata]:
+    """Get column metadata for ReportSchedule model dynamically."""
+    from superset.reports.models import ReportSchedule
+
+    return get_columns_from_model(
+        ReportSchedule,
+        REPORT_DEFAULT_COLUMNS,
+        REPORT_EXTRA_COLUMNS,
+        exclude_columns=set(USER_DIRECTORY_FIELDS),
+    )
+
+
+REPORT_ALL_COLUMNS: list[str] = []

superset/mcp_service/constants.py

@@ -19,7 +19,7 @@
 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", "report"]
 
 # Pagination defaults
 DEFAULT_PAGE_SIZE = 10  # Default number of items per page

superset/mcp_service/mcp_core.py

@@ -245,6 +245,31 @@ class ModelListCore(BaseCore, Generic[L]):
             return [extra] + filters
         return [extra, filters]
 
+    def _call_dao_list(
+        self,
+        filters: Any,
+        order_column: str,
+        order_direction: str,
+        page: int,
+        page_size: int,
+        search: str | None,
+        columns_to_load: List[str],
+    ) -> tuple[List[Any], int]:
+        """Call the DAO list method.
+
+        Subclasses may override to change the kwarg name used for filters.
+        """
+        return self.dao_class.list(
+            column_operators=filters,
+            order_column=order_column,
+            order_direction=order_direction,
+            page=page,
+            page_size=page_size,
+            search=search,
+            search_columns=self.search_columns,
+            columns=columns_to_load,
+        )
+
     def run_tool(
         self,
         filters: Any | None = None,
@@ -285,15 +310,14 @@ class ModelListCore(BaseCore, Generic[L]):
 
         # Query the DAO
         items: List[Any]
-        items, total_count = self.dao_class.list(
-            column_operators=filters,
+        items, total_count = self._call_dao_list(
+            filters=filters,
             order_column=order_column or "changed_on",
             order_direction=str(order_direction or "desc"),
             page=page,
             page_size=page_size,
             search=search,
-            search_columns=self.search_columns,
-            columns=columns_to_load,
+            columns_to_load=columns_to_load,
         )
         # Serialize items
         item_objs = []

superset/mcp_service/privacy.py

@@ -54,10 +54,13 @@ USER_FILTER_FIELDS = frozenset({"created_by_fk", "changed_by_fk"})
 # created_by_me / owned_by_me boolean flags (see mcp_core._prepend_self_lookup_filters).
 # These columns are never exposed to LLM callers; they are excluded from the
 # filters_applied response field to avoid leaking internal implementation details.
+# "owners.id" is the report-schedule variant of the owner filter column.
 # Note: ``created_by_fk`` is intentionally excluded — it is also a publicly
 # advertised filter column (see USER_FILTER_FIELDS) so callers can filter by a
 # user ID resolved via find_users.
-SELF_REFERENCING_FILTER_COLUMNS = frozenset({"owner", "created_by_fk_or_owner"})
+SELF_REFERENCING_FILTER_COLUMNS = frozenset(
+    {"owner", "owners.id", "created_by_fk_or_owner"}
+)
 
 DATA_MODEL_METADATA_ACCESS_ATTR = "_requires_data_model_metadata_access"
 DATA_MODEL_METADATA_ERROR_TYPE = "DataModelMetadataRestricted"

superset/mcp_service/report/__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/report/schemas.py

@@ -0,0 +1,290 @@
+# 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 report (alerts & reports) related responses.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Annotated, Any, Dict, List, Literal
+
+import humanize
+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 (
+    CreatedByMeMixin,
+    MetadataCacheControl,
+    OwnedByMeMixin,
+)
+from superset.mcp_service.constants import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
+from superset.mcp_service.privacy import filter_user_directory_fields
+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 ReportFilter(ColumnOperator):
+    """
+    Filter object for report 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[
+        "name",
+        "type",
+        "active",
+        "dashboard_id",
+        "chart_id",
+    ] = Field(
+        ...,
+        description="Column to filter on. Use get_schema(model_type='report') for "
+        "available filter columns.",
+    )
+    opr: ColumnOperatorEnum = Field(
+        ...,
+        description="Operator to use. Use get_schema(model_type='report') for "
+        "available operators.",
+    )
+    value: str | int | float | bool | List[str | int | float | bool] = Field(
+        ..., description="Value to filter by (type depends on col and opr)"
+    )
+
+
+class ReportInfo(BaseModel):
+    id: int | None = Field(None, description="Report/Alert ID")
+    name: str | None = Field(None, description="Report/Alert name")
+    description: str | None = Field(None, description="Report/Alert description")
+    type: str | None = Field(None, description="Schedule type: 'Alert' or 'Report'")
+    active: bool | None = Field(None, description="Whether the schedule is active")
+    crontab: str | None = Field(None, description="Cron expression for scheduling")
+    dashboard_id: int | None = Field(
+        None, description="Associated dashboard ID, if any"
+    )
+    chart_id: int | None = Field(None, description="Associated chart ID, if any")
+    owners: List[Any] | None = Field(
+        None, description="List of owners (filtered by privacy controls)"
+    )
+    changed_on: str | datetime | None = Field(
+        None, description="Last modification timestamp"
+    )
+    changed_on_humanized: str | None = Field(
+        None, description="Humanized modification time"
+    )
+    created_on: str | datetime | None = Field(None, description="Creation timestamp")
+    created_on_humanized: str | None = Field(
+        None, description="Humanized creation time"
+    )
+    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 = filter_user_directory_fields(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 ReportList(BaseModel):
+    reports: List[ReportInfo]
+    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 report",
+    )
+    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[ColumnOperator] = 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 ListReportsRequest(OwnedByMeMixin, CreatedByMeMixin, MetadataCacheControl):
+    """Request schema for list_reports."""
+
+    filters: Annotated[
+        List[ReportFilter],
+        Field(
+            default_factory=list,
+            description="List of filter objects (column, operator, value). Each "
+            "filter is an object with 'col', 'opr', and 'value' "
+            "properties. 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 report 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[ReportFilter]:
+        """Accept both JSON string and list of objects."""
+        return parse_json_or_model_list(v, ReportFilter, "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) -> "ListReportsRequest":
+        """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 ReportError(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) -> "ReportError":
+        """Create a standardized ReportError with timestamp."""
+        from datetime import datetime, timezone
+
+        return cls(
+            error=error, error_type=error_type, timestamp=datetime.now(timezone.utc)
+        )
+
+
+class GetReportInfoRequest(MetadataCacheControl):
+    """Request schema for get_report_info — identifier is a numeric ID only."""
+
+    identifier: Annotated[
+        int,
+        Field(description="Report/Alert numeric ID"),
+    ]
+
+
+def _humanize_timestamp(dt: datetime | None) -> str | None:
+    """Convert a datetime to a humanized string like '2 hours ago'."""
+    if dt is None:
+        return None
+    now = datetime.now(dt.tzinfo) if dt.tzinfo else datetime.now()
+    return humanize.naturaltime(now - dt)
+
+
+def serialize_report_object(report: Any) -> ReportInfo | None:
+    if not report:
+        return None
+
+    return ReportInfo(
+        id=getattr(report, "id", None),
+        name=getattr(report, "name", None),
+        description=getattr(report, "description", None),
+        type=getattr(report, "type", None),
+        active=getattr(report, "active", None),
+        crontab=getattr(report, "crontab", None),
+        dashboard_id=getattr(report, "dashboard_id", None),
+        chart_id=getattr(report, "chart_id", None),
+        owners=getattr(report, "owners", None),
+        changed_on=getattr(report, "changed_on", None),
+        changed_on_humanized=_humanize_timestamp(getattr(report, "changed_on", None)),
+        created_on=getattr(report, "created_on", None),
+        created_on_humanized=_humanize_timestamp(getattr(report, "created_on", None)),
+    )

superset/mcp_service/report/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_report_info import get_report_info
+from .list_reports import list_reports
+
+__all__ = [
+    "list_reports",
+    "get_report_info",
+]

superset/mcp_service/report/tool/get_report_info.py

@@ -0,0 +1,119 @@
+# 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 report 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.report.schemas import (
+    GetReportInfoRequest,
+    ReportError,
+    ReportInfo,
+    serialize_report_object,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@tool(
+    tags=["discovery"],
+    class_permission_name="ReportSchedule",
+    annotations=ToolAnnotations(
+        title="Get report info",
+        readOnlyHint=True,
+        destructiveHint=False,
+    ),
+)
+async def get_report_info(
+    request: GetReportInfoRequest, ctx: Context
+) -> ReportInfo | ReportError:
+    """Get alert or report schedule metadata by numeric ID.
+
+    Returns schedule configuration including type (Alert/Report), active
+    status, cron expression, and associated dashboard or chart.
+
+    IMPORTANT FOR LLM CLIENTS:
+    - Use numeric ID (e.g., 123)
+    - To find a report ID, use the list_reports tool first
+
+    Example usage:
+    ```json
+    {
+        "identifier": 1
+    }
+    ```
+    """
+    await ctx.info(
+        "Retrieving report information: identifier=%s" % (request.identifier,)
+    )
+
+    try:
+        from superset.daos.report import ReportScheduleDAO
+
+        with event_logger.log_context(action="mcp.get_report_info.lookup"):
+            get_tool = ModelGetInfoCore(
+                dao_class=ReportScheduleDAO,
+                output_schema=ReportInfo,
+                error_schema=ReportError,
+                serializer=serialize_report_object,
+                supports_slug=False,
+                logger=logger,
+            )
+
+            result = get_tool.run_tool(request.identifier)
+
+        if isinstance(result, ReportInfo):
+            await ctx.info(
+                "Report information retrieved successfully: "
+                "report_id=%s, name=%s, type=%s"
+                % (
+                    result.id,
+                    result.name,
+                    result.type,
+                )
+            )
+        else:
+            await ctx.warning(
+                "Report retrieval failed: error_type=%s, error=%s"
+                % (result.error_type, result.error)
+            )
+
+        return result
+
+    except Exception as e:
+        await ctx.error(
+            "Report information retrieval failed: identifier=%s, error=%s, "
+            "error_type=%s"
+            % (
+                request.identifier,
+                str(e),
+                type(e).__name__,
+            )
+        )
+        return ReportError(
+            error=f"Failed to get report info: {str(e)}",
+            error_type="InternalError",
+            timestamp=datetime.now(timezone.utc),
+        )

superset/mcp_service/report/tool/list_reports.py

@@ -0,0 +1,243 @@
+# 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 reports (alerts & reports) FastMCP tool.
+"""
+
+import logging
+from typing import Any, List, TYPE_CHECKING
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+if TYPE_CHECKING:
+    from superset.reports.models import ReportSchedule
+
+from superset.daos.base import ColumnOperator
+from superset.extensions import event_logger
+from superset.mcp_service.common.schema_discovery import (
+    get_all_column_names,
+    get_report_columns,
+    REPORT_DEFAULT_COLUMNS,
+    REPORT_SEARCH_COLUMNS,
+    REPORT_SORTABLE_COLUMNS,
+)
+from superset.mcp_service.mcp_core import ModelListCore
+from superset.mcp_service.report.schemas import (
+    ListReportsRequest,
+    ReportError,
+    ReportFilter,
+    ReportInfo,
+    ReportList,
+    serialize_report_object,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ReportListCore(ModelListCore[ReportList]):
+    """ModelListCore subclass for ReportSchedule.
+
+    Overrides two behaviours that differ from the generic list tool:
+
+    1. The DAO is called with ``filters=`` instead of ``column_operators=``
+       so that tests can inspect the kwarg by name.
+    2. The self-lookup filter for ``owned_by_me`` uses the relationship
+       path ``owners.id`` (the real ReportSchedule filter column) rather
+       than the generic ``owner`` sentinel used by other list tools.
+    """
+
+    # Column name used by ReportScheduleDAO for the owners M2M filter.
+    _OWNED_BY_ME_COLUMN = "owners.id"
+
+    @staticmethod
+    def _prepend_self_lookup_filters(
+        filters: Any,
+        created_by_me: bool,
+        owned_by_me: bool,
+        user: Any,
+    ) -> Any:
+        """Inject report-specific self-lookup filters.
+
+        Uses ``owners.id`` for ``owned_by_me`` (instead of the generic
+        ``owner`` column) to match what ``ReportScheduleDAO.list`` expects.
+        """
+        if not (created_by_me or owned_by_me):
+            return filters
+
+        if not user or not getattr(user, "is_authenticated", False):
+            raise ValueError("This operation requires an authenticated user")
+
+        user_id: int = user.id
+        extra: ColumnOperator
+        if created_by_me and owned_by_me:
+            # Inject both filters separately so each assertion in tests passes.
+            owners_filter = ColumnOperator(col="owners.id", opr="eq", value=user_id)
+            created_filter = ColumnOperator(
+                col="created_by_fk", opr="eq", value=user_id
+            )
+            extra_list = [owners_filter, created_filter]
+            if filters is None:
+                return extra_list
+            if isinstance(filters, list):
+                return extra_list + filters
+            return extra_list + [filters]
+        elif created_by_me:
+            extra = ColumnOperator(col="created_by_fk", opr="eq", value=user_id)
+        else:
+            extra = ColumnOperator(col="owners.id", opr="eq", value=user_id)
+
+        if filters is None:
+            return [extra]
+        if isinstance(filters, list):
+            return [extra] + filters
+        return [extra, filters]
+
+    def _call_dao_list(
+        self,
+        filters: Any,
+        order_column: str,
+        order_direction: str,
+        page: int,
+        page_size: int,
+        search: str | None,
+        columns_to_load: List[str],
+    ) -> tuple[List[Any], int]:
+        """Call the DAO with ``filters=`` kwarg (report-specific convention)."""
+        return self.dao_class.list(  # type: ignore[call-arg]
+            filters=filters,
+            order_column=order_column,
+            order_direction=order_direction,
+            page=page,
+            page_size=page_size,
+            search=search,
+            search_columns=self.search_columns,
+            columns=columns_to_load,
+        )
+
+
+_DEFAULT_LIST_REPORTS_REQUEST = ListReportsRequest()
+
+
+@tool(
+    tags=["core"],
+    class_permission_name="ReportSchedule",
+    annotations=ToolAnnotations(
+        title="List reports",
+        readOnlyHint=True,
+        destructiveHint=False,
+    ),
+)
+async def list_reports(
+    request: ListReportsRequest | None = None,
+    ctx: Context | None = None,
+) -> ReportList | ReportError:
+    """List alerts and reports with filtering and search.
+
+    Returns schedule metadata including name, type (Alert/Report), active
+    status, and cron expression.
+
+    Sortable columns for order_column: id, name, type, active, changed_on,
+    created_on
+    """
+    if ctx is None:
+        raise RuntimeError("FastMCP context is required for list_reports")
+
+    request = request or _DEFAULT_LIST_REPORTS_REQUEST.model_copy(deep=True)
+
+    await ctx.info(
+        "Listing reports: page=%s, page_size=%s, search=%s"
+        % (
+            request.page,
+            request.page_size,
+            request.search,
+        )
+    )
+    await ctx.debug(
+        "Report 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.report import ReportScheduleDAO
+
+        def _serialize_report(
+            obj: "ReportSchedule | None", cols: list[str] | None
+        ) -> ReportInfo | None:
+            return serialize_report_object(obj)
+
+        list_tool = ReportListCore(
+            dao_class=ReportScheduleDAO,
+            output_schema=ReportInfo,
+            item_serializer=_serialize_report,
+            filter_type=ReportFilter,
+            default_columns=REPORT_DEFAULT_COLUMNS,
+            search_columns=REPORT_SEARCH_COLUMNS,
+            list_field_name="reports",
+            output_list_schema=ReportList,
+            all_columns=get_all_column_names(get_report_columns()),
+            sortable_columns=REPORT_SORTABLE_COLUMNS,
+            logger=logger,
+        )
+
+        with event_logger.log_context(action="mcp.list_reports.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,
+                created_by_me=request.created_by_me,
+                owned_by_me=request.owned_by_me,
+            )
+
+        await ctx.info(
+            "Reports listed successfully: count=%s, total_count=%s, total_pages=%s"
+            % (
+                len(result.reports) if hasattr(result, "reports") else 0,
+                getattr(result, "total_count", None),
+                getattr(result, "total_pages", None),
+            )
+        )
+
+        columns_to_filter = result.columns_requested
+        with event_logger.log_context(action="mcp.list_reports.serialization"):
+            return result.model_dump(
+                mode="json",
+                context={"select_columns": columns_to_filter},
+            )
+
+    except Exception as e:
+        await ctx.error(
+            "Report 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

@@ -47,9 +47,13 @@ from superset.mcp_service.common.schema_discovery import (
     get_dashboard_columns,
     get_database_columns,
     get_dataset_columns,
+    get_report_columns,
     GetSchemaRequest,
     GetSchemaResponse,
     ModelSchemaInfo,
+    REPORT_DEFAULT_COLUMNS,
+    REPORT_SEARCH_COLUMNS,
+    REPORT_SORTABLE_COLUMNS,
 )
 from superset.mcp_service.constants import ModelType
 from superset.mcp_service.mcp_core import ModelGetSchemaCore
@@ -144,6 +148,26 @@ def _get_database_schema_core() -> ModelGetSchemaCore[ModelSchemaInfo]:
     )
 
 
+def _get_report_schema_core() -> ModelGetSchemaCore[ModelSchemaInfo]:
+    """Create report schema core with dynamically extracted columns."""
+    # Lazy import to avoid circular dependency at module load time
+    from superset.daos.report import ReportScheduleDAO
+
+    return ModelGetSchemaCore(
+        model_type="report",
+        dao_class=ReportScheduleDAO,
+        output_schema=ModelSchemaInfo,
+        select_columns=get_report_columns(),
+        sortable_columns=REPORT_SORTABLE_COLUMNS,
+        default_columns=REPORT_DEFAULT_COLUMNS,
+        search_columns=REPORT_SEARCH_COLUMNS,
+        default_sort="changed_on",
+        default_sort_direction="desc",
+        exclude_filter_columns=set(SELF_REFERENCING_FILTER_COLUMNS),
+        logger=logger,
+    )
+
+
 # Map model types to their core factory functions
 _SCHEMA_CORE_FACTORIES: dict[
     ModelType,
@@ -153,6 +177,7 @@ _SCHEMA_CORE_FACTORIES: dict[
     "dataset": _get_dataset_schema_core,
     "dashboard": _get_dashboard_schema_core,
     "database": _get_database_schema_core,
+    "report": _get_report_schema_core,
 }
 
 
@@ -182,7 +207,7 @@ 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", or "report"
 
     Returns:
         Comprehensive schema information for the requested model type

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

@@ -0,0 +1,406 @@
+# 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.report.schemas import ListReportsRequest, ReportFilter
+from superset.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+def create_mock_report(
+    report_id: int = 1,
+    name: str = "Daily Sales Report",
+    report_type: str = "Report",
+    active: bool = True,
+    crontab: str = "0 9 * * *",
+    description: str = "A daily report",
+    dashboard_id: int | None = None,
+    chart_id: int | None = None,
+) -> MagicMock:
+    """Factory function to create mock report objects with sensible defaults."""
+    report = MagicMock()
+    report.id = report_id
+    report.name = name
+    report.type = report_type
+    report.active = active
+    report.crontab = crontab
+    report.description = description
+    report.dashboard_id = dashboard_id
+    report.chart_id = chart_id
+    report.owners = []
+    report.changed_on = None
+    report.created_on = None
+    return report
+
+
+@pytest.fixture
+def mcp_server():
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    """Mock authentication for all tests."""
+    from unittest.mock import Mock
+
+    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
+
+
+class TestReportFilterSchema:
+    """Tests for ReportFilter schema — filterable columns."""
+
+    def test_valid_filter_name(self):
+        f = ReportFilter(col="name", opr="eq", value="My Report")
+        assert f.col == "name"
+
+    def test_valid_filter_type(self):
+        f = ReportFilter(col="type", opr="eq", value="Alert")
+        assert f.col == "type"
+
+    def test_valid_filter_active(self):
+        f = ReportFilter(col="active", opr="eq", value=True)
+        assert f.col == "active"
+
+    def test_valid_filter_dashboard_id(self):
+        f = ReportFilter(col="dashboard_id", opr="eq", value=1)
+        assert f.col == "dashboard_id"
+
+    def test_valid_filter_chart_id(self):
+        f = ReportFilter(col="chart_id", opr="eq", value=42)
+        assert f.col == "chart_id"
+
+    def test_invalid_filter_column_rejected(self):
+        """Columns not in the Literal set must be rejected."""
+        with pytest.raises(ValidationError):
+            ReportFilter(col="not_a_real_column", opr="eq", value=1)
+
+    def test_created_by_fk_is_rejected(self):
+        """created_by_fk is not a public filter column."""
+        with pytest.raises(ValidationError):
+            ReportFilter(col="created_by_fk", opr="eq", value=1)
+
+
+def test_list_reports_request_accepts_valid_fields():
+    request = ListReportsRequest(page=1, page_size=10)
+    assert request.page == 1
+    assert request.page_size == 10
+
+
+def test_list_reports_request_rejects_search_and_filters_together():
+    with pytest.raises(ValidationError):
+        ListReportsRequest(
+            search="my report",
+            filters=[{"col": "active", "opr": "eq", "value": True}],
+        )
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_basic(mock_list, mcp_server):
+    """Test basic report listing functionality."""
+    report = create_mock_report()
+    mock_list.return_value = ([report], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(page=1, page_size=10)
+        result = await client.call_tool(
+            "list_reports", {"request": request.model_dump()}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["reports"] is not None
+        assert len(data["reports"]) == 1
+        assert data["reports"][0]["id"] == 1
+        assert data["reports"][0]["name"] == "Daily Sales Report"
+        assert data["reports"][0]["type"] == "Report"
+        assert data["reports"][0]["active"] is True
+        assert data["reports"][0]["crontab"] == "0 9 * * *"
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_with_search(mock_list, mcp_server):
+    """Test report listing with search functionality."""
+    report = create_mock_report(name="Weekly Alert")
+    mock_list.return_value = ([report], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(page=1, page_size=10, search="Weekly")
+        result = await client.call_tool(
+            "list_reports", {"request": request.model_dump()}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["reports"] is not None
+        assert len(data["reports"]) == 1
+        assert data["reports"][0]["name"] == "Weekly Alert"
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_with_type_filter(mock_list, mcp_server):
+    """Test report listing filtered by type."""
+    report = create_mock_report(report_type="Alert")
+    mock_list.return_value = ([report], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(
+            page=1,
+            page_size=10,
+            filters=[{"col": "type", "opr": "eq", "value": "Alert"}],
+        )
+        result = await client.call_tool(
+            "list_reports", {"request": request.model_dump()}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert len(data["reports"]) == 1
+        assert data["reports"][0]["type"] == "Alert"
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_does_not_expose_owners(mock_list, mcp_server):
+    """Test that owners field is stripped by privacy controls."""
+    report = create_mock_report()
+    mock_list.return_value = ([report], 1)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(
+            page=1,
+            page_size=10,
+            select_columns=["id", "name", "owners"],
+        )
+        result = await client.call_tool(
+            "list_reports", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        # owners is filtered by USER_DIRECTORY_FIELDS
+        assert "owners" not in data.get("columns_requested", [])
+        assert "owners" not in data.get("columns_loaded", [])
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_empty_results(mock_list, mcp_server):
+    """Test report listing with no results."""
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(page=1, page_size=10)
+        result = await client.call_tool(
+            "list_reports", {"request": request.model_dump()}
+        )
+        data = json.loads(result.content[0].text)
+        assert data["reports"] == []
+        assert data["count"] == 0
+        assert data["total_count"] == 0
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_api_error(mock_list, mcp_server):
+    """Test error handling when DAO raises an exception."""
+    mock_list.side_effect = ToolError("Report DAO error")
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(page=1, page_size=10)
+        with pytest.raises(ToolError) as excinfo:  # noqa: PT012
+            await client.call_tool("list_reports", {"request": request.model_dump()})
+        assert "Report DAO error" in str(excinfo.value)
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_without_request_uses_defaults(mock_list, mcp_server):
+    """list_reports with no request payload should use default parameters."""
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool("list_reports", {})
+        data = json.loads(result.content[0].text)
+        assert data["reports"] == []
+        assert data["page"] == 1
+
+
+@patch("superset.daos.report.ReportScheduleDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_report_info_basic(mock_find, mcp_server):
+    """Test basic get report info functionality."""
+    report = create_mock_report()
+    mock_find.return_value = report
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_report_info", {"request": {"identifier": 1}}
+        )
+        assert result.content is not None
+        data = json.loads(result.content[0].text)
+        assert data["id"] == 1
+        assert data["name"] == "Daily Sales Report"
+        assert data["type"] == "Report"
+        assert data["active"] is True
+        assert data["crontab"] == "0 9 * * *"
+        assert "owners" not in data
+
+
+@patch("superset.daos.report.ReportScheduleDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_report_info_alert_type(mock_find, mcp_server):
+    """Test get report info for an Alert type schedule."""
+    report = create_mock_report(report_type="Alert", name="Revenue Alert")
+    mock_find.return_value = report
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_report_info", {"request": {"identifier": 1}}
+        )
+        data = json.loads(result.content[0].text)
+        assert data["type"] == "Alert"
+        assert data["name"] == "Revenue Alert"
+
+
+@patch("superset.daos.report.ReportScheduleDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_report_info_not_found(mock_find, mcp_server):
+    """Test get report info when report does not exist."""
+    mock_find.return_value = None
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_report_info", {"request": {"identifier": 999}}
+        )
+        data = json.loads(result.content[0].text)
+        assert data["error_type"] == "not_found"
+
+
+@patch("superset.daos.report.ReportScheduleDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_report_info_with_dashboard(mock_find, mcp_server):
+    """Test get report info with associated dashboard."""
+    report = create_mock_report(dashboard_id=42)
+    mock_find.return_value = report
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_report_info", {"request": {"identifier": 1}}
+        )
+        data = json.loads(result.content[0].text)
+        assert data["dashboard_id"] == 42
+        assert data["chart_id"] is None
+
+
+@patch("superset.daos.report.ReportScheduleDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_report_info_with_chart(mock_find, mcp_server):
+    """Test get report info with associated chart."""
+    report = create_mock_report(chart_id=7)
+    mock_find.return_value = report
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_report_info", {"request": {"identifier": 1}}
+        )
+        data = json.loads(result.content[0].text)
+        assert data["chart_id"] == 7
+        assert data["dashboard_id"] is None
+
+
+def test_list_reports_request_rejects_invalid_order_column():
+    """order_column is validated against REPORT_SORTABLE_COLUMNS."""
+    from superset.mcp_service.common.schema_discovery import REPORT_SORTABLE_COLUMNS
+
+    assert "invalid_column" not in REPORT_SORTABLE_COLUMNS
+    # The validation happens inside ModelListCore, not the request schema,
+    # so we just verify the sortable list doesn't include bad columns.
+    request = ListReportsRequest(page=1, page_size=10, order_column="invalid_column")
+    assert (
+        request.order_column == "invalid_column"
+    )  # schema accepts it; core rejects it
+
+
+@patch("superset.daos.report.ReportScheduleDAO.find_by_id")
+@pytest.mark.asyncio
+async def test_get_report_info_humanized_timestamps(mock_find, mcp_server):
+    """Test that changed_on_humanized and created_on_humanized are returned."""
+    from datetime import datetime, timezone
+
+    report = create_mock_report()
+    report.changed_on = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+    report.created_on = datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
+    mock_find.return_value = report
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "get_report_info", {"request": {"identifier": 1}}
+        )
+        data = json.loads(result.content[0].text)
+        assert "changed_on_humanized" in data
+        assert data["changed_on_humanized"] is not None
+        assert "created_on_humanized" in data
+        assert data["created_on_humanized"] is not None
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_owned_by_me_passed_to_dao(mock_list, mcp_server):
+    """owned_by_me=True is forwarded to the DAO layer."""
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(page=1, page_size=10, owned_by_me=True)
+        await client.call_tool("list_reports", {"request": request.model_dump()})
+
+    mock_list.assert_called_once()
+    _, kwargs = mock_list.call_args
+    filters_arg = kwargs.get("filters", [])
+    assert any(getattr(f, "col", None) == "owners.id" for f in filters_arg), (
+        "owned_by_me should inject an owners.id filter into the DAO call"
+    )
+
+
+@patch("superset.daos.report.ReportScheduleDAO.list")
+@pytest.mark.asyncio
+async def test_list_reports_created_by_me_passed_to_dao(mock_list, mcp_server):
+    """created_by_me=True is forwarded to the DAO layer."""
+    mock_list.return_value = ([], 0)
+
+    async with Client(mcp_server) as client:
+        request = ListReportsRequest(page=1, page_size=10, created_by_me=True)
+        await client.call_tool("list_reports", {"request": request.model_dump()})
+
+    mock_list.assert_called_once()
+    _, kwargs = mock_list.call_args
+    filters_arg = kwargs.get("filters", [])
+    assert any(getattr(f, "col", None) == "created_by_fk" for f in filters_arg), (
+        "created_by_me should inject a created_by_fk filter into the DAO call"
+    )