feat(mcp): add duplicate_dashboard tool

Adds a duplicate_dashboard MCP tool that clones an existing dashboard
via CopyDashboardCommand. The source dashboard can be identified by
numeric ID, UUID, or slug. By default the copy references the same
charts; duplicate_slices=true deep-copies every chart into new objects
owned by the caller.

The tool builds the required json_metadata payload (source metadata
plus a positions key from position_json), mirroring what the frontend
"Save as" flow sends to the /copy/ endpoint. The new title is
sanitized for XSS, and the tool is excluded from MCP response caching.

helm/superset/Chart.yaml

@@ -29,7 +29,7 @@ maintainers:
   - name: craig-rueda
     email: craig@craigrueda.com
     url: https://github.com/craig-rueda
-version: 0.17.0 # See [README](https://github.com/apache/superset/blob/master/helm/superset/README.md#versioning) for version details.
+version: 0.16.0 # See [README](https://github.com/apache/superset/blob/master/helm/superset/README.md#versioning) for version details.
 dependencies:
   - name: postgresql
     version: 16.7.27

helm/superset/README.md

@@ -23,7 +23,7 @@ NOTE: This file is generated by helm-docs: https://github.com/norwoodj/helm-docs
 
 # superset
 
-![Version: 0.17.0](https://img.shields.io/badge/Version-0.17.0-informational?style=flat-square)
+![Version: 0.16.0](https://img.shields.io/badge/Version-0.16.0-informational?style=flat-square)
 
 Apache Superset is a modern, enterprise-ready business intelligence web application
 
@@ -111,6 +111,9 @@ On helm this can be set on `extraSecretEnv.SUPERSET_SECRET_KEY` or `configOverri
 | init.resources | object | `{}` |  |
 | init.tolerations | list | `[]` |  |
 | init.topologySpreadConstraints | list | `[]` | TopologySpreadConstrains to be added to init job |
+| initImage.pullPolicy | string | `"IfNotPresent"` |  |
+| initImage.repository | string | `"apache/superset"` |  |
+| initImage.tag | string | `"dockerize"` |  |
 | nameOverride | string | `nil` | Provide a name to override the name of the chart |
 | nodeSelector | object | `{}` |  |
 | postgresql | object | see `values.yaml` | Configuration values for the postgresql dependency. ref: https://github.com/bitnami/charts/tree/main/bitnami/postgresql |

helm/superset/values.yaml

@@ -194,6 +194,11 @@ image:
 
 imagePullSecrets: []
 
+initImage:
+  repository: apache/superset
+  tag: dockerize
+  pullPolicy: IfNotPresent
+
 service:
   type: ClusterIP
   port: 8088
@@ -298,28 +303,15 @@ supersetNode:
   # @default -- a container waiting for postgres
   initContainers:
     - name: wait-for-postgres
-      image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-      imagePullPolicy: "{{ .Values.image.pullPolicy }}"
+      image: "{{ .Values.initImage.repository }}:{{ .Values.initImage.tag }}"
+      imagePullPolicy: "{{ .Values.initImage.pullPolicy }}"
       envFrom:
         - secretRef:
             name: "{{ tpl .Values.envFromSecret . }}"
       command:
-        - /bin/bash
+        - /bin/sh
         - -c
-        - |
-          # bash's /dev/tcp redirect performs a TCP connect; no external
-          # `dockerize`, `nc`, or busybox needed. SECONDS-based deadline
-          # mirrors the prior `dockerize -timeout 120s` behaviour.
-          SECONDS=0
-          until (echo > /dev/tcp/"$DB_HOST"/"$DB_PORT") 2>/dev/null; do
-            if [ "$SECONDS" -ge 120 ]; then
-              echo "timeout waiting for postgres at $DB_HOST:$DB_PORT after 120s" >&2
-              exit 1
-            fi
-            echo "waiting for postgres at $DB_HOST:$DB_PORT (elapsed ${SECONDS}s)"
-            sleep 2
-          done
-          echo "postgres at $DB_HOST:$DB_PORT is up"
+        - dockerize -wait "tcp://$DB_HOST:$DB_PORT" -timeout 120s
       resources:
         limits:
           memory: "256Mi"
@@ -415,31 +407,15 @@ supersetWorker:
   # @default -- a container waiting for postgres and redis
   initContainers:
     - name: wait-for-postgres-redis
-      image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-      imagePullPolicy: "{{ .Values.image.pullPolicy }}"
+      image: "{{ .Values.initImage.repository }}:{{ .Values.initImage.tag }}"
+      imagePullPolicy: "{{ .Values.initImage.pullPolicy }}"
       envFrom:
         - secretRef:
             name: "{{ tpl .Values.envFromSecret . }}"
       command:
-        - /bin/bash
+        - /bin/sh
         - -c
-        - |
-          # See supersetNode.initContainers for the rationale.
-          SECONDS=0
-          wait_for() {
-            local host=$1 port=$2 name=$3
-            until (echo > /dev/tcp/"$host"/"$port") 2>/dev/null; do
-              if [ "$SECONDS" -ge 120 ]; then
-                echo "timeout waiting for $name at $host:$port after 120s" >&2
-                exit 1
-              fi
-              echo "waiting for $name at $host:$port (elapsed ${SECONDS}s)"
-              sleep 2
-            done
-            echo "$name at $host:$port is up"
-          }
-          wait_for "$DB_HOST" "$DB_PORT" postgres
-          wait_for "$REDIS_HOST" "$REDIS_PORT" redis
+        - dockerize -wait "tcp://$DB_HOST:$DB_PORT" -wait "tcp://$REDIS_HOST:$REDIS_PORT" -timeout 120s
       resources:
         limits:
           memory: "256Mi"
@@ -519,31 +495,15 @@ supersetCeleryBeat:
   # @default -- a container waiting for postgres
   initContainers:
     - name: wait-for-postgres-redis
-      image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-      imagePullPolicy: "{{ .Values.image.pullPolicy }}"
+      image: "{{ .Values.initImage.repository }}:{{ .Values.initImage.tag }}"
+      imagePullPolicy: "{{ .Values.initImage.pullPolicy }}"
       envFrom:
         - secretRef:
             name: "{{ tpl .Values.envFromSecret . }}"
       command:
-        - /bin/bash
+        - /bin/sh
         - -c
-        - |
-          # See supersetNode.initContainers for the rationale.
-          SECONDS=0
-          wait_for() {
-            local host=$1 port=$2 name=$3
-            until (echo > /dev/tcp/"$host"/"$port") 2>/dev/null; do
-              if [ "$SECONDS" -ge 120 ]; then
-                echo "timeout waiting for $name at $host:$port after 120s" >&2
-                exit 1
-              fi
-              echo "waiting for $name at $host:$port (elapsed ${SECONDS}s)"
-              sleep 2
-            done
-            echo "$name at $host:$port is up"
-          }
-          wait_for "$DB_HOST" "$DB_PORT" postgres
-          wait_for "$REDIS_HOST" "$REDIS_PORT" redis
+        - dockerize -wait "tcp://$DB_HOST:$DB_PORT" -wait "tcp://$REDIS_HOST:$REDIS_PORT" -timeout 120s
       resources:
         limits:
           memory: "256Mi"
@@ -634,31 +594,15 @@ supersetCeleryFlower:
   # @default -- a container waiting for postgres and redis
   initContainers:
     - name: wait-for-postgres-redis
-      image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-      imagePullPolicy: "{{ .Values.image.pullPolicy }}"
+      image: "{{ .Values.initImage.repository }}:{{ .Values.initImage.tag }}"
+      imagePullPolicy: "{{ .Values.initImage.pullPolicy }}"
       envFrom:
         - secretRef:
             name: "{{ tpl .Values.envFromSecret . }}"
       command:
-        - /bin/bash
+        - /bin/sh
         - -c
-        - |
-          # See supersetNode.initContainers for the rationale.
-          SECONDS=0
-          wait_for() {
-            local host=$1 port=$2 name=$3
-            until (echo > /dev/tcp/"$host"/"$port") 2>/dev/null; do
-              if [ "$SECONDS" -ge 120 ]; then
-                echo "timeout waiting for $name at $host:$port after 120s" >&2
-                exit 1
-              fi
-              echo "waiting for $name at $host:$port (elapsed ${SECONDS}s)"
-              sleep 2
-            done
-            echo "$name at $host:$port is up"
-          }
-          wait_for "$DB_HOST" "$DB_PORT" postgres
-          wait_for "$REDIS_HOST" "$REDIS_PORT" redis
+        - dockerize -wait "tcp://$DB_HOST:$DB_PORT" -wait "tcp://$REDIS_HOST:$REDIS_PORT" -timeout 120s
       resources:
         limits:
           memory: "256Mi"
@@ -820,26 +764,15 @@ init:
   # @default -- a container waiting for postgres
   initContainers:
     - name: wait-for-postgres
-      image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-      imagePullPolicy: "{{ .Values.image.pullPolicy }}"
+      image: "{{ .Values.initImage.repository }}:{{ .Values.initImage.tag }}"
+      imagePullPolicy: "{{ .Values.initImage.pullPolicy }}"
       envFrom:
         - secretRef:
             name: "{{ tpl .Values.envFromSecret . }}"
       command:
-        - /bin/bash
+        - /bin/sh
         - -c
-        - |
-          # See supersetNode.initContainers for the rationale.
-          SECONDS=0
-          until (echo > /dev/tcp/"$DB_HOST"/"$DB_PORT") 2>/dev/null; do
-            if [ "$SECONDS" -ge 120 ]; then
-              echo "timeout waiting for postgres at $DB_HOST:$DB_PORT after 120s" >&2
-              exit 1
-            fi
-            echo "waiting for postgres at $DB_HOST:$DB_PORT (elapsed ${SECONDS}s)"
-            sleep 2
-          done
-          echo "postgres at $DB_HOST:$DB_PORT is up"
+        - dockerize -wait "tcp://$DB_HOST:$DB_PORT" -timeout 120s
       resources:
         limits:
           memory: "256Mi"

superset/mcp_service/app.py

@@ -129,6 +129,7 @@ Dashboard Management:
 - get_dashboard_info: Get detailed dashboard information by ID
 - get_dashboard_layout: Get parsed tabs and chart positions for a dashboard (companion to get_dashboard_info when its omitted_fields hint flags position_json)
 - generate_dashboard: Create a dashboard from chart IDs (requires write access)
+- duplicate_dashboard: Duplicate an existing dashboard, optionally deep-copying its charts (requires write access)
 - add_chart_to_existing_dashboard: Add a chart to an existing dashboard (requires write access)
 
 Annotation Layers:
@@ -413,8 +414,9 @@ Input format:
 {_feature_availability}Permission Awareness:
 {_instance_info_role_bullet}- ALWAYS check the user's roles BEFORE suggesting write operations (creating datasets,
   charts, or dashboards). SQL execution is a separate permission — see execute_sql below.
-- Write tools (generate_chart, generate_dashboard, update_chart, create_virtual_dataset,
-  save_sql_query, add_chart_to_existing_dashboard, update_chart_preview) require write
+- Write tools (generate_chart, generate_dashboard, duplicate_dashboard, update_chart,
+  create_virtual_dataset, save_sql_query, add_chart_to_existing_dashboard,
+  update_chart_preview) require write
   permissions. These tools are only listed for users who have the necessary access.
   If a write tool does not appear in the tool list, the current user lacks write access.
 - execute_sql requires SQL Lab access (execute_sql_query permission), which is separate
@@ -679,6 +681,7 @@ from superset.mcp_service.chart.tool import (  # noqa: F401, E402
 )
 from superset.mcp_service.dashboard.tool import (  # noqa: F401, E402
     add_chart_to_existing_dashboard,
+    duplicate_dashboard,
     generate_dashboard,
     get_dashboard_info,
     get_dashboard_layout,

superset/mcp_service/dashboard/schemas.py

@@ -708,6 +708,138 @@ class GenerateDashboardResponse(BaseModel):
     )
 
 
+class DuplicateDashboardRequest(BaseModel):
+    """Request schema for duplicating an existing dashboard."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    dashboard_id: Annotated[
+        int | str,
+        Field(
+            description=(
+                "Source dashboard identifier - can be numeric ID, UUID string, or slug"
+            )
+        ),
+    ]
+    dashboard_title: str = Field(
+        ...,
+        description="Title for the new (duplicated) dashboard",
+        validation_alias=AliasChoices("dashboard_title", "title", "name"),
+    )
+    duplicate_slices: bool = Field(
+        default=False,
+        description=(
+            "When true, every chart on the source dashboard is deep-copied "
+            "into a new chart object owned by the caller. When false "
+            "(default), the new dashboard references the same charts as the "
+            "source."
+        ),
+    )
+    sanitization_warnings: List[str] = Field(
+        default_factory=list,
+        description=(
+            "Internal: warnings emitted when user input was altered by "
+            "sanitization. Populated by the ``mode='before'`` validator "
+            "before dashboard_title is rewritten, so the tool can surface "
+            "a notice to the caller instead of silently dropping content."
+        ),
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def _detect_dashboard_title_sanitization(cls, data: Any) -> Any:
+        """Reject empty-after-sanitization titles and warn on partial strip.
+
+        Runs before the ``dashboard_title`` field validator rewrites the
+        value. If the caller supplied a title that sanitization would strip
+        entirely (XSS-only content), we raise so the caller gets a clear
+        error instead of a blank-titled dashboard. When the sanitizer only
+        trims part of the title, we record a warning the tool can return
+        alongside the successful result.
+
+        ``sanitization_warnings`` is a server-only field — any value the
+        caller supplied is discarded here so the tool cannot be tricked
+        into echoing attacker-controlled text back through the response.
+        """
+        if not isinstance(data, dict):
+            return data
+        data["sanitization_warnings"] = []
+        for key in ("dashboard_title", "title", "name"):
+            if key in data:
+                raw = data[key]
+                break
+        else:
+            raw = None
+        if not isinstance(raw, str) or not raw.strip():
+            return data
+        sanitized, was_modified = sanitize_user_input_with_changes(
+            raw, "Dashboard title", max_length=500, allow_empty=True
+        )
+        if was_modified and not sanitized:
+            raise ValueError(
+                "dashboard_title contained only disallowed content "
+                "(HTML/script/URL schemes) and was removed entirely by "
+                "sanitization. Provide a dashboard_title with plain text."
+            )
+        if was_modified:
+            data["sanitization_warnings"].append(
+                "dashboard_title was modified during sanitization to "
+                "remove potentially unsafe content; the stored title "
+                "differs from the input."
+            )
+        return data
+
+    @field_validator("dashboard_title")
+    @classmethod
+    def sanitize_dashboard_title(cls, v: str) -> str:
+        """Sanitize dashboard title to prevent XSS."""
+        sanitized = sanitize_user_input(
+            v, "Dashboard title", max_length=500, allow_empty=True
+        )
+        if not sanitized:
+            raise ValueError("dashboard_title cannot be empty")
+        return sanitized
+
+
+class DuplicateDashboardResponse(BaseModel):
+    """Response schema for dashboard duplication."""
+
+    dashboard: DashboardInfo | None = Field(
+        None, description="The newly created dashboard info, if successful"
+    )
+    dashboard_url: str | None = Field(None, description="URL to view the new dashboard")
+    duplicated_slices: bool = Field(
+        default=False,
+        description=(
+            "True when the source dashboard's charts were deep-copied into "
+            "new chart objects; False when the new dashboard references the "
+            "original charts."
+        ),
+    )
+    error: str | None = Field(None, description="Error message, if duplication failed")
+    warnings: List[str] = Field(
+        default_factory=list,
+        description=(
+            "Non-fatal advisory messages about the duplicated dashboard — "
+            "for example, that the supplied title was altered by "
+            "sanitization."
+        ),
+    )
+
+    @field_validator("error")
+    @classmethod
+    def sanitize_error_for_llm_context(cls, value: str | None) -> str | None:
+        """Wrap error text before it is exposed to LLM context.
+
+        The error may echo dashboard-controlled content such as the source
+        dashboard title — wrap it so the LLM treats it as data, not
+        instructions.
+        """
+        if value is None:
+            return value
+        return sanitize_for_llm_context(value, field_path=("error",))
+
+
 class ChartPosition(BaseModel):
     """Position and identity of a chart within a dashboard layout."""
 

superset/mcp_service/dashboard/tool/__init__.py

@@ -16,6 +16,7 @@
 # under the License.
 
 from .add_chart_to_existing_dashboard import add_chart_to_existing_dashboard
+from .duplicate_dashboard import duplicate_dashboard
 from .generate_dashboard import generate_dashboard
 from .get_dashboard_info import get_dashboard_info
 from .get_dashboard_layout import get_dashboard_layout
@@ -26,5 +27,6 @@ __all__ = [
     "get_dashboard_info",
     "get_dashboard_layout",
     "generate_dashboard",
+    "duplicate_dashboard",
     "add_chart_to_existing_dashboard",
 ]

superset/mcp_service/dashboard/tool/duplicate_dashboard.py

@@ -0,0 +1,269 @@
+# 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.
+
+"""
+MCP tool: duplicate_dashboard
+
+Duplicates an existing dashboard, optionally deep-copying its charts.
+Canonical workflow: clone a template dashboard, then edit the copy
+(e.g. to create a regional or staging variant).
+"""
+
+import logging
+from typing import Any
+
+from fastmcp import Context
+from sqlalchemy.exc import SQLAlchemyError
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.dashboard.schemas import (
+    DashboardInfo,
+    DuplicateDashboardRequest,
+    DuplicateDashboardResponse,
+    serialize_chart_summary,
+)
+from superset.mcp_service.privacy import user_can_view_data_model_metadata
+from superset.mcp_service.utils.url_utils import get_superset_base_url
+from superset.utils import json
+
+logger = logging.getLogger(__name__)
+
+
+def _build_copy_payload(
+    source: Any, dashboard_title: str, duplicate_slices: bool
+) -> dict[str, Any]:
+    """Build the data payload expected by ``CopyDashboardCommand``.
+
+    Mirrors what the frontend "Save as" flow sends to the
+    ``/api/v1/dashboard/<id>/copy/`` endpoint: the source dashboard's
+    current ``json_metadata`` with a ``positions`` key holding the current
+    layout (``position_json``). ``DashboardCopySchema`` requires
+    ``json_metadata``, and ``DashboardDAO.copy_dashboard`` reads
+    ``positions`` from it to remap chart IDs when ``duplicate_slices``
+    is enabled.
+    """
+    try:
+        metadata = json.loads(source.json_metadata or "{}")
+    except (json.JSONDecodeError, TypeError):
+        metadata = {}
+    if not isinstance(metadata, dict):
+        metadata = {}
+
+    try:
+        positions = json.loads(source.position_json or "{}")
+    except (json.JSONDecodeError, TypeError):
+        positions = {}
+    if not isinstance(positions, dict):
+        positions = {}
+
+    metadata["positions"] = positions
+
+    return {
+        "dashboard_title": dashboard_title,
+        "css": source.css,
+        "duplicate_slices": duplicate_slices,
+        "json_metadata": json.dumps(metadata),
+    }
+
+
+def _serialize_new_dashboard(dashboard: Any) -> tuple[DashboardInfo, str]:
+    """Build the response ``DashboardInfo`` and URL for the new dashboard."""
+    from superset.mcp_service.dashboard.schemas import serialize_tag_object
+
+    dashboard_url = f"{get_superset_base_url()}/superset/dashboard/{dashboard.id}/"
+    include_data_model_metadata = user_can_view_data_model_metadata()
+    info = DashboardInfo(
+        id=dashboard.id,
+        dashboard_title=dashboard.dashboard_title,
+        slug=dashboard.slug,
+        description=dashboard.description,
+        published=dashboard.published,
+        created_on=dashboard.created_on,
+        changed_on=dashboard.changed_on,
+        uuid=str(dashboard.uuid) if dashboard.uuid else None,
+        url=dashboard_url,
+        chart_count=len(dashboard.slices),
+        tags=[
+            obj
+            for tag in getattr(dashboard, "tags", [])
+            if (obj := serialize_tag_object(tag)) is not None
+        ],
+        charts=[
+            obj
+            for chart in getattr(dashboard, "slices", [])
+            if (
+                obj := serialize_chart_summary(
+                    chart,
+                    include_data_model_metadata=include_data_model_metadata,
+                )
+            )
+            is not None
+        ],
+    )
+    return info, dashboard_url
+
+
+@tool(
+    tags=["mutate"],
+    class_permission_name="Dashboard",
+    method_permission_name="write",
+    annotations=ToolAnnotations(
+        title="Duplicate dashboard",
+        readOnlyHint=False,
+        destructiveHint=False,
+    ),
+)
+async def duplicate_dashboard(
+    request: DuplicateDashboardRequest, ctx: Context
+) -> DuplicateDashboardResponse:
+    """
+    Duplicate an existing dashboard under a new title.
+
+    By default the copy references the same charts as the source.
+    Set duplicate_slices=true to also deep-copy every chart into new
+    chart objects owned by you, so edits to the copies never affect
+    the originals.
+
+    The source dashboard can be identified by numeric ID, UUID, or slug.
+    Returns the new dashboard's ID, title, and URL.
+    """
+    await ctx.info(
+        "Duplicating dashboard: dashboard_id=%s, duplicate_slices=%s"
+        % (request.dashboard_id, request.duplicate_slices)
+    )
+
+    from superset.commands.dashboard.copy import CopyDashboardCommand
+    from superset.commands.dashboard.exceptions import (
+        DashboardAccessDeniedError,
+        DashboardCopyError,
+        DashboardForbiddenError,
+        DashboardInvalidError,
+        DashboardNotFoundError,
+    )
+    from superset.daos.dashboard import DashboardDAO
+
+    try:
+        with event_logger.log_context(action="mcp.duplicate_dashboard.lookup"):
+            try:
+                source = DashboardDAO.get_by_id_or_slug(str(request.dashboard_id))
+            except DashboardNotFoundError:
+                return DuplicateDashboardResponse(
+                    error=(
+                        f"Dashboard '{request.dashboard_id}' not found. "
+                        "Use list_dashboards to get valid dashboard IDs."
+                    ),
+                )
+            except DashboardAccessDeniedError:
+                return DuplicateDashboardResponse(
+                    error=(
+                        f"You don't have access to dashboard "
+                        f"'{request.dashboard_id}', so it cannot be duplicated."
+                    ),
+                )
+
+        data = _build_copy_payload(
+            source, request.dashboard_title, request.duplicate_slices
+        )
+
+        with event_logger.log_context(action="mcp.duplicate_dashboard.copy"):
+            new_dashboard = CopyDashboardCommand(source, data).run()
+
+        # Re-fetch with eager-loaded relationships to avoid lazy-loading on
+        # a session that the command's commit may have invalidated.
+        from sqlalchemy.orm import subqueryload
+
+        from superset.models.dashboard import Dashboard
+        from superset.models.slice import Slice
+
+        try:
+            new_dashboard = (
+                DashboardDAO.find_by_id(
+                    new_dashboard.id,
+                    query_options=[
+                        subqueryload(Dashboard.slices).subqueryload(Slice.tags),
+                        subqueryload(Dashboard.tags),
+                    ],
+                )
+                or new_dashboard
+            )
+            info, dashboard_url = _serialize_new_dashboard(new_dashboard)
+        except SQLAlchemyError:
+            logger.warning(
+                "Re-fetch of dashboard %s failed; returning minimal response",
+                new_dashboard.id,
+                exc_info=True,
+            )
+            dashboard_url = (
+                f"{get_superset_base_url()}/superset/dashboard/{new_dashboard.id}/"
+            )
+            info = DashboardInfo(
+                id=new_dashboard.id,
+                dashboard_title=request.dashboard_title,
+                url=dashboard_url,
+            )
+
+        logger.info(
+            "Duplicated dashboard %s into dashboard %s (duplicate_slices=%s)",
+            request.dashboard_id,
+            new_dashboard.id,
+            request.duplicate_slices,
+        )
+
+        return DuplicateDashboardResponse(
+            dashboard=info,
+            dashboard_url=dashboard_url,
+            duplicated_slices=request.duplicate_slices,
+            warnings=list(request.sanitization_warnings),
+        )
+
+    except DashboardForbiddenError:
+        await ctx.error(
+            "Dashboard duplication forbidden: dashboard_id=%s" % (request.dashboard_id,)
+        )
+        return DuplicateDashboardResponse(
+            error=(
+                f"You don't have permission to duplicate dashboard "
+                f"'{request.dashboard_id}'."
+            ),
+        )
+    except DashboardInvalidError:
+        return DuplicateDashboardResponse(
+            error=(
+                "Dashboard duplication parameters were invalid. "
+                "Provide a non-empty dashboard_title."
+            ),
+        )
+    except DashboardCopyError as exc:
+        from superset import db
+
+        try:
+            db.session.rollback()  # pylint: disable=consider-using-transaction
+        except SQLAlchemyError:
+            logger.warning(
+                "Database rollback failed during error handling", exc_info=True
+            )
+        await ctx.error("Dashboard duplication failed: %s" % (str(exc),))
+        return DuplicateDashboardResponse(
+            error=f"Failed to duplicate dashboard: {exc}",
+        )
+    except Exception as exc:
+        await ctx.error(
+            "Unexpected error duplicating dashboard: %s: %s"
+            % (type(exc).__name__, str(exc))
+        )
+        raise

superset/mcp_service/mcp_config.py

@@ -212,6 +212,7 @@ MCP_CACHE_CONFIG: Dict[str, Any] = {
     "excluded_tools": [  # Tools that should never be cached (side effects, dynamic)
         "execute_sql",
         "generate_dashboard",
+        "duplicate_dashboard",
         "generate_chart",
         "update_chart",
     ],

tests/unit_tests/mcp_service/dashboard/tool/test_duplicate_dashboard.py

@@ -0,0 +1,361 @@
+# 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.
+
+"""
+Unit tests for the duplicate_dashboard MCP tool.
+
+Follows the same pattern used in test_add_chart_to_existing_dashboard.py:
+- Tests run through the async MCP Client (not direct function calls)
+- Patches applied at source locations (superset.daos.dashboard.*,
+  superset.commands.dashboard.copy.*)
+- auth is mocked via the autouse mock_auth fixture
+
+Covers:
+- Duplicate referencing the same charts (duplicate_slices=False, default)
+- Duplicate with deep-copied charts (duplicate_slices=True)
+- Source dashboard not found
+- Source dashboard access denied / copy forbidden
+- Title sanitization (XSS stripped, XSS-only title rejected)
+"""
+
+import logging
+from unittest.mock import Mock, patch
+
+import pytest
+from fastmcp import Client
+
+from superset.mcp_service.app import mcp
+from superset.utils import json
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def mcp_server() -> object:
+    """Return the FastMCP app instance for use in MCP client tests."""
+    return mcp
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    """Mock authentication for all tests."""
+    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
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+SOURCE_POSITIONS = {
+    "DASHBOARD_VERSION_KEY": "v2",
+    "ROOT_ID": {"children": ["GRID_ID"], "id": "ROOT_ID", "type": "ROOT"},
+    "GRID_ID": {
+        "children": ["CHART-10"],
+        "id": "GRID_ID",
+        "parents": ["ROOT_ID"],
+        "type": "GRID",
+    },
+    "CHART-10": {
+        "children": [],
+        "id": "CHART-10",
+        "meta": {"chartId": 10, "height": 50, "width": 4},
+        "parents": ["ROOT_ID", "GRID_ID"],
+        "type": "CHART",
+    },
+}
+
+
+def _mock_chart(id: int = 10, slice_name: str = "Test Chart") -> Mock:
+    """Create a minimal mock Slice object with the given ID and name."""
+    chart = Mock()
+    chart.id = id
+    chart.slice_name = slice_name
+    chart.uuid = f"chart-uuid-{id}"
+    chart.tags = []
+    chart.owners = []
+    chart.viz_type = "table"
+    chart.datasource_name = None
+    chart.description = None
+    return chart
+
+
+def _mock_dashboard(
+    id: int = 1,
+    title: str = "Sales Dashboard",
+    slices: list[Mock] | None = None,
+    json_metadata: str | None = None,
+    position_json: str | None = None,
+) -> Mock:
+    """Create a minimal mock Dashboard object."""
+    dashboard = Mock()
+    dashboard.id = id
+    dashboard.dashboard_title = title
+    dashboard.slug = f"test-dashboard-{id}"
+    dashboard.description = None
+    dashboard.published = True
+    dashboard.created_on = None
+    dashboard.changed_on = None
+    dashboard.uuid = f"dashboard-uuid-{id}"
+    dashboard.slices = slices or []
+    dashboard.owners = []
+    dashboard.tags = []
+    dashboard.roles = []
+    dashboard.position_json = position_json or json.dumps(SOURCE_POSITIONS)
+    dashboard.json_metadata = json_metadata
+    dashboard.css = None
+    dashboard.certified_by = None
+    dashboard.certification_details = None
+    dashboard.is_managed_externally = False
+    dashboard.external_url = None
+    return dashboard
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_duplicate_referencing_same_charts(
+    mock_get_by_id_or_slug: Mock,
+    mock_copy_cmd_cls: Mock,
+    mock_find_by_id: Mock,
+    mcp_server: object,
+) -> None:
+    """Happy path: the copy references the same charts (default)."""
+    chart = _mock_chart(id=10)
+    source = _mock_dashboard(
+        id=1,
+        slices=[chart],
+        json_metadata=json.dumps({"color_scheme": "supersetColors"}),
+    )
+    new_dashboard = _mock_dashboard(id=2, title="Staging Copy", slices=[chart])
+
+    mock_get_by_id_or_slug.return_value = source
+    mock_copy_cmd_cls.return_value.run.return_value = new_dashboard
+    mock_find_by_id.return_value = new_dashboard
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "duplicate_dashboard",
+            {"request": {"dashboard_id": 1, "dashboard_title": "Staging Copy"}},
+        )
+
+    content = result.structured_content
+    assert content["error"] is None
+    assert content["duplicated_slices"] is False
+    assert content["dashboard"]["id"] == 2
+    assert content["dashboard"]["dashboard_title"] == "Staging Copy"
+    assert "/superset/dashboard/2/" in content["dashboard_url"]
+
+    # The copy data contract must mirror what the frontend "Save as" sends:
+    # required json_metadata containing the source's metadata + positions.
+    mock_copy_cmd_cls.assert_called_once()
+    cmd_source, cmd_data = mock_copy_cmd_cls.call_args.args
+    assert cmd_source is source
+    assert cmd_data["dashboard_title"] == "Staging Copy"
+    assert cmd_data["duplicate_slices"] is False
+    assert cmd_data["css"] is None
+    sent_metadata = json.loads(cmd_data["json_metadata"])
+    assert sent_metadata["color_scheme"] == "supersetColors"
+    assert sent_metadata["positions"] == SOURCE_POSITIONS
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_duplicate_with_duplicate_slices(
+    mock_get_by_id_or_slug: Mock,
+    mock_copy_cmd_cls: Mock,
+    mock_find_by_id: Mock,
+    mcp_server: object,
+) -> None:
+    """duplicate_slices=True is forwarded to the command and reported back."""
+    source = _mock_dashboard(id=1, slices=[_mock_chart(id=10)])
+    new_chart = _mock_chart(id=20)
+    new_dashboard = _mock_dashboard(id=3, title="Regional Variant", slices=[new_chart])
+
+    mock_get_by_id_or_slug.return_value = source
+    mock_copy_cmd_cls.return_value.run.return_value = new_dashboard
+    mock_find_by_id.return_value = new_dashboard
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "duplicate_dashboard",
+            {
+                "request": {
+                    "dashboard_id": 1,
+                    "dashboard_title": "Regional Variant",
+                    "duplicate_slices": True,
+                }
+            },
+        )
+
+    content = result.structured_content
+    assert content["error"] is None
+    assert content["duplicated_slices"] is True
+    assert content["dashboard"]["id"] == 3
+    assert "/superset/dashboard/3/" in content["dashboard_url"]
+
+    _, cmd_data = mock_copy_cmd_cls.call_args.args
+    assert cmd_data["duplicate_slices"] is True
+    # positions must always be present in json_metadata: the DAO reads it to
+    # remap chart IDs when duplicating slices.
+    assert "positions" in json.loads(cmd_data["json_metadata"])
+
+
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_source_not_found(
+    mock_get_by_id_or_slug: Mock, mcp_server: object
+) -> None:
+    """Returns a clear error when the source dashboard does not exist."""
+    from superset.commands.dashboard.exceptions import DashboardNotFoundError
+
+    mock_get_by_id_or_slug.side_effect = DashboardNotFoundError()
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "duplicate_dashboard",
+            {"request": {"dashboard_id": 999, "dashboard_title": "Copy"}},
+        )
+
+    content = result.structured_content
+    assert content["dashboard"] is None
+    assert content["dashboard_url"] is None
+    assert "not found" in (content["error"] or "").lower()
+
+
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_source_access_denied(
+    mock_get_by_id_or_slug: Mock, mcp_server: object
+) -> None:
+    """Returns an error when the user cannot access the source dashboard."""
+    from superset.commands.dashboard.exceptions import DashboardAccessDeniedError
+
+    mock_get_by_id_or_slug.side_effect = DashboardAccessDeniedError()
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "duplicate_dashboard",
+            {"request": {"dashboard_id": 1, "dashboard_title": "Copy"}},
+        )
+
+    content = result.structured_content
+    assert content["dashboard"] is None
+    assert "access" in (content["error"] or "").lower()
+
+
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_copy_forbidden(
+    mock_get_by_id_or_slug: Mock,
+    mock_copy_cmd_cls: Mock,
+    mcp_server: object,
+) -> None:
+    """Returns an error when the copy command raises DashboardForbiddenError
+    (e.g. DASHBOARD_RBAC requires ownership of the source)."""
+    from superset.commands.dashboard.exceptions import DashboardForbiddenError
+
+    mock_get_by_id_or_slug.return_value = _mock_dashboard(id=1)
+    mock_copy_cmd_cls.return_value.run.side_effect = DashboardForbiddenError()
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "duplicate_dashboard",
+            {"request": {"dashboard_id": 1, "dashboard_title": "Copy"}},
+        )
+
+    content = result.structured_content
+    assert content["dashboard"] is None
+    assert "permission" in (content["error"] or "").lower()
+
+
+@patch("superset.daos.dashboard.DashboardDAO.find_by_id")
+@patch("superset.commands.dashboard.copy.CopyDashboardCommand")
+@patch("superset.daos.dashboard.DashboardDAO.get_by_id_or_slug")
+@pytest.mark.asyncio
+async def test_title_xss_is_sanitized(
+    mock_get_by_id_or_slug: Mock,
+    mock_copy_cmd_cls: Mock,
+    mock_find_by_id: Mock,
+    mcp_server: object,
+) -> None:
+    """HTML/script content is stripped from the title and a warning surfaced."""
+    source = _mock_dashboard(id=1)
+    new_dashboard = _mock_dashboard(id=4, title="Regional Copy")
+
+    mock_get_by_id_or_slug.return_value = source
+    mock_copy_cmd_cls.return_value.run.return_value = new_dashboard
+    mock_find_by_id.return_value = new_dashboard
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "duplicate_dashboard",
+            {
+                "request": {
+                    "dashboard_id": 1,
+                    "dashboard_title": "<script>alert('x')</script>Regional Copy",
+                }
+            },
+        )
+
+    content = result.structured_content
+    assert content["error"] is None
+    # The sanitized title — not the raw payload — is sent to the command.
+    _, cmd_data = mock_copy_cmd_cls.call_args.args
+    assert cmd_data["dashboard_title"] == "Regional Copy"
+    assert content["warnings"], "expected a sanitization warning"
+
+
+def test_title_xss_only_rejected_by_schema() -> None:
+    """A title that sanitizes to nothing is rejected with a clear error."""
+    from pydantic import ValidationError
+
+    from superset.mcp_service.dashboard.schemas import DuplicateDashboardRequest
+
+    with pytest.raises(ValidationError):
+        DuplicateDashboardRequest(
+            dashboard_id=1, dashboard_title="<script>alert(1)</script>"
+        )
+
+
+def test_empty_title_rejected_by_schema() -> None:
+    """An empty title is rejected at the schema layer."""
+    from pydantic import ValidationError
+
+    from superset.mcp_service.dashboard.schemas import DuplicateDashboardRequest
+
+    with pytest.raises(ValidationError):
+        DuplicateDashboardRequest(dashboard_id=1, dashboard_title="")

tests/unit_tests/utils/test_split.py

@@ -0,0 +1,81 @@
+# 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 superset.utils.core import split
+
+
+def test_split_empty_string():
+    assert list(split("")) == [""]
+
+
+def test_split_leading_delimiter():
+    assert list(split(" a")) == [
+        "",
+        "a",
+    ]
+
+
+def test_split_trailing_delimiter():
+    assert list(split("a ")) == [
+        "a",
+        "",
+    ]
+
+
+def test_split_only_delimiter():
+    assert list(split(" ")) == [
+        "",
+        "",
+    ]
+
+
+def test_split_nested_parentheses():
+    assert list(
+        split(
+            "a,(b,(c,d))",
+            delimiter=",",
+        )
+    ) == [
+        "a",
+        "(b,(c,d))",
+    ]
+
+
+def test_branch_separator_found():
+    assert list(split("a b")) == [
+        "a",
+        "b",
+    ]
+
+
+def test_branch_separator_not_found():
+    assert list(split("ab")) == [
+        "ab",
+    ]
+
+
+def test_branch_parentheses():
+    assert list(split("(a b)")) == [
+        "(a b)",
+    ]
+
+
+def test_branch_escaped_quote():
+    assert list(split(r'"a\"b c" d')) == [
+        r'"a\"b c"',
+        "d",
+    ]