feat(mcp): add save_sql_query tool for persisting SQL queries

superset/mcp_service/app.py

@@ -350,6 +350,7 @@ def create_mcp_app(
 from superset.mcp_service.sql_lab.tool import (  # noqa: F401, E402
     execute_sql,
     open_sql_lab_with_context,
+    save_sql_query,
 )
 from superset.mcp_service.system import (  # noqa: F401, E402
     prompts as system_prompts,

superset/mcp_service/sql_lab/schemas.py

@@ -139,3 +139,56 @@ class SqlLabResponse(BaseModel):
     schema_name: str | None = Field(None, description="Schema selected", alias="schema")
     title: str | None = Field(None, description="Query title")
     error: str | None = Field(None, description="Error message if failed")
+
+
+class SaveSqlQueryRequest(BaseModel):
+    """Request schema for saving a SQL query."""
+
+    database_id: int = Field(
+        ..., description="Database connection ID for this saved query"
+    )
+    sql: str = Field(
+        ...,
+        description="SQL query to save",
+    )
+    label: str = Field(
+        ...,
+        description="Name/label for the saved query",
+        min_length=1,
+        max_length=256,
+    )
+    description: str | None = Field(
+        None, description="Description of what this query does"
+    )
+    schema_name: str | None = Field(
+        None,
+        description="Default schema for this query",
+        alias="schema",
+        max_length=128,
+    )
+    catalog: str | None = Field(
+        None, description="Default catalog for this query", max_length=256
+    )
+    template_parameters: str | None = Field(
+        None,
+        description="Jinja2 template parameters as JSON string",
+    )
+
+    @field_validator("sql")
+    @classmethod
+    def sql_not_empty(cls, v: str) -> str:
+        if not v or not v.strip():
+            raise ValueError("SQL query cannot be empty")
+        return v.strip()
+
+
+class SaveSqlQueryResponse(BaseModel):
+    """Response schema for saved query creation."""
+
+    id: int = Field(..., description="Saved query ID")
+    label: str = Field(..., description="Saved query label/name")
+    sql_lab_url: str = Field(..., description="URL to open the saved query in SQL Lab")
+    database_name: str | None = Field(
+        None, description="Name of the associated database"
+    )
+    error: str | None = Field(None, description="Error message if save failed")

superset/mcp_service/sql_lab/tool/__init__.py

@@ -23,8 +23,10 @@
 from superset.mcp_service.sql_lab.tool.open_sql_lab_with_context import (
     open_sql_lab_with_context,
 )
+from superset.mcp_service.sql_lab.tool.save_sql_query import save_sql_query
 
 __all__ = [
     "execute_sql",
     "open_sql_lab_with_context",
+    "save_sql_query",
 ]

superset/mcp_service/sql_lab/tool/save_sql_query.py

@@ -0,0 +1,121 @@
+# 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: save_sql_query
+
+Save a SQL query so it appears in the Saved Queries list and can be
+opened in SQL Lab via a direct URL.
+"""
+
+import logging
+
+from fastmcp import Context
+from superset_core.mcp import tool
+
+from superset.exceptions import SupersetSecurityException
+from superset.extensions import event_logger
+from superset.mcp_service.sql_lab.schemas import (
+    SaveSqlQueryRequest,
+    SaveSqlQueryResponse,
+)
+from superset.mcp_service.utils.schema_utils import parse_request
+from superset.mcp_service.utils.url_utils import get_superset_base_url
+
+logger = logging.getLogger(__name__)
+
+
+@tool(tags=["mutate"])
+@parse_request(SaveSqlQueryRequest)
+def save_sql_query(request: SaveSqlQueryRequest, ctx: Context) -> SaveSqlQueryResponse:
+    """Save a SQL query as a named saved query.
+
+    The saved query appears in the Saved Queries list and can be opened
+    directly in SQL Lab via the returned URL.
+
+    Typical workflow:
+    1. execute_sql(database_id, sql) — run and verify the query
+    2. save_sql_query(database_id, sql, label) — persist it
+
+    Returns:
+    - Saved query ID, label, and SQL Lab URL
+    """
+    try:
+        from flask import g
+
+        from superset import db, security_manager
+        from superset.models.core import Database
+        from superset.models.sql_lab import SavedQuery
+
+        # Validate database exists and user has access
+        with event_logger.log_context(action="mcp.save_sql_query.db_validation"):
+            database = (
+                db.session.query(Database).filter_by(id=request.database_id).first()
+            )
+            if not database:
+                return SaveSqlQueryResponse(
+                    id=0,
+                    label=request.label,
+                    sql_lab_url="",
+                    error=f"Database with ID {request.database_id} not found",
+                )
+
+            if not security_manager.can_access_database(database):
+                return SaveSqlQueryResponse(
+                    id=0,
+                    label=request.label,
+                    sql_lab_url="",
+                    error=f"Access denied to database {database.database_name}",
+                )
+
+        # Create the saved query
+        with event_logger.log_context(action="mcp.save_sql_query.create"):
+            saved_query = SavedQuery(
+                db_id=request.database_id,
+                sql=request.sql,
+                label=request.label,
+                description=request.description or "",
+                schema=request.schema_name or "",
+                catalog=request.catalog,
+                template_parameters=request.template_parameters,
+                user_id=g.user.id if hasattr(g, "user") and g.user else None,
+            )
+
+            db.session.add(saved_query)
+            db.session.commit()
+
+        sql_lab_url = f"{get_superset_base_url()}/sqllab?savedQueryId={saved_query.id}"
+
+        logger.info(
+            "Saved query created: id=%s, label=%s", saved_query.id, saved_query.label
+        )
+
+        return SaveSqlQueryResponse(
+            id=saved_query.id,
+            label=saved_query.label,
+            sql_lab_url=sql_lab_url,
+            database_name=database.database_name,
+        )
+
+    except SupersetSecurityException as exc:
+        logger.warning("Security error saving query: %s", exc)
+        return SaveSqlQueryResponse(
+            id=0,
+            label=request.label,
+            sql_lab_url="",
+            error=f"Permission denied: {exc}",
+        )

tests/unit_tests/mcp_service/sql_lab/tool/test_save_sql_query.py

@@ -0,0 +1,112 @@
+# 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 MCP save_sql_query tool."""
+
+import pytest
+
+from superset.mcp_service.sql_lab.schemas import (
+    SaveSqlQueryRequest,
+    SaveSqlQueryResponse,
+)
+
+
+class TestSaveSqlQuerySchemas:
+    """Tests for save_sql_query request/response schemas."""
+
+    def test_valid_request(self):
+        """Test creating a valid SaveSqlQueryRequest."""
+        request = SaveSqlQueryRequest(
+            database_id=1,
+            sql="SELECT * FROM users",
+            label="All Users",
+        )
+        assert request.database_id == 1
+        assert request.sql == "SELECT * FROM users"
+        assert request.label == "All Users"
+        assert request.description is None
+        assert request.schema_name is None
+        assert request.catalog is None
+        assert request.template_parameters is None
+
+    def test_request_with_all_fields(self):
+        """Test request with all optional fields populated."""
+        request = SaveSqlQueryRequest(
+            database_id=1,
+            sql="SELECT * FROM {{ table }}",
+            label="Parameterized Query",
+            description="A query with Jinja2 templates",
+            schema="public",
+            catalog="main",
+            template_parameters='{"table": "users"}',
+        )
+        assert request.description == "A query with Jinja2 templates"
+        assert request.schema_name == "public"
+        assert request.catalog == "main"
+        assert request.template_parameters == '{"table": "users"}'
+
+    def test_request_rejects_empty_sql(self):
+        """Test that empty SQL is rejected."""
+        with pytest.raises(ValueError, match="SQL query cannot be empty"):
+            SaveSqlQueryRequest(
+                database_id=1,
+                sql="   ",
+                label="Bad Query",
+            )
+
+    def test_request_rejects_empty_label(self):
+        """Test that empty label is rejected."""
+        with pytest.raises(ValueError):
+            SaveSqlQueryRequest(
+                database_id=1,
+                sql="SELECT 1",
+                label="",
+            )
+
+    def test_request_strips_sql_whitespace(self):
+        """Test that SQL whitespace is stripped."""
+        request = SaveSqlQueryRequest(
+            database_id=1,
+            sql="  SELECT 1  ",
+            label="Simple",
+        )
+        assert request.sql == "SELECT 1"
+
+    def test_response_structure(self):
+        """Test SaveSqlQueryResponse structure."""
+        response = SaveSqlQueryResponse(
+            id=42,
+            label="My Query",
+            sql_lab_url="http://localhost:8088/sqllab?savedQueryId=42",
+            database_name="examples",
+        )
+        assert response.id == 42
+        assert response.label == "My Query"
+        assert "savedQueryId=42" in response.sql_lab_url
+        assert response.database_name == "examples"
+        assert response.error is None
+
+    def test_response_with_error(self):
+        """Test SaveSqlQueryResponse with error."""
+        response = SaveSqlQueryResponse(
+            id=0,
+            label="Failed Query",
+            sql_lab_url="",
+            error="Database not found",
+        )
+        assert response.id == 0
+        assert response.error == "Database not found"