mirror of
https://github.com/apache/superset.git
synced 2026-04-19 08:04:53 +00:00
feat(mcp): add configurable branding for MCP service (#36033)
This commit is contained in:
@@ -29,10 +29,20 @@ from fastmcp import FastMCP
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Default instructions for the Superset MCP service
|
||||
DEFAULT_INSTRUCTIONS = """
|
||||
You are connected to the Apache Superset MCP (Model Context Protocol) service.
|
||||
This service provides programmatic access to Superset dashboards, charts, datasets,
|
||||
|
||||
def get_default_instructions(branding: str = "Apache Superset") -> str:
|
||||
"""Get default instructions with configurable branding.
|
||||
|
||||
Args:
|
||||
branding: Product name to use in instructions
|
||||
(e.g., "ACME Analytics", "Apache Superset")
|
||||
|
||||
Returns:
|
||||
Formatted instructions string with branding applied
|
||||
"""
|
||||
return f"""
|
||||
You are connected to the {branding} MCP (Model Context Protocol) service.
|
||||
This service provides programmatic access to {branding} dashboards, charts, datasets,
|
||||
SQL Lab, and instance metadata via a comprehensive set of tools.
|
||||
|
||||
Available tools:
|
||||
@@ -107,9 +117,9 @@ Common Visualization Types:
|
||||
|
||||
Query Examples:
|
||||
- List all interactive tables:
|
||||
filters=[{"col": "viz_type", "opr": "in", "value": ["table", "pivot_table_v2"]}]
|
||||
filters=[{{"col": "viz_type", "opr": "in", "value": ["table", "pivot_table_v2"]}}]
|
||||
- List time series charts:
|
||||
filters=[{"col": "viz_type", "opr": "sw", "value": "echarts_timeseries"}]
|
||||
filters=[{{"col": "viz_type", "opr": "sw", "value": "echarts_timeseries"}}]
|
||||
- Search by name: search="sales"
|
||||
|
||||
General usage tips:
|
||||
@@ -124,6 +134,10 @@ or use the superset_quickstart prompt for an interactive guide.
|
||||
"""
|
||||
|
||||
|
||||
# For backwards compatibility, keep DEFAULT_INSTRUCTIONS pointing to default branding
|
||||
DEFAULT_INSTRUCTIONS = get_default_instructions()
|
||||
|
||||
|
||||
def _build_mcp_kwargs(
|
||||
name: str,
|
||||
instructions: str,
|
||||
@@ -185,6 +199,7 @@ def _log_instance_creation(
|
||||
def create_mcp_app(
|
||||
name: str = "Superset MCP Server",
|
||||
instructions: str | None = None,
|
||||
branding: str | None = None,
|
||||
auth: Any | None = None,
|
||||
lifespan: Callable[..., Any] | None = None,
|
||||
tools: List[Any] | None = None,
|
||||
@@ -203,6 +218,7 @@ def create_mcp_app(
|
||||
Args:
|
||||
name: Human-readable server name
|
||||
instructions: Server description and usage instructions
|
||||
branding: Product name for instructions (e.g., "ACME Analytics")
|
||||
auth: Authentication provider for securing HTTP transports
|
||||
lifespan: Async context manager for startup/shutdown logic
|
||||
tools: List of tools or functions to add to the server
|
||||
@@ -216,7 +232,11 @@ def create_mcp_app(
|
||||
"""
|
||||
# Use default instructions if none provided
|
||||
if instructions is None:
|
||||
instructions = DEFAULT_INSTRUCTIONS
|
||||
# If branding is provided, use it to generate instructions
|
||||
if branding is not None:
|
||||
instructions = get_default_instructions(branding)
|
||||
else:
|
||||
instructions = DEFAULT_INSTRUCTIONS
|
||||
|
||||
# Build FastMCP constructor arguments
|
||||
mcp_kwargs = _build_mcp_kwargs(
|
||||
@@ -290,7 +310,7 @@ from superset.mcp_service.system.tool import ( # noqa: F401, E402
|
||||
|
||||
|
||||
def init_fastmcp_server(
|
||||
name: str = "Superset MCP Server",
|
||||
name: str | None = None,
|
||||
instructions: str | None = None,
|
||||
auth: Any | None = None,
|
||||
lifespan: Callable[..., Any] | None = None,
|
||||
@@ -308,16 +328,33 @@ def init_fastmcp_server(
|
||||
a new instance will be created with those settings.
|
||||
|
||||
Args:
|
||||
Same as create_mcp_app()
|
||||
name: Server name (defaults to "{APP_NAME} MCP Server")
|
||||
instructions: Custom instructions (defaults to branded with APP_NAME)
|
||||
auth, lifespan, tools, include_tags, exclude_tags, config: FastMCP configuration
|
||||
**kwargs: Additional FastMCP configuration
|
||||
|
||||
Returns:
|
||||
FastMCP instance (either the global one or a new custom one)
|
||||
"""
|
||||
# Read branding from Flask config's APP_NAME
|
||||
from superset.mcp_service.flask_singleton import app as flask_app
|
||||
|
||||
# Derive branding from Superset's APP_NAME config (defaults to "Superset")
|
||||
app_name = flask_app.config.get("APP_NAME", "Superset")
|
||||
branding = app_name
|
||||
default_name = f"{app_name} MCP Server"
|
||||
|
||||
# Apply branding defaults if not explicitly provided
|
||||
if name is None:
|
||||
name = default_name
|
||||
if instructions is None:
|
||||
instructions = get_default_instructions(branding)
|
||||
|
||||
# If any custom parameters are provided, create a new instance
|
||||
custom_params_provided = any(
|
||||
[
|
||||
name != "Superset MCP Server",
|
||||
instructions is not None,
|
||||
name != default_name,
|
||||
instructions != get_default_instructions(branding),
|
||||
auth is not None,
|
||||
lifespan is not None,
|
||||
tools is not None,
|
||||
|
||||
@@ -62,8 +62,9 @@ MCP_CSRF_CONFIG = {
|
||||
|
||||
# FastMCP Factory Configuration
|
||||
MCP_FACTORY_CONFIG = {
|
||||
"name": "Superset MCP Server",
|
||||
"instructions": None, # Will use default from app.py
|
||||
"name": None, # Will derive from APP_NAME in app.py
|
||||
"branding": None, # Will derive from APP_NAME in app.py
|
||||
"instructions": None, # Will use default from app.py (parameterized with branding)
|
||||
"auth": None, # No authentication by default
|
||||
"lifespan": None, # No custom lifespan
|
||||
"tools": None, # Auto-discover tools
|
||||
|
||||
137
tests/unit_tests/mcp_service/test_mcp_config.py
Normal file
137
tests/unit_tests/mcp_service/test_mcp_config.py
Normal file
@@ -0,0 +1,137 @@
|
||||
# Licensed to the Apache Software Foundation (ASF) under one
|
||||
# or more contributor license agreements. See the NOTICE file
|
||||
# distributed with this work for additional information
|
||||
# regarding copyright ownership. The ASF licenses this file
|
||||
# to you under the Apache License, Version 2.0 (the
|
||||
# "License"); you may not use this file except in compliance
|
||||
# with the License. You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing,
|
||||
# software distributed under the License is distributed on an
|
||||
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
# KIND, either express or implied. See the License for the
|
||||
# specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
"""Tests for MCP service configuration and branding."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from superset.mcp_service.app import get_default_instructions, init_fastmcp_server
|
||||
|
||||
|
||||
def test_get_default_instructions_with_default_branding():
|
||||
"""Test that default branding produces Apache Superset in instructions."""
|
||||
instructions = get_default_instructions()
|
||||
|
||||
assert "Apache Superset" in instructions
|
||||
assert "Apache Superset MCP" in instructions
|
||||
assert "model context protocol" in instructions.lower()
|
||||
|
||||
|
||||
def test_get_default_instructions_with_custom_branding():
|
||||
"""Test that custom branding is reflected in instructions."""
|
||||
custom_branding = "ACME Analytics"
|
||||
instructions = get_default_instructions(branding=custom_branding)
|
||||
|
||||
assert custom_branding in instructions
|
||||
assert f"{custom_branding} MCP" in instructions
|
||||
# Should not contain default Apache Superset branding
|
||||
assert "Apache Superset" not in instructions
|
||||
|
||||
|
||||
def test_get_default_instructions_with_enterprise_branding():
|
||||
"""Test instructions with enterprise/white-label branding."""
|
||||
enterprise_branding = "DataViz Platform"
|
||||
instructions = get_default_instructions(branding=enterprise_branding)
|
||||
|
||||
assert enterprise_branding in instructions
|
||||
assert f"{enterprise_branding} MCP" in instructions
|
||||
# Verify it contains expected tool documentation
|
||||
assert "list_dashboards" in instructions
|
||||
assert "list_charts" in instructions
|
||||
assert "execute_sql" in instructions
|
||||
|
||||
|
||||
def test_init_fastmcp_server_with_default_app_name():
|
||||
"""Test that default APP_NAME produces Superset branding."""
|
||||
# Mock Flask app config with default APP_NAME
|
||||
mock_flask_app = MagicMock()
|
||||
mock_flask_app.config.get.return_value = "Superset"
|
||||
|
||||
# Patch at the import location to avoid actual Flask app creation
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{"superset.mcp_service.flask_singleton": MagicMock(app=mock_flask_app)},
|
||||
):
|
||||
with patch("superset.mcp_service.app.create_mcp_app") as mock_create:
|
||||
mock_mcp = MagicMock()
|
||||
mock_create.return_value = mock_mcp
|
||||
|
||||
# Call with custom name to force create_mcp_app path
|
||||
init_fastmcp_server(name="Custom Name")
|
||||
|
||||
# Verify create_mcp_app was called
|
||||
assert mock_create.called
|
||||
# Verify instructions use Superset branding (not Apache Superset)
|
||||
call_kwargs = mock_create.call_args[1]
|
||||
assert "Superset MCP" in call_kwargs["instructions"]
|
||||
assert "Superset dashboards" in call_kwargs["instructions"]
|
||||
|
||||
|
||||
def test_init_fastmcp_server_with_custom_app_name():
|
||||
"""Test that custom APP_NAME produces branded instructions."""
|
||||
custom_app_name = "ACME Analytics"
|
||||
# Mock Flask app config with custom APP_NAME
|
||||
mock_flask_app = MagicMock()
|
||||
mock_flask_app.config.get.return_value = custom_app_name
|
||||
|
||||
# Patch at the import location to avoid actual Flask app creation
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{"superset.mcp_service.flask_singleton": MagicMock(app=mock_flask_app)},
|
||||
):
|
||||
with patch("superset.mcp_service.app.create_mcp_app") as mock_create:
|
||||
mock_mcp = MagicMock()
|
||||
mock_create.return_value = mock_mcp
|
||||
|
||||
# Call with custom name to force create_mcp_app path
|
||||
init_fastmcp_server(name="Custom Name")
|
||||
|
||||
# Verify create_mcp_app was called
|
||||
assert mock_create.called
|
||||
# Verify instructions use custom branding
|
||||
call_kwargs = mock_create.call_args[1]
|
||||
assert custom_app_name in call_kwargs["instructions"]
|
||||
# Should not contain default Apache Superset branding
|
||||
assert "Apache Superset" not in call_kwargs["instructions"]
|
||||
|
||||
|
||||
def test_init_fastmcp_server_derives_server_name_from_app_name():
|
||||
"""Test that server name is derived from APP_NAME."""
|
||||
custom_app_name = "DataViz Platform"
|
||||
expected_server_name = f"{custom_app_name} MCP Server"
|
||||
|
||||
# Mock Flask app config
|
||||
mock_flask_app = MagicMock()
|
||||
mock_flask_app.config.get.return_value = custom_app_name
|
||||
|
||||
# Patch at the import location to avoid actual Flask app creation
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{"superset.mcp_service.flask_singleton": MagicMock(app=mock_flask_app)},
|
||||
):
|
||||
with patch("superset.mcp_service.app.create_mcp_app") as mock_create:
|
||||
mock_mcp = MagicMock()
|
||||
mock_create.return_value = mock_mcp
|
||||
|
||||
# Call without name parameter (should use default derived name)
|
||||
# Force custom params by passing instructions
|
||||
init_fastmcp_server(instructions="custom")
|
||||
|
||||
# Verify create_mcp_app was called with derived name
|
||||
assert mock_create.called
|
||||
call_kwargs = mock_create.call_args[1]
|
||||
assert call_kwargs["name"] == expected_server_name
|
||||
Reference in New Issue
Block a user