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()