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"],
  )