Superset MCP Integration

Instructions

Implementation Reference

• main.py:368-411 (handler)

  The handler function implementing the superset_auth_refresh_token MCP tool. It refreshes the Superset JWT access token by calling the /api/v1/security/refresh endpoint, updates the stored token and client headers upon success.

  @mcp.tool()
  @handle_api_errors
  async def superset_auth_refresh_token(ctx: Context) -> Dict[str, Any]:
      """
      Refresh the access token using the refresh endpoint
  
      Makes a request to the /api/v1/security/refresh endpoint to get a new access token
      without requiring re-authentication with username/password.
  
      Returns:
          A dictionary with the new access token or error information
      """
      superset_ctx: SupersetContext = ctx.request_context.lifespan_context
  
      if not superset_ctx.access_token:
          return {"error": "No access token to refresh. Please authenticate first."}
  
      try:
          # Use the refresh endpoint to get a new token
          response = await superset_ctx.client.post("/api/v1/security/refresh")
  
          if response.status_code != 200:
              return {
                  "error": f"Failed to refresh token: {response.status_code} - {response.text}"
              }
  
          data = response.json()
          access_token = data.get("access_token")
  
          if not access_token:
              return {"error": "No access token returned from refresh"}
  
          # Save and set the new access token
          save_access_token(access_token)
          superset_ctx.access_token = access_token
          superset_ctx.client.headers.update({"Authorization": f"Bearer {access_token}"})
  
          return {
              "message": "Successfully refreshed access token",
              "access_token": access_token,
          }
      except Exception as e:
          return {"error": f"Error refreshing token: {str(e)}"}

• main.py:368-368 (registration)

  The @mcp.tool() decorator registers the superset_auth_refresh_token function as an MCP tool.

  @mcp.tool()

• main.py:189-240 (helper)

  Helper function with_auto_refresh that automatically calls superset_auth_refresh_token when API calls return 401 Unauthorized.

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