"""
Looker SDK client initialization and configuration.
"""
import os
import re
import sys
import logging
import traceback
from typing import Any, Dict, Optional, Union
import looker_sdk
from looker_sdk.error import SDKError
from dotenv import load_dotenv
# Set up logging
logger = logging.getLogger('looker-mcp.client')
# Load environment variables from .env file if present
load_dotenv()
def init_looker_sdk():
"""Initialize Looker SDK client."""
logger.info("Initializing Looker SDK client")
try:
# Log environment variable state
logger.debug("Checking Looker SDK environment variables:")
required_vars = ["LOOKER_BASE_URL", "LOOKER_CLIENT_ID", "LOOKER_CLIENT_SECRET"]
for var in required_vars:
is_set = var in os.environ and os.environ[var]
sdk_var = f"LOOKERSDK_{var[7:]}" # Convert LOOKER_X to LOOKERSDK_X
sdk_is_set = sdk_var in os.environ and os.environ[sdk_var]
logger.debug(f" {var}: {'set' if is_set else 'NOT SET'}")
logger.debug(f" {sdk_var}: {'set' if sdk_is_set else 'NOT SET'}")
if not (is_set or sdk_is_set):
logger.error(f"Required environment variable {var} or {sdk_var} is not set")
# Check if the SDK environment variables are mapped properly
if "LOOKER_TIMEOUT" in os.environ and "LOOKERSDK_TIMEOUT" not in os.environ:
logger.warning("LOOKER_TIMEOUT is set but LOOKERSDK_TIMEOUT is not. The SDK may not use the correct timeout.")
logger.warning(f"LOOKER_TIMEOUT value: {os.environ.get('LOOKER_TIMEOUT')}")
# Automatically map the variable
os.environ["LOOKERSDK_TIMEOUT"] = os.environ["LOOKER_TIMEOUT"]
logger.info(f"Mapped LOOKER_TIMEOUT to LOOKERSDK_TIMEOUT: {os.environ['LOOKERSDK_TIMEOUT']}")
# Map other important variables if needed
for looker_var, sdk_var in [
("LOOKER_BASE_URL", "LOOKERSDK_BASE_URL"),
("LOOKER_CLIENT_ID", "LOOKERSDK_CLIENT_ID"),
("LOOKER_CLIENT_SECRET", "LOOKERSDK_CLIENT_SECRET"),
("LOOKER_VERIFY_SSL", "LOOKERSDK_VERIFY_SSL")
]:
if looker_var in os.environ and sdk_var not in os.environ:
os.environ[sdk_var] = os.environ[looker_var]
logger.info(f"Mapped {looker_var} to {sdk_var}")
# Configure the Looker SDK settings
logger.debug("Attempting to initialize looker_sdk.init40()")
client = looker_sdk.init40()
logger.info("SDK initialization successful")
return client
except Exception as e:
logger.error(f"Error initializing Looker SDK: {e}")
if hasattr(e, '__traceback__'):
logger.error("Error traceback: " + "".join(traceback.format_tb(e.__traceback__)))
# Check for specific error conditions
error_str = str(e).lower()
if "no such file" in error_str and "looker.ini" in error_str:
logger.error("looker.ini file not found. The SDK is looking for configuration but can't find it.")
elif "timeout" in error_str:
logger.error("Timeout error detected. Check your network connection and timeout settings.")
elif "authentication" in error_str or "unauthorized" in error_str:
logger.error("Authentication error detected. Check your client ID and secret.")
elif "certificate" in error_str or "ssl" in error_str:
logger.error("SSL certificate error detected. Check your verify_ssl setting.")
return None
# Global SDK instance
SDK_INSTANCE = None
def get_sdk():
"""Get or initialize the SDK instance."""
global SDK_INSTANCE
if SDK_INSTANCE is None:
logger.debug("Creating new SDK instance")
SDK_INSTANCE = init_looker_sdk()
if SDK_INSTANCE is None:
logger.error("Failed to initialize SDK")
# Instead of raising an exception, return a custom error object
error_obj = type('ErrorSDK', (), {})()
error_obj.error = True
error_obj.me = lambda: {"error": "Failed to initialize Looker SDK. Check logs for details."}
logger.debug("Returning error SDK object")
return error_obj
return SDK_INSTANCE
def clean_api_error(error_message: str) -> str:
"""Clean API error messages to remove sensitive information."""
# Remove sensitive URL data
error_message = re.sub(r'https?://[^/\s]+/api/\d+\.\d+/login', 'https://[redacted]/api/X.Y/login', error_message)
# Remove anything that looks like client_id or client_secret
error_message = re.sub(r'client_id=[^\s&]+', 'client_id=[redacted]', error_message)
error_message = re.sub(r'client_secret=[^\s&]+', 'client_secret=[redacted]', error_message)
# Remove IP addresses
error_message = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '[redacted-ip]', error_message)
return error_message
def reset_sdk():
"""Reset the global SDK instance, forcing a new initialization."""
global SDK_INSTANCE
SDK_INSTANCE = None
logger.info("SDK instance reset")
