Superset MCP Integration

Instructions

Implementation Reference

• main.py:857-874 (handler)

  The main execution handler for the superset_database_schemas MCP tool. This function takes a database_id parameter and delegates to the make_api_request helper to perform a GET request to the Superset REST API endpoint /api/v1/database/{database_id}/schemas/ to fetch available schemas.
  @requires_auth @handle_api_errors async def superset_database_schemas(ctx: Context, database_id: int) -> Dict[str, Any]: """ Get schemas for a specific database Makes a request to the /api/v1/database/{id}/schemas/ endpoint to retrieve all schemas available in the database. Args: database_id: ID of the database Returns: A dictionary with list of schema names """ return await make_api_request( ctx, "get", f"/api/v1/database/{database_id}/schemas/" )
• main.py:271-330 (helper)

  Key helper function used by the tool handler to execute the HTTP GET request to the Superset API. Handles authentication headers, CSRF tokens, automatic token refresh on 401 errors, and error handling.
  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 handler that ensures the user is authenticated (has a valid access token) before executing the tool.
  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:170-186 (helper)

  Decorator applied to the handler that catches exceptions and returns standardized error responses.
  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:141-145 (registration)

  Initialization of the FastMCP server instance where all @mcp.tool() decorated functions are automatically registered as MCP tools.
  mcp = FastMCP( "superset", lifespan=superset_lifespan, dependencies=["fastapi", "uvicorn", "python-dotenv", "httpx"], )