Superset MCP Integration

Instructions

Implementation Reference

• main.py:780-833 (handler)

  The core handler function implementing the superset_database_create tool. It is decorated as an MCP tool, requires authentication, handles errors, constructs a payload from input parameters (engine, configuration_method, database_name, sqlalchemy_uri) with default permissions, and makes a POST request to Superset's /api/v1/database/ endpoint using the make_api_request helper.
  @mcp.tool() @requires_auth @handle_api_errors async def superset_database_create( ctx: Context, engine: str, configuration_method: str, database_name: str, sqlalchemy_uri: str, ) -> Dict[str, Any]: """ Create a new database connection in Superset IMPORTANT: Don't call this tool, unless user have given connection details. This function will only create database connections with explicit user consent and input. No default values or assumptions will be made without user confirmation. All connection parameters, including sensitive credentials, must be explicitly provided by the user. Makes a POST request to /api/v1/database/ to create a new database connection in Superset. The endpoint requires a valid SQLAlchemy URI and database configuration parameters. The engine parameter will be automatically determined from the SQLAlchemy URI prefix if not specified: - 'postgresql://' -> engine='postgresql' - 'mysql://' -> engine='mysql' - 'mssql://' -> engine='mssql' - 'oracle://' -> engine='oracle' - 'sqlite://' -> engine='sqlite' The SQLAlchemy URI must follow the format: dialect+driver://username:password@host:port/database If the URI is not provided, the function will prompt for individual connection parameters to construct it. All required parameters must be provided and validated before creating the connection. The configuration_method parameter should typically be set to 'sqlalchemy_form'. Args: engine: Database engine (e.g., 'postgresql', 'mysql', etc.) configuration_method: Method used for configuration (typically 'sqlalchemy_form') database_name: Name for the database connection sqlalchemy_uri: SQLAlchemy URI for the connection (e.g., 'postgresql://user:pass@host/db') Returns: A dictionary with the created database connection information including its ID """ payload = { "engine": engine, "configuration_method": configuration_method, "database_name": database_name, "sqlalchemy_uri": sqlalchemy_uri, "allow_dml": True, "allow_cvas": True, "allow_ctas": True, "expose_in_sqllab": True, } return await make_api_request(ctx, "post", "/api/v1/database/", data=payload)
• main.py:271-330 (helper)

  Key helper function used by the tool to perform authenticated HTTP requests to the Superset API, handling CSRF tokens, auto token refresh on 401, and error responses.
  async def make_api_request( ctx: Context, method: str, endpoint: str, data: Dict[str, Any] = None, params: Dict[str, Any] = None, auto_refresh: bool = True, ) -> Dict[str, Any]: """ Helper function to make API requests to Superset Args: ctx: MCP context method: HTTP method (get, post, put, delete) endpoint: API endpoint (without base URL) data: Optional JSON payload for POST/PUT requests params: Optional query parameters auto_refresh: Whether to auto-refresh token on 401 """ superset_ctx: SupersetContext = ctx.request_context.lifespan_context client = superset_ctx.client # For non-GET requests, make sure we have a CSRF token if method.lower() != "get" and not superset_ctx.csrf_token: await get_csrf_token(ctx) async def make_request() -> httpx.Response: headers = {} # Add CSRF token for non-GET requests if method.lower() != "get" and superset_ctx.csrf_token: headers["X-CSRFToken"] = superset_ctx.csrf_token if method.lower() == "get": return await client.get(endpoint, params=params) elif method.lower() == "post": return await client.post( endpoint, json=data, params=params, headers=headers ) elif method.lower() == "put": return await client.put(endpoint, json=data, headers=headers) elif method.lower() == "delete": return await client.delete(endpoint, headers=headers) else: raise ValueError(f"Unsupported HTTP method: {method}") # Use auto_refresh if requested response = ( await with_auto_refresh(ctx, make_request) if auto_refresh else await make_request() ) if response.status_code not in [200, 201]: return { "error": f"API request failed: {response.status_code} - {response.text}" } return response.json()
• main.py:154-169 (helper)

  Decorator applied to the tool handler to ensure authentication before execution.
  def requires_auth( func: Callable[..., Awaitable[Dict[str, Any]]], ) -> Callable[..., Awaitable[Dict[str, Any]]]: """Decorator to check authentication before executing a function""" @wraps(func) async def wrapper(ctx: Context, *args, **kwargs) -> Dict[str, Any]: superset_ctx: SupersetContext = ctx.request_context.lifespan_context if not superset_ctx.access_token: return {"error": "Not authenticated. Please authenticate first."} return await func(ctx, *args, **kwargs) return wrapper
• main.py:171-186 (helper)

  Decorator applied to the tool handler to catch and format exceptions consistently.
  def handle_api_errors( func: Callable[..., Awaitable[Dict[str, Any]]], ) -> Callable[..., Awaitable[Dict[str, Any]]]: """Decorator to handle API errors in a consistent way""" @wraps(func) async def wrapper(ctx: Context, *args, **kwargs) -> Dict[str, Any]: try: return await func(ctx, *args, **kwargs) except Exception as e: # Extract function name for better error context function_name = func.__name__ return {"error": f"Error in {function_name}: {str(e)}"} return wrapper
• main.py:780-780 (registration)

  MCP tool registration decorator that registers the superset_database_create function as an available tool.
  @mcp.tool()