This is an automated email from the ASF dual-hosted git repository.
aminghadersohi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new 33292f249da feat(mcp): support tab-scoped dashboard layouts (#44797)
33292f249da is described below
commit 33292f249dad32c08921d6eac2a866be53c6c5e5
Author: Amin Ghadersohi <[email protected]>
AuthorDate: Fri Oct 2 08:42:33 2026 +1000
feat(mcp): support tab-scoped dashboard layouts (#44797)
---
docs/admin_docs/configuration/mcp-server.mdx | 30 +
superset/mcp_service/dashboard/schemas.py | 103 ++-
.../dashboard/tool/get_dashboard_layout.py | 161 ++++-
superset/mcp_service/middleware.py | 7 +-
superset/mcp_service/utils/response_size_utils.py | 176 +++++
.../dashboard/tool/test_get_dashboard_layout.py | 744 ++++++++++++++++++++-
6 files changed, 1202 insertions(+), 19 deletions(-)
diff --git a/docs/admin_docs/configuration/mcp-server.mdx
b/docs/admin_docs/configuration/mcp-server.mdx
index a2b91b10479..6a247989508 100644
--- a/docs/admin_docs/configuration/mcp-server.mdx
+++ b/docs/admin_docs/configuration/mcp-server.mdx
@@ -1268,6 +1268,36 @@ filters update the `BASE_AXIS` column's `timeGrain`,
regardless of its position
in the query's `columns` list. Other columns are left untouched. Queries
without
a `BASE_AXIS` column continue to use `extras.time_grain_sqla`.
+### Exploring large dashboard layouts
+
+`get_dashboard_layout` returns the full parsed layout by default. For large
+dashboards, pass `tabs_only=true` to retrieve just the tab tree in `tab_tree`:
+`id`, `name` (the title), `parent_tab_id`, zero-based `depth`, and
+`chart_count` (distinct charts in the tab and its descendants). In this mode
+`tabs` and `charts` are empty by request, not because the dashboard is empty.
+Then pass `tab="<ID or exact title>"` to retrieve that tab and its descendants
+with chart positions. Matching is case-sensitive; IDs take precedence and
+ambiguous titles require an ID. Both options can be combined to summarize a
+subtree. Parent IDs, depths, and chart paths retain their full-dashboard
context.
+
+Charts placed outside every tab are counted in `untabbed_chart_count` in every
+response; pass `untabbed_only=true` (without `tab` or `tabs_only`) to list only
+those charts. The `scope` field echoes the applied `tabs_only`, resolved
+`tab_id`, and `untabbed_only` values, and is `null` for the full layout.
+Scoping only removes tabs and chart placements from the layout the caller can
+already retrieve. Conflicting scope options return an actionable
+`invalid_scope` error naming the options to remove; blank tab selectors return
+`tab_not_found` rather than widening the scope. On dashboards without tabs,
+`tabs_only=true` returns `tab_not_found` with guidance to use
+`untabbed_only=true` or request the full layout instead.
+
+Permalink active-tab and filter state remain unchanged; only an explicit `tab`
+scopes the response. Oversized layout errors suggest the next narrower scope:
+the tab tree and untabbed charts for a full tabbed layout, a top-level tab for
+an oversized tab tree, a nested tab for a tab that has one, and `list_charts`
+paging when a dashboard without tabs or a tab without nested tabs is still too
+large.
+
### Reading applied dashboard filters
When answering about a filtered dashboard, call `get_dashboard_info` with the
diff --git a/superset/mcp_service/dashboard/schemas.py
b/superset/mcp_service/dashboard/schemas.py
index 54cab9a0914..82073216082 100644
--- a/superset/mcp_service/dashboard/schemas.py
+++ b/superset/mcp_service/dashboard/schemas.py
@@ -335,29 +335,41 @@ class GetDashboardInfoRequest(MetadataCacheControl):
class GetDashboardLayoutRequest(BaseModel):
- """Request a dashboard layout by its identifier or shared permalink.
-
- Permalink requests resolve the dashboard while preserving shared active-tab
- and filter state in the response.
- """
+ """Dashboard layout, optionally scoped."""
identifier: Annotated[
int | str | None,
Field(
default=None,
description=(
- "Dashboard ID, UUID, slug, bare permalink key, or a shared URL
"
- "containing /superset/dashboard/p/<key>/. Omit when "
- "permalink_key is provided."
+ "Dashboard ID, UUID, slug, or permalink key/URL
(/dashboard/p/<key>/). "
+ "Omit with permalink_key."
),
),
]
permalink_key: str | None = Field(
default=None,
description=(
- "Key from a shared dashboard URL such as "
- "'/superset/dashboard/p/<key>/'. Resolves the dashboard and
includes "
- "the shared active-tab and filter context in the layout response."
+ "Permalink key; resolves dashboard and preserves shared "
+ "active-tab/filter state."
+ ),
+ )
+ tabs_only: bool = Field(
+ default=False,
+ description=(
+ "Tab tree only (ID, name, parent, depth, chart_count), in
tab_tree."
+ ),
+ )
+ tab: str | None = Field(
+ default=None,
+ description=(
+ "Tab subtree by ID or exact title (case-sensitive); IDs win over
titles."
+ ),
+ )
+ untabbed_only: bool = Field(
+ default=False,
+ description=(
+ "Only charts outside every tab; cannot combine with tab or
tabs_only."
),
)
@@ -1519,6 +1531,39 @@ class DashboardTab(BaseModel):
)
+class DashboardTabSummary(BaseModel):
+ """Compact tab-tree entry without chart IDs or chart positions."""
+
+ id: str = Field(..., description="Tab component ID from position_json")
+ name: str | None = Field(None, description="Tab display name")
+ parent_tab_id: str | None = Field(None, description="ID of the enclosing
tab")
+ depth: int = Field(
+ ..., description="Tab nesting depth; top-level tabs have depth 0"
+ )
+ chart_count: int = Field(
+ ...,
+ description="Distinct chart IDs directly or indirectly under this tab",
+ )
+
+
+class DashboardLayoutScope(BaseModel):
+ """Scope applied to a get_dashboard_layout response."""
+
+ tabs_only: bool = Field(
+ False,
+ description=(
+ "True when chart positions were omitted by request; charts is then
"
+ "empty regardless of what the dashboard contains."
+ ),
+ )
+ tab_id: str | None = Field(
+ None, description="Resolved ID of the tab the response is limited to"
+ )
+ untabbed_only: bool = Field(
+ False, description="True when only charts outside every tab are
returned"
+ )
+
+
class DashboardLayout(BaseModel):
"""Parsed layout data for a dashboard, derived from position_json."""
@@ -1528,12 +1573,38 @@ class DashboardLayout(BaseModel):
tabs: List[DashboardTab] = Field(
default_factory=list,
description=(
- "Tabs declared in the dashboard layout (empty for untabbed
dashboards)"
+ "Tabs declared in the dashboard layout, limited to the selected "
+ "subtree when scoped by tab. Empty for untabbed dashboards and
when "
+ "tabs_only or untabbed_only is requested."
+ ),
+ )
+ tab_tree: List[DashboardTabSummary] = Field(
+ default_factory=list,
+ description=(
+ "Compact tab tree without chart IDs or positions; populated only
when "
+ "tabs_only is requested."
),
)
charts: List[ChartPosition] = Field(
default_factory=list,
- description="Charts placed in the dashboard layout with their tab
context",
+ description=(
+ "Charts placed in the dashboard layout with their tab context,
limited "
+ "to the requested scope. Always empty when tabs_only is requested;
"
+ "see scope."
+ ),
+ )
+ untabbed_chart_count: int = Field(
+ 0,
+ description=(
+ "Count of distinct charts outside every tab in the full layout; "
+ "untabbed_only lists each placement."
+ ),
+ )
+ scope: DashboardLayoutScope | None = Field(
+ None,
+ description=(
+ "Scope applied to tabs, tab_tree, and charts; None for the full
layout."
+ ),
)
has_layout: bool = Field(
default=False,
@@ -2080,12 +2151,18 @@ def dashboard_layout_serializer(dashboard: "Dashboard")
-> DashboardLayout:
"""Serialize a Dashboard model to a parsed DashboardLayout."""
position_json_str = getattr(dashboard, "position_json", None)
tabs, charts = _extract_layout_from_position(position_json_str)
+ untabbed_chart_ids = {
+ chart.chart_id
+ for chart in charts
+ if chart.tab_id is None and chart.chart_id is not None
+ }
return DashboardLayout(
id=dashboard.id,
dashboard_title=dashboard.dashboard_title or "Untitled",
uuid=str(dashboard.uuid) if dashboard.uuid else None,
tabs=tabs,
charts=charts,
+ untabbed_chart_count=len(untabbed_chart_ids),
has_layout=bool(position_json_str),
)
diff --git a/superset/mcp_service/dashboard/tool/get_dashboard_layout.py
b/superset/mcp_service/dashboard/tool/get_dashboard_layout.py
index facfcad3a76..7a7e710f048 100644
--- a/superset/mcp_service/dashboard/tool/get_dashboard_layout.py
+++ b/superset/mcp_service/dashboard/tool/get_dashboard_layout.py
@@ -39,6 +39,9 @@ from superset.mcp_service.dashboard.schemas import (
dashboard_layout_serializer,
DashboardError,
DashboardLayout,
+ DashboardLayoutScope,
+ DashboardTab,
+ DashboardTabSummary,
GetDashboardLayoutRequest,
)
from superset.mcp_service.mcp_core import ModelGetInfoCore
@@ -46,6 +49,146 @@ from superset.mcp_service.mcp_core import ModelGetInfoCore
logger = logging.getLogger(__name__)
+def _resolve_tab(
+ tabs: list[DashboardTab], selector: str
+) -> DashboardTab | DashboardError:
+ """Match a tab by ID first, then by exact title."""
+ if not selector.strip():
+ return DashboardError.create(
+ "tab must not be blank. Omit tab for the full layout or use "
+ "tabs_only=true to discover tab IDs and titles.",
+ "tab_not_found",
+ )
+ if not tabs:
+ return DashboardError.create(
+ "This dashboard has no tabs. Omit tab to get the full layout.",
+ "tab_not_found",
+ )
+ matches = [tab for tab in tabs if tab.id == selector] or [
+ tab for tab in tabs if tab.name == selector
+ ]
+ if not matches:
+ return DashboardError.create(
+ "Tab not found. Use tabs_only=true to discover tab IDs and
titles.",
+ "tab_not_found",
+ )
+ if len(matches) > 1:
+ return DashboardError.create(
+ "Multiple tabs have that title. Use tabs_only=true to discover "
+ "their IDs, then pass a unique tab ID.",
+ "ambiguous_tab",
+ )
+ return matches[0]
+
+
+def _subtree_ids(tab_id: str, children: dict[str | None, list[str]]) ->
set[str]:
+ """Return a tab ID and the IDs of every tab nested under it."""
+ selected_ids: set[str] = set()
+ pending = [tab_id]
+ while pending:
+ current = pending.pop()
+ if current not in selected_ids:
+ selected_ids.add(current)
+ pending.extend(children.get(current, []))
+ return selected_ids
+
+
+def _tab_summaries(
+ tabs: list[DashboardTab], children: dict[str | None, list[str]]
+) -> list[DashboardTabSummary]:
+ """Summarize tabs with absolute depths from the full dashboard tree."""
+ # The parser only emits reachable tabs and assigns each tab a single
+ # enclosing parent, so walking from the top-level tabs reaches every tab.
+ depths: dict[str, int] = {}
+ stack = [(tab_id, 0) for tab_id in children.get(None, [])]
+ while stack:
+ tab_id, depth = stack.pop()
+ depths[tab_id] = depth
+ stack.extend((child, depth + 1) for child in children.get(tab_id, []))
+ return [
+ DashboardTabSummary(
+ id=tab.id,
+ name=tab.name,
+ parent_tab_id=tab.parent_tab_id,
+ depth=depths[tab.id],
+ chart_count=len(tab.chart_ids),
+ )
+ for tab in tabs
+ ]
+
+
+def _scope_layout(
+ layout: DashboardLayout, request: GetDashboardLayoutRequest
+) -> DashboardLayout | DashboardError:
+ """Project the parsed layout without changing permalink or ancestry
context.
+
+ Scoping only ever removes tabs and chart placements from the parsed layout,
+ so a scoped response never contains a chart the full layout would not.
+ """
+ if request.untabbed_only and (request.tabs_only or request.tab is not
None):
+ return DashboardError.create(
+ "untabbed_only cannot be combined with tab or tabs_only. "
+ "Drop tab and tabs_only to retrieve charts outside every tab, "
+ "or drop untabbed_only to use a tab scope.",
+ "invalid_scope",
+ )
+
+ if not request.tabs_only and request.tab is None and not
request.untabbed_only:
+ return layout
+
+ if request.untabbed_only:
+ return layout.model_copy(
+ update={
+ "tabs": [],
+ "charts": [chart for chart in layout.charts if chart.tab_id is
None],
+ "scope": DashboardLayoutScope(untabbed_only=True),
+ }
+ )
+
+ tabs = layout.tabs
+ if request.tabs_only and not tabs:
+ return DashboardError.create(
+ "This dashboard has no tabs. Use untabbed_only=true without tab "
+ "or tabs_only to get charts outside every tab, or omit the scope "
+ "options to get the full layout.",
+ "tab_not_found",
+ )
+ children: dict[str | None, list[str]] = {}
+ for tab in tabs:
+ children.setdefault(tab.parent_tab_id, []).append(tab.id)
+
+ selected_tab_id: str | None = None
+ if request.tab is not None:
+ selected = _resolve_tab(tabs, request.tab)
+ if isinstance(selected, DashboardError):
+ return selected
+ selected_tab_id = selected.id
+ subtree_ids = _subtree_ids(selected.id, children)
+ tabs = [tab for tab in tabs if tab.id in subtree_ids]
+
+ scope = DashboardLayoutScope(tabs_only=request.tabs_only,
tab_id=selected_tab_id)
+ if request.tabs_only:
+ return layout.model_copy(
+ update={
+ "tabs": [],
+ "tab_tree": _tab_summaries(tabs, children),
+ "charts": [],
+ "scope": scope,
+ }
+ )
+
+ selected_ids: set[str] = {tab.id for tab in tabs}
+ return layout.model_copy(
+ update={
+ "tabs": tabs,
+ "charts": [
+ chart for chart in layout.charts if chart.tab_id in
selected_ids
+ ],
+ "scope": scope,
+ }
+ )
+
+
@tool(
tags=["discovery"],
class_permission_name="Dashboard",
@@ -68,6 +211,15 @@ async def get_dashboard_layout(
explain which charts live under which tab, or to locate a chart by
its parent tab).
+ For large dashboards, pass ``tabs_only=true`` to discover the complete tab
+ tree with nesting depths and descendant chart counts, without chart
positions.
+ Then pass ``tab="<ID or exact title>"`` to retrieve only that tab's subtree
+ and chart positions. IDs take precedence; duplicate titles require an ID.
+ Combine both options to summarize a subtree. Pass ``untabbed_only=true``
+ for charts outside every tab.
+ Parent IDs and tab paths remain relative to the full dashboard, and
+ permalink state is preserved unchanged.
+
If the user gives you a shared URL containing ``/dashboard/p/<key>/``, pass
the URL or bare key as ``identifier`` (or use ``permalink_key`` alone). The
response identifies the active tab and includes the shared filter state.
@@ -129,14 +281,19 @@ async def get_dashboard_layout(
"permalink_key belongs to a different dashboard;
ignoring "
"its active-tab and filter state."
)
+ result = _scope_layout(result, request)
+
+ if isinstance(result, DashboardLayout):
await ctx.info(
"Dashboard layout retrieved: id=%s, tab_count=%s,
chart_count=%s, "
- "has_layout=%s"
+ "untabbed_chart_count=%s, has_layout=%s, scope=%s"
% (
result.id,
- len(result.tabs),
+ len(result.tab_tree) if request.tabs_only else
len(result.tabs),
len(result.charts),
+ result.untabbed_chart_count,
result.has_layout,
+ result.scope.model_dump() if result.scope else None,
)
)
else:
diff --git a/superset/mcp_service/middleware.py
b/superset/mcp_service/middleware.py
index 5617225f425..339d44484bf 100644
--- a/superset/mcp_service/middleware.py
+++ b/superset/mcp_service/middleware.py
@@ -70,6 +70,7 @@ from superset.mcp_service.utils.response_size_utils import (
format_size_limit_error,
get_response_size_bytes,
INFO_TOOLS,
+ RESPONSE_AWARE_SUGGESTION_TOOLS,
string_clip_chars,
STRING_FIELD_TRUNCATION_TOOLS,
truncate_oversized_response,
@@ -2116,7 +2117,11 @@ class ResponseSizeGuardMiddleware(Middleware):
params=params,
actual_bytes=actual_bytes,
max_bytes=self.max_bytes,
- response=None,
+ response=(
+ self._extract_payload_from_tool_result(response)
+ if tool_name in RESPONSE_AWARE_SUGGESTION_TOOLS
+ else None
+ ),
)
)
diff --git a/superset/mcp_service/utils/response_size_utils.py
b/superset/mcp_service/utils/response_size_utils.py
index 09cee73d368..91d72459115 100644
--- a/superset/mcp_service/utils/response_size_utils.py
+++ b/superset/mcp_service/utils/response_size_utils.py
@@ -116,6 +116,10 @@ def extract_query_params(params: Dict[str, Any] | None) ->
Dict[str, Any]:
"filters",
# Search
"search",
+ # Dashboard layout scope
+ "tab",
+ "tabs_only",
+ "untabbed_only",
]
return {k: params[k] for k in extract_keys if k in params}
@@ -277,6 +281,175 @@ def _identify_large_fields(response: ToolResponse) ->
List[str]:
return large_fields
+# Tools whose oversized-response suggestions depend on the blocked payload, so
+# the size guard passes the parsed payload to format_size_limit_error.
+RESPONSE_AWARE_SUGGESTION_TOOLS = frozenset({"get_dashboard_layout"})
+
+_MAX_SUGGESTED_TAB_IDS = 5
+_MAX_SUGGESTED_TAB_ID_CHARS = 64
+
+
+def _format_tab_ids(tab_ids: List[str]) -> str:
+ """Quote a bounded sample of tab IDs for inclusion in a suggestion."""
+ from superset.utils import json
+
+ sample = [
+ json.dumps(tab_id[:_MAX_SUGGESTED_TAB_ID_CHARS])
+ for tab_id in tab_ids[:_MAX_SUGGESTED_TAB_IDS]
+ ]
+ suffix = ", ..." if len(tab_ids) > _MAX_SUGGESTED_TAB_IDS else ""
+ return ", ".join(sample) + suffix
+
+
+def _layout_tab_entries(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
+ """Return the tab entries (full tabs or tab_tree summaries) in a
payload."""
+ entries: List[Dict[str, Any]] = []
+ for key in ("tabs", "tab_tree"):
+ value = payload.get(key)
+ if isinstance(value, list):
+ entries.extend(
+ entry
+ for entry in value
+ if isinstance(entry, dict) and isinstance(entry.get("id"), str)
+ )
+ return entries
+
+
+def _selected_tab_suggestions(
+ selected_tab: str,
+ tabs: List[Dict[str, Any]],
+ *,
+ tabs_only: bool,
+ known: bool,
+ list_charts_hint: str,
+) -> List[str]:
+ """Suggest how to narrow a response already scoped to one tab."""
+ child_ids = [tab["id"] for tab in tabs if tab.get("parent_tab_id") ==
selected_tab]
+ if child_ids:
+ suggestions = [
+ f"Pass a nested tab ID ({_format_tab_ids(child_ids)}) as tab to "
+ "narrow the response further."
+ ]
+ if not tabs_only:
+ suggestions.append(
+ "Add tabs_only=true to the same tab to see its subtree with "
+ "chart counts but without chart positions."
+ )
+ return suggestions
+ if not known:
+ return [
+ "If this tab has nested tabs, pass one of their IDs (see tab_tree "
+ "from tabs_only=true) as tab. A tab without nested tabs cannot be "
+ "narrowed further by tab.",
+ list_charts_hint,
+ ]
+ if tabs_only:
+ return [
+ "This tab has no nested tabs, so its tab tree cannot be narrowed "
+ "further by tab."
+ ]
+ return [
+ "This tab has no nested tabs, so tab cannot narrow it further. "
+ "Add tabs_only=true for its chart count only.",
+ list_charts_hint,
+ ]
+
+
+def _unscoped_layout_suggestions(
+ tabs: List[Dict[str, Any]],
+ untabbed_count: int,
+ *,
+ known: bool,
+ list_charts_hint: str,
+) -> List[str]:
+ """Suggest how to narrow a full dashboard layout."""
+ if known and not tabs:
+ return [
+ "This dashboard has no tabs, so tab and tabs_only cannot narrow
its "
+ "layout. " + list_charts_hint
+ ]
+ suggestions = [
+ "Call get_dashboard_layout with tabs_only=true to discover the tab
tree "
+ "without chart positions.",
+ 'Then pass tab="<ID or title>" to get only that tab\'s subtree and
chart '
+ "positions; use a nested tab ID for a smaller response.",
+ ]
+ if untabbed_count or not known:
+ suggestions.append(
+ "Pass untabbed_only=true to get the charts placed outside every
tab"
+ + (
+ f" ({untabbed_count} distinct "
+ f"{'chart' if untabbed_count == 1 else 'charts'})."
+ if untabbed_count
+ else "."
+ )
+ )
+ return suggestions
+
+
+def _dashboard_layout_suggestions(
+ query_params: Dict[str, Any], response: ToolResponse | None
+) -> List[str]:
+ """Suggest the next narrower get_dashboard_layout scope.
+
+ The blocked payload, when available, tells whether the dashboard has tabs,
+ whether the selected tab has nested tabs, and whether charts sit outside
+ every tab. Without it, suggestions depend on the request parameters only.
+ """
+ payload: Dict[str, Any] = {}
+ if isinstance(response, BaseModel):
+ payload = response.model_dump()
+ elif isinstance(response, dict):
+ payload = response
+ known = bool(payload)
+ tabs = _layout_tab_entries(payload)
+ raw_scope = payload.get("scope")
+ scope: Dict[str, Any] = raw_scope if isinstance(raw_scope, dict) else {}
+ dashboard_id = payload.get("id")
+ list_charts_hint = (
+ "Use list_charts with filters=[{col: 'dashboards', opr: 'eq', value: "
+ f"{dashboard_id if isinstance(dashboard_id, int) else '<ID>'}}}] and a
"
+ "small page_size to page through chart metadata without layout
positions."
+ )
+ tabs_only = bool(query_params.get("tabs_only"))
+
+ if query_params.get("untabbed_only"):
+ return [
+ "The charts outside every tab cannot be narrowed further by "
+ "get_dashboard_layout. " + list_charts_hint
+ ]
+
+ selected_tab = scope.get("tab_id") or query_params.get("tab")
+ if isinstance(selected_tab, str):
+ return _selected_tab_suggestions(
+ selected_tab,
+ tabs,
+ tabs_only=tabs_only,
+ known=known,
+ list_charts_hint=list_charts_hint,
+ )
+
+ if tabs_only:
+ top_level_ids = [tab["id"] for tab in tabs if tab.get("depth") == 0]
+ example = (
+ f" (top-level tabs: {_format_tab_ids(top_level_ids)})"
+ if top_level_ids
+ else ""
+ )
+ return [
+ 'Keep tabs_only=true and pass tab="<top-level tab ID>" to get one '
+ f"branch of the tab tree at a time{example}."
+ ]
+
+ raw_untabbed = payload.get("untabbed_chart_count")
+ return _unscoped_layout_suggestions(
+ tabs,
+ raw_untabbed if isinstance(raw_untabbed, int) else 0,
+ known=known,
+ list_charts_hint=list_charts_hint,
+ )
+
+
def _get_tool_specific_suggestions(
tool_name: str,
query_params: Dict[str, Any],
@@ -319,6 +492,9 @@ def _get_tool_specific_suggestions(
"to shorten the generated query."
)
+ elif tool_name == "get_dashboard_layout":
+ suggestions.extend(_dashboard_layout_suggestions(query_params,
response))
+
elif tool_name in ("get_chart_info", "get_dashboard_info",
"get_dataset_info"):
suggestions.append(
f"For {tool_name}, use 'select_columns' to fetch only specific
metadata "
diff --git
a/tests/unit_tests/mcp_service/dashboard/tool/test_get_dashboard_layout.py
b/tests/unit_tests/mcp_service/dashboard/tool/test_get_dashboard_layout.py
index 4a3af4eb2df..7c575b0125d 100644
--- a/tests/unit_tests/mcp_service/dashboard/tool/test_get_dashboard_layout.py
+++ b/tests/unit_tests/mcp_service/dashboard/tool/test_get_dashboard_layout.py
@@ -17,10 +17,11 @@
"""Unit tests for the MCP get_dashboard_layout tool."""
+from typing import Any
from unittest.mock import Mock, patch
import pytest
-from fastmcp import Client
+from fastmcp import Client, FastMCP
from superset.mcp_service.app import mcp
from superset.mcp_service.dashboard.schemas import (
@@ -261,9 +262,10 @@ async def test_get_dashboard_layout_not_found(mock_find,
mcp_server):
@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
[email protected]("options", [{}, {"tabs_only": True}, {"tab":
"TAB-1"}])
@pytest.mark.asyncio
async def test_get_dashboard_layout_resolves_shared_permalink(
- mock_permalink, mock_find, mcp_server
+ mock_permalink, mock_find, mcp_server, options
):
mock_permalink.return_value = (
"shared-key",
@@ -284,7 +286,8 @@ async def
test_get_dashboard_layout_resolves_shared_permalink(
"get_dashboard_layout",
{
"request": {
- "identifier":
"https://example.test/superset/dashboard/p/shared-key/";
+ "identifier":
"https://example.test/superset/dashboard/p/shared-key/";,
+ **options,
}
},
)
@@ -295,6 +298,27 @@ async def
test_get_dashboard_layout_resolves_shared_permalink(
assert data["is_permalink_state"] is True
assert data["filter_state"]["activeTabs"] == [_wrapped("TAB-2")]
assert mock_find.call_args_list[-1].args == (42,)
+ # The explicit scope applies on top of the preserved permalink state, even
+ # when it selects a tab other than the permalink's active tab.
+ if options.get("tabs_only"):
+ assert data["scope"] == {
+ "tabs_only": True,
+ "tab_id": None,
+ "untabbed_only": False,
+ }
+ assert [tab["id"] for tab in data["tab_tree"]] == ["TAB-1", "TAB-2"]
+ assert data["charts"] == []
+ elif options.get("tab"):
+ assert data["scope"] == {
+ "tabs_only": False,
+ "tab_id": "TAB-1",
+ "untabbed_only": False,
+ }
+ assert [tab["id"] for tab in data["tabs"]] == ["TAB-1"]
+ assert [chart["chart_id"] for chart in data["charts"]] == [10]
+ else:
+ assert data["scope"] is None
+ assert [chart["chart_id"] for chart in data["charts"]] == [10, 20]
@patch("superset.mcp_service.dashboard.permalink.get_dashboard_permalink")
@@ -416,3 +440,717 @@ def
test_get_dashboard_info_omitted_fields_references_layout_tool():
json_metadata_str=None, position_json_str='{"ROOT_ID": {}}'
)
assert "get_dashboard_layout" in omitted["position_json"]
+
+
[email protected]
+def large_tabbed_layout() -> str:
+ """Generate 114 nested tabs and 404 charts without storing a large
fixture."""
+ position: dict[str, Any] = {
+ "ROOT_ID": {"type": "ROOT", "children": ["TABS-root"]},
+ "TABS-root": {
+ "type": "TABS",
+ "children": [f"TAB-group-{i}" for i in range(14)],
+ },
+ }
+ for group in range(14):
+ position[f"TAB-group-{group}"] = {
+ "type": "TAB",
+ "meta": {"text": f"Group {group}"},
+ "children": [f"TABS-group-{group}"],
+ }
+ position[f"TABS-group-{group}"] = {
+ "type": "TABS",
+ "children": [f"TAB-leaf-{i}" for i in range(group, 100, 14)],
+ }
+ for leaf in range(100):
+ position[f"TAB-leaf-{leaf}"] = {
+ "type": "TAB",
+ "meta": {"text": f"Detail {leaf}"},
+ "children": [f"CHART-{i}" for i in range(leaf * 4, leaf * 4 + 4)],
+ }
+ position["ROW-shared"] = {
+ "type": "ROW",
+ "children": [f"CHART-{i}" for i in range(400, 404)],
+ }
+ position["ROOT_ID"]["children"].append("ROW-shared")
+ for chart in range(404):
+ position[f"CHART-{chart}"] = {
+ "type": "CHART",
+ "meta": {
+ "chartId": chart,
+ "sliceName": f"Chart {chart}: " + "Detailed metric description
" * 12,
+ "width": 6,
+ "height": 40,
+ },
+ }
+ return json.dumps(position)
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_tabs_only_large_tree(
+ mock_find: Mock, mcp_server: FastMCP, large_tabbed_layout: str
+) -> None:
+ """All tabs fit comfortably under the guard without any chart positions."""
+ from fastmcp.tools.tool import ToolResult
+
+ from superset.mcp_service.utils.response_size_utils import
get_response_size_bytes
+
+ mock_find.return_value =
_build_dashboard_mock(position_json=large_tabbed_layout)
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tabs_only":
True}}
+ )
+ data = json.loads(result.content[0].text)
+ assert len(data["tab_tree"]) == 114
+ assert data["tabs"] == []
+ assert data["charts"] == []
+ assert data["untabbed_chart_count"] == 4
+ assert data["scope"] == {"tabs_only": True, "tab_id": None,
"untabbed_only": False}
+ tabs = {tab["id"]: tab for tab in data["tab_tree"]}
+ assert tabs["TAB-group-0"] == {
+ "id": "TAB-group-0",
+ "name": "Group 0",
+ "parent_tab_id": None,
+ "depth": 0,
+ "chart_count": 32,
+ }
+ assert tabs["TAB-leaf-0"] == {
+ "id": "TAB-leaf-0",
+ "name": "Detail 0",
+ "parent_tab_id": "TAB-group-0",
+ "depth": 1,
+ "chart_count": 4,
+ }
+ assert all("chart_ids" not in tab for tab in data["tab_tree"])
+ # Master guards exact UTF-8 bytes; also bound the legacy bytes/4 estimate.
+ response_bytes =
get_response_size_bytes(ToolResult(content=result.content))
+ assert response_bytes < 25_000
+ assert response_bytes / 4 < 20_000 / 3
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected](
+ "tab, tab_id",
+ [
+ ("TAB-group-0", "TAB-group-0"),
+ ("Group 0", "TAB-group-0"),
+ ("TAB-leaf-0", "TAB-leaf-0"),
+ ("Detail 0", "TAB-leaf-0"),
+ ],
+)
[email protected]
+async def test_layout_tab_filter_large_tree(
+ mock_find: Mock,
+ mcp_server: FastMCP,
+ large_tabbed_layout: str,
+ tab: str,
+ tab_id: str,
+) -> None:
+ """IDs and exact titles select only a tab and its descendants."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=large_tabbed_layout)
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tab": tab}}
+ )
+ data = json.loads(result.content[0].text)
+ leaves = range(0, 100, 14) if tab_id == "TAB-group-0" else [0]
+ expected_chart_ids = {leaf * 4 + offset for leaf in leaves for offset in
range(4)}
+ assert {chart["chart_id"] for chart in data["charts"]} ==
expected_chart_ids
+ assert len(data["tabs"]) == (9 if len(expected_chart_ids) == 32 else 1)
+ assert all(
+ chart["width"] == 6 and chart["height"] == 40 for chart in
data["charts"]
+ )
+ assert data["charts"][0]["tab_path"] == ["Group 0", "Detail 0"]
+ assert data["scope"] == {
+ "tabs_only": False,
+ "tab_id": tab_id,
+ "untabbed_only": False,
+ }
+ assert data["tab_tree"] == []
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_default_response_unchanged(
+ mock_find: Mock, mcp_server: FastMCP
+) -> None:
+ """Omitting scope options preserves the complete existing response
shape."""
+ dashboard = _build_dashboard_mock(position_json=_tabbed_layout())
+ mock_find.return_value = dashboard
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1}}
+ )
+ assert json.loads(result.content[0].text) == {
+ "id": 1,
+ "dashboard_title": "Test Dashboard",
+ "uuid": "dashboard-uuid-1",
+ "has_layout": True,
+ "permalink_key": None,
+ "filter_state": None,
+ "is_permalink_state": False,
+ "tab_tree": [],
+ "untabbed_chart_count": 0,
+ "scope": None,
+ "tabs": [
+ {
+ "id": "TAB-1",
+ "name": "Overview",
+ "parent_tab_id": None,
+ "chart_ids": [10],
+ },
+ {
+ "id": "TAB-2",
+ "name": "Details",
+ "parent_tab_id": None,
+ "chart_ids": [20],
+ },
+ ],
+ "charts": [
+ {
+ "chart_id": 10,
+ "slice_name": "Top KPIs",
+ "tab_id": "TAB-1",
+ "tab_path": ["Overview"],
+ "width": 6,
+ "height": 40,
+ },
+ {
+ "chart_id": 20,
+ "slice_name": "Detail Chart",
+ "tab_id": "TAB-2",
+ "tab_path": ["Details"],
+ "width": 12,
+ "height": 60,
+ },
+ ],
+ }
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_tabs_only_with_filter(
+ mock_find: Mock, mcp_server: FastMCP, large_tabbed_layout: str
+) -> None:
+ """Summary mode composes with tab scoping and keeps absolute depth."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=large_tabbed_layout)
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "tabs_only": True, "tab": "Detail
0"}},
+ )
+ data = json.loads(result.content[0].text)
+ assert data["charts"] == []
+ assert data["tabs"] == []
+ assert len(data["tab_tree"]) == 1
+ assert data["tab_tree"][0]["depth"] == 1
+ assert data["tab_tree"][0]["parent_tab_id"] == "TAB-group-0"
+ assert data["scope"] == {
+ "tabs_only": True,
+ "tab_id": "TAB-leaf-0",
+ "untabbed_only": False,
+ }
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected](
+ "tab, error_type", [("Missing", "tab_not_found"), ("Same",
"ambiguous_tab")]
+)
[email protected]
+async def test_layout_tab_filter_errors(
+ mock_find: Mock, mcp_server: FastMCP, tab: str, error_type: str
+) -> None:
+ """Do not silently choose among duplicate titles or fall back to a full
layout."""
+ position = json.loads(_tabbed_layout())
+ position["TAB-1"]["meta"]["text"] = "Same"
+ position["TAB-2"]["meta"]["text"] = "Same"
+ mock_find.return_value =
_build_dashboard_mock(position_json=json.dumps(position))
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tab": tab}}
+ )
+ data = json.loads(result.content[0].text)
+ assert data["error_type"] == error_type
+ assert "tabs_only" in data["error"]
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_tab_id_precedes_title_and_filters_placements(
+ mock_find: Mock, mcp_server: FastMCP
+) -> None:
+ """A chart reused in another tab must not leak that tab's placement."""
+ position = json.loads(_tabbed_layout())
+ position["TAB-2"]["meta"]["text"] = "TAB-1"
+ position["CHART-b"]["meta"]["chartId"] = 10
+ mock_find.return_value =
_build_dashboard_mock(position_json=json.dumps(position))
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tab":
"TAB-1"}}
+ )
+ data = json.loads(result.content[0].text)
+ assert [tab["id"] for tab in data["tabs"]] == ["TAB-1"]
+ assert len(data["charts"]) == 1
+ assert data["charts"][0]["tab_id"] == "TAB-1"
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]("position", [None, "{}", "invalid", _simple_layout()])
[email protected]
+async def test_layout_tabs_only_without_tabs(
+ mock_find: Mock, mcp_server: FastMCP, position: str | None
+) -> None:
+ """Tab discovery without tabs offers recovery instead of an empty
layout."""
+ mock_find.return_value = _build_dashboard_mock(position_json=position)
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tabs_only":
True}}
+ )
+ data = json.loads(result.content[0].text)
+ assert data["error_type"] == "tab_not_found"
+ assert "no tabs" in data["error"]
+ assert "untabbed_only=true" in data["error"]
+ assert "full layout" in data["error"]
+ assert "charts" not in data
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]("tab", ["", " "])
[email protected]
+async def test_layout_rejects_blank_tab(
+ mock_find: Mock, mcp_server: FastMCP, tab: str
+) -> None:
+ """Blank selectors return actionable errors through the MCP client."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=_tabbed_layout())
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tab": tab}}
+ )
+ data = json.loads(result.content[0].text)
+ assert data["error_type"] == "tab_not_found"
+ assert "tab must not be blank" in data["error"]
+ assert "tabs_only=true" in data["error"]
+ assert "charts" not in data
+
+
+def _mixed_layout() -> str:
+ """A ROOT/GRID layout with one chart in a ROW above the tabs."""
+ position = json.loads(_tabbed_layout())
+ position["ROOT_ID"]["children"] = ["GRID_ID"]
+ position["GRID_ID"] = {
+ "type": "GRID",
+ "id": "GRID_ID",
+ "parents": ["ROOT_ID"],
+ "children": ["ROW-top", "TABS-1"],
+ }
+ position["ROW-top"] = {
+ "type": "ROW",
+ "id": "ROW-top",
+ "parents": ["ROOT_ID", "GRID_ID"],
+ "children": ["CHART-u"],
+ "meta": {},
+ }
+ position["CHART-u"] = {
+ "type": "CHART",
+ "id": "CHART-u",
+ "parents": ["ROOT_ID", "GRID_ID", "ROW-top"],
+ "children": [],
+ "meta": {"chartId": 30, "sliceName": "Headline KPI", "width": 12},
+ }
+ return json.dumps(position)
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_untabbed_charts_are_reachable(
+ mock_find: Mock, mcp_server: FastMCP
+) -> None:
+ """Charts outside every tab are counted in every scope and selectable
alone."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=_mixed_layout())
+ async with Client(mcp_server) as client:
+ full = json.loads(
+ (
+ await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1}}
+ )
+ )
+ .content[0]
+ .text
+ )
+ summary = json.loads(
+ (
+ await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "tabs_only": True}},
+ )
+ )
+ .content[0]
+ .text
+ )
+ untabbed = json.loads(
+ (
+ await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "untabbed_only": True}},
+ )
+ )
+ .content[0]
+ .text
+ )
+
+ assert {chart["chart_id"] for chart in full["charts"]} == {10, 20, 30}
+ assert full["untabbed_chart_count"] == 1
+ assert summary["untabbed_chart_count"] == 1
+ assert [tab["id"] for tab in summary["tab_tree"]] == ["TAB-1", "TAB-2"]
+ assert untabbed["untabbed_chart_count"] == 1
+ assert untabbed["tabs"] == []
+ assert untabbed["scope"] == {
+ "tabs_only": False,
+ "tab_id": None,
+ "untabbed_only": True,
+ }
+ assert untabbed["charts"] == [
+ {
+ "chart_id": 30,
+ "slice_name": "Headline KPI",
+ "tab_id": None,
+ "tab_path": [],
+ "width": 12,
+ "height": None,
+ }
+ ]
+ # Scoping only removes placements; it never adds one the full layout lacks.
+ assert all(chart in full["charts"] for chart in untabbed["charts"])
+
+
+def test_layout_untabbed_count_is_distinct_charts() -> None:
+ """untabbed_chart_count counts distinct charts; untabbed_only lists
placements."""
+ from superset.mcp_service.utils.response_size_utils import (
+ format_size_limit_error,
+ )
+
+ position = json.loads(_mixed_layout())
+ position["ROW-top"]["children"] = ["CHART-u", "CHART-u2", "CHART-none"]
+ position["CHART-u2"] = {
+ **position["CHART-u"],
+ "id": "CHART-u2",
+ }
+ position["CHART-none"] = {
+ "type": "CHART",
+ "id": "CHART-none",
+ "parents": ["ROOT_ID", "GRID_ID", "ROW-top"],
+ "children": [],
+ "meta": {"sliceName": "Unsaved"},
+ }
+ position_json = json.dumps(position)
+
+ full = _scoped_payload(position_json)
+ untabbed = _scoped_payload(position_json, untabbed_only=True)
+ assert full["untabbed_chart_count"] == 1
+ assert [chart["chart_id"] for chart in untabbed["charts"]] == [30, 30,
None]
+
+ message = format_size_limit_error(
+ tool_name="get_dashboard_layout",
+ params={"request": {"identifier": 1}},
+ actual_bytes=200_000,
+ max_bytes=100_000,
+ response=full,
+ )
+ assert "outside every tab (1 distinct chart)." in message
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_untabbed_only_on_untabbed_dashboard(
+ mock_find: Mock, mcp_server: FastMCP
+) -> None:
+ """On a dashboard without tabs, untabbed_only is the full chart list."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=_simple_layout())
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "untabbed_only": True}},
+ )
+ data = json.loads(result.content[0].text)
+ assert [chart["chart_id"] for chart in data["charts"]] == [42]
+ assert data["untabbed_chart_count"] == 1
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_tab_on_untabbed_dashboard_does_not_loop(
+ mock_find: Mock, mcp_server: FastMCP
+) -> None:
+ """Selecting a tab on an untabbed dashboard must not point back at
tabs_only."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=_simple_layout())
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout", {"request": {"identifier": 1, "tab":
"TAB-1"}}
+ )
+ data = json.loads(result.content[0].text)
+ assert data["error_type"] == "tab_not_found"
+ assert "no tabs" in data["error"]
+ assert "tabs_only" not in data["error"]
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected](
+ "options", [{"tabs_only": True}, {"tab": "TAB-1"}, {"tab": "X",
"tabs_only": True}]
+)
[email protected]
+async def test_layout_rejects_untabbed_only_with_tab_options(
+ mock_find: Mock, mcp_server: FastMCP, options: dict[str, Any]
+) -> None:
+ """Conflicting scopes reach the client as actionable errors, not
validation text."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=_mixed_layout())
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "untabbed_only": True, **options}},
+ )
+ data = json.loads(result.content[0].text)
+ assert data["error_type"] == "invalid_scope"
+ assert "untabbed_only cannot be combined" in data["error"]
+ assert "Drop tab and tabs_only" in data["error"]
+ assert "drop untabbed_only" in data["error"]
+ assert "charts" not in data
+
+
+def _scoped_payload(position_json: str, **options: Any) -> dict[str, Any]:
+ from superset.mcp_service.dashboard.schemas import (
+ dashboard_layout_serializer,
+ DashboardLayout,
+ GetDashboardLayoutRequest,
+ )
+ from superset.mcp_service.dashboard.tool.get_dashboard_layout import (
+ _scope_layout,
+ )
+
+ layout = _scope_layout(
+
dashboard_layout_serializer(_build_dashboard_mock(position_json=position_json)),
+ GetDashboardLayoutRequest(identifier=1, **options),
+ )
+ assert isinstance(layout, DashboardLayout)
+ return layout.model_dump(mode="json")
+
+
[email protected](
+ "layout_name, options, expected, unexpected",
+ [
+ (
+ "large",
+ {},
+ [
+ "tabs_only=true",
+ 'tab="<ID or title>"',
+ "untabbed_only=true",
+ "outside every tab (4 distinct charts)",
+ ],
+ [],
+ ),
+ (
+ "tabbed",
+ {},
+ ["tabs_only=true", 'tab="<ID or title>"'],
+ ["untabbed_only"],
+ ),
+ (
+ "simple",
+ {},
+ ["no tabs", "list_charts", "'dashboards'", "value: 1"],
+ ["tabs_only=true", "untabbed_only"],
+ ),
+ (
+ "large",
+ {"tabs_only": True},
+ ["Keep tabs_only=true", '"TAB-group-0"', '"TAB-group-4", ...'],
+ ["TAB-leaf"],
+ ),
+ (
+ "large",
+ {"tab": "TAB-group-0"},
+ ['"TAB-leaf-0"', "nested tab ID", "Add tabs_only=true"],
+ ["no nested tabs"],
+ ),
+ (
+ "large",
+ {"tab": "TAB-group-0", "tabs_only": True},
+ ['"TAB-leaf-0"', "nested tab ID"],
+ ["Add tabs_only=true", "no nested tabs"],
+ ),
+ (
+ "large",
+ {"tab": "TAB-leaf-0"},
+ ["no nested tabs", "tab cannot narrow it", "list_charts"],
+ ["Pass a nested tab ID", 'tab="<ID or title>"'],
+ ),
+ (
+ "large",
+ {"tab": "Detail 0", "tabs_only": True},
+ ["no nested tabs", "cannot be narrowed further by tab"],
+ ["Pass a nested tab ID", "Keep tabs_only=true"],
+ ),
+ (
+ "mixed",
+ {"untabbed_only": True},
+ ["cannot be narrowed further", "list_charts"],
+ ["tabs_only=true", "Pass a nested tab ID"],
+ ),
+ ],
+)
+def test_layout_oversized_hint_follows_scope(
+ large_tabbed_layout: str,
+ layout_name: str,
+ options: dict[str, Any],
+ expected: list[str],
+ unexpected: list[str],
+) -> None:
+ """Each oversized scope suggests only a narrower scope that actually
exists."""
+ from superset.mcp_service.utils.response_size_utils import (
+ format_size_limit_error,
+ )
+
+ position_json = {
+ "large": large_tabbed_layout,
+ "tabbed": _tabbed_layout(),
+ "simple": _simple_layout(),
+ "mixed": _mixed_layout(),
+ }[layout_name]
+ message = format_size_limit_error(
+ tool_name="get_dashboard_layout",
+ params={"request": {"identifier": 1, **options}},
+ actual_bytes=200_000,
+ max_bytes=100_000,
+ response=_scoped_payload(position_json, **options),
+ )
+ for text in expected:
+ assert text in message
+ for text in unexpected:
+ assert text not in message
+
+
[email protected](
+ "options, expected",
+ [
+ ({}, ["tabs_only=true", "untabbed_only=true", "outside every tab."]),
+ ({"tabs_only": True}, ['pass tab="<top-level tab ID>"']),
+ ({"tab": "TAB-1"}, ["If this tab has nested tabs", "list_charts"]),
+ ({"untabbed_only": True}, ["cannot be narrowed further"]),
+ ],
+)
+def test_layout_oversized_hint_without_payload(
+ options: dict[str, Any], expected: list[str]
+) -> None:
+ """When the payload is unavailable, the hint relies on the request
alone."""
+ from superset.mcp_service.utils.response_size_utils import (
+ format_size_limit_error,
+ )
+
+ message = format_size_limit_error(
+ tool_name="get_dashboard_layout",
+ params={"request": {"identifier": 1, **options}},
+ actual_bytes=200_000,
+ max_bytes=100_000,
+ )
+ for text in expected:
+ assert text in message
+
+
[email protected](
+ "layout_name, expected, unexpected",
+ [
+ ("large", ["tabs_only=true", 'tab="<ID or title>"'], ["no tabs"]),
+ ("simple", ["no tabs", "list_charts"], ["tabs_only=true"]),
+ ],
+)
[email protected]
+async def test_layout_oversized_guard_hint(
+ large_tabbed_layout: str,
+ layout_name: str,
+ expected: list[str],
+ unexpected: list[str],
+) -> None:
+ """The real guard hands the blocked layout to the hint, so it knows the
tabs."""
+ from unittest.mock import AsyncMock
+
+ from fastmcp.exceptions import ToolError
+ from fastmcp.tools.tool import ToolResult
+ from mcp.types import TextContent
+
+ from superset.mcp_service.dashboard.schemas import
dashboard_layout_serializer
+ from superset.mcp_service.middleware import ResponseSizeGuardMiddleware
+
+ position_json = large_tabbed_layout if layout_name == "large" else
_simple_layout()
+ layout = dashboard_layout_serializer(
+ _build_dashboard_mock(position_json=position_json)
+ )
+ response = ToolResult(
+ content=[TextContent(type="text", text=layout.model_dump_json())]
+ )
+ middleware = ResponseSizeGuardMiddleware(max_bytes=100)
+ context = Mock()
+ context.message.name = "get_dashboard_layout"
+ context.message.arguments = {"request": {"identifier": 1}}
+ with (
+ patch("superset.mcp_service.middleware.get_user_id", return_value=1),
+ patch("superset.mcp_service.middleware.event_logger"),
+ pytest.raises(ToolError) as exc_info,
+ ):
+ await middleware.on_call_tool(context,
AsyncMock(return_value=response))
+ message = str(exc_info.value)
+ assert "Response too large" in message
+ for text in expected:
+ assert text in message
+ for text in unexpected:
+ assert text not in message
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]
+async def test_layout_logs_scoped_counts(
+ mock_find: Mock, mcp_server: FastMCP, large_tabbed_layout: str
+) -> None:
+ """The retrieval log reports what was returned, not the unscoped layout."""
+ mock_find.return_value =
_build_dashboard_mock(position_json=large_tabbed_layout)
+ messages: list[str] = []
+
+ async def log_handler(message: Any) -> None:
+ data = message.data
+ messages.append(data.get("msg", "") if isinstance(data, dict) else
str(data))
+
+ async with Client(mcp_server, log_handler=log_handler) as client:
+ await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "tab": "TAB-leaf-0"}},
+ )
+ retrieved = [m for m in messages if "Dashboard layout retrieved" in m]
+ assert len(retrieved) == 1
+ assert "tab_count=1," in retrieved[0]
+ assert "chart_count=4," in retrieved[0]
+ assert "'tab_id': 'TAB-leaf-0'" in retrieved[0]
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
[email protected]("tabs_only", [False, True])
[email protected]
+async def test_layout_empty_tab(
+ mock_find: Mock, mcp_server: FastMCP, tabs_only: bool
+) -> None:
+ """A valid empty tab remains discoverable and can be selected."""
+ position = json.loads(_tabbed_layout())
+ position["TAB-1"]["children"] = []
+ mock_find.return_value =
_build_dashboard_mock(position_json=json.dumps(position))
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "get_dashboard_layout",
+ {"request": {"identifier": 1, "tab": "TAB-1", "tabs_only":
tabs_only}},
+ )
+ data = json.loads(result.content[0].text)
+ assert data["charts"] == []
+ if tabs_only:
+ assert len(data["tab_tree"]) == 1
+ assert data["tab_tree"][0]["chart_count"] == 0
+ else:
+ assert len(data["tabs"]) == 1
+ assert data["tabs"][0]["chart_ids"] == []
