Superset MCP Integration

Instructions

Implementation Reference

• main.py:523-542 (handler)

  The handler function for the 'superset_dashboard_get_by_id' tool. It requires authentication and handles errors. The function makes a GET request to the Superset API endpoint `/api/v1/dashboard/{dashboard_id}` via the `make_api_request` helper to retrieve detailed dashboard information.
  @mcp.tool() @requires_auth @handle_api_errors async def superset_dashboard_get_by_id( ctx: Context, dashboard_id: int ) -> Dict[str, Any]: """ Get details for a specific dashboard Makes a request to the /api/v1/dashboard/{id} endpoint to retrieve detailed information about a specific dashboard. Args: dashboard_id: ID of the dashboard to retrieve Returns: A dictionary with complete dashboard information including components and layout """ return await make_api_request(ctx, "get", f"/api/v1/dashboard/{dashboard_id}")
• main.py:271-330 (helper)

  Core helper function that implements the API call logic for all Superset tools, including superset_dashboard_get_by_id. Handles authentication, CSRF tokens, auto token refresh on 401, and standardizes 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 that ensures the user is authenticated before executing the tool logic.
  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 that catches exceptions and returns standardized error messages.
  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:188-240 (helper)

  Helper function used by make_api_request to automatically refresh the access token on 401 Unauthorized errors and retry the API call.
  async def with_auto_refresh( ctx: Context, api_call: Callable[[], Awaitable[httpx.Response]] ) -> httpx.Response: """ Helper function to handle automatic token refreshing for API calls This function will attempt to execute the provided API call. If the call fails with a 401 Unauthorized error, it will try to refresh the token and retry the API call once. Args: ctx: The MCP context api_call: The API call function to execute (should be a callable that returns a response) """ superset_ctx: SupersetContext = ctx.request_context.lifespan_context if not superset_ctx.access_token: raise HTTPException(status_code=401, detail="Not authenticated") # First attempt try: response = await api_call() # If not an auth error, return the response if response.status_code != 401: return response except httpx.HTTPStatusError as e: if e.response.status_code != 401: raise e response = e.response except Exception as e: # For other errors, just raise raise e # If we got a 401, try to refresh the token logger.info("Received 401 Unauthorized. Attempting to refresh token...") refresh_result = await superset_auth_refresh_token(ctx) if refresh_result.get("error"): # If refresh failed, try to re-authenticate logger.info( f"Token refresh failed: {refresh_result.get('error')}. Attempting re-authentication..." ) auth_result = await superset_auth_authenticate_user(ctx) if auth_result.get("error"): # If re-authentication failed, raise an exception raise HTTPException(status_code=401, detail="Authentication failed") # Retry the API call with the new token return await api_call()