#!/usr/bin/env python3
"""
Server utilities - currently only debug functions.
"""
import os
import sys
import logging
import socket
import urllib.parse
import time
import requests
import platform
from typing import Dict, Any
# Import looker_sdk and specific functions using absolute paths
import looker_sdk
from looker_mcp.server.looker import init_looker_sdk, SDKInitializationError
logger = logging.getLogger('looker-mcp.server.utils')
# --- Debug Functions (Moved from looker.py, decorators removed) ---
async def debug_network_connectivity() -> Dict[str, Any]:
logger.info("debug_network_connectivity util called")
try:
base_url = os.environ.get('LOOKERSDK_BASE_URL', '')
if not base_url:
return {"error": "LOOKERSDK_BASE_URL not set"}
parsed_url = urllib.parse.urlparse(base_url)
host = parsed_url.netloc.split(":")[0] if ":" in parsed_url.netloc else parsed_url.netloc
port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)
socket_result = {"success": False, "error": None, "time_seconds": 0}
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
start_time = time.time()
try:
sock.connect((host, port))
socket_result["success"] = True
except Exception as e:
socket_result["error"] = str(e)
finally:
socket_result["time_seconds"] = time.time() - start_time
sock.close()
http_result = {"success": False, "status_code": None, "error": None, "time_seconds": 0}
start_time = time.time()
try:
# Use a specific, lightweight endpoint like /alive if available
# Defaulting to /api/4.0/alive, adjust if needed
alive_url = f"{base_url.rstrip('/')}/api/4.0/alive"
logger.debug(f"HTTP test requesting: {alive_url}")
response = requests.get(alive_url, verify=True, timeout=10)
http_result["success"] = response.status_code < 400
http_result["status_code"] = response.status_code
http_result["content"] = response.text[:100] + "..." if len(response.text) > 100 else response.text
logger.debug(f"HTTP test response: {response.status_code}")
except Exception as e:
http_result["error"] = str(e)
logger.error(f"HTTP test failed: {e}")
finally:
http_result["time_seconds"] = time.time() - start_time
env_info = {
"python_version": sys.version,
"platform": platform.platform(),
"looker_sdk_version": getattr(looker_sdk, "__version__", "unknown"),
"hostname": socket.gethostname()
}
return {
"base_url": base_url,
"host": host,
"port": port,
"socket_test": socket_result,
"http_test": http_result,
"environment": env_info
}
except Exception as e:
logger.error(f"Error in debug_network_connectivity: {e}")
return {"error": str(e)}
async def debug_sdk_config() -> Dict[str, Any]:
logger.info("debug_sdk_config util called")
env_vars = {}
for prefix in ['LOOKER_', 'LOOKERSDK_']:
for key, value in os.environ.items():
if key.startswith(prefix):
# Mask sensitive values
if 'SECRET' in key or 'PASSWORD' in key or 'KEY' in key:
masked_value = f"<set: {len(value)} characters>" if value else "<empty string>"
env_vars[key] = masked_value
else:
env_vars[key] = value
# Check for presence of config files
cwd = os.getcwd()
env_file_exists = os.path.exists(os.path.join(cwd, '.env'))
ini_file_exists = os.path.exists(os.path.join(cwd, 'looker.ini'))
sdk_error = None
test_sdk = None
sdk_initialization_working = False
try:
# NOTE: This debug function modifies global SDK state if called.
# It might be better practice to avoid side effects in debug tools.
# Consider if test_sdk.logout() is needed in finally block if login occurs.
logger.debug("Attempting test SDK initialization for debug_sdk_config")
# Use the imported init_looker_sdk
test_sdk = init_looker_sdk()
logger.debug("Testing SDK with me() call for debug_sdk_config")
test_sdk.me() # Simple API call to test authentication
sdk_initialization_working = True
logger.debug("SDK test successful for debug_sdk_config")
except SDKInitializationError as e:
sdk_error = f"SDK initialization failed: {str(e)}"
logger.error(f"Debug SDK test failed (init): {sdk_error}")
except Exception as e:
# Catch errors during the me() call after successful init
sdk_error = f"SDK initialized but API call failed: {str(e)}"
logger.error(f"Debug SDK test failed (API call): {sdk_error}")
# No finally block needed to reset SDK_INSTANCE as this util shouldn't affect the main instance.
return {
"environment_variables": env_vars,
"env_file_exists": env_file_exists,
"looker_ini_exists": ini_file_exists,
"sdk_error_message": sdk_error,
"sdk_initialization_working": sdk_initialization_working,
"runtime_info": {
"python_version": sys.version,
"sdk_version": getattr(looker_sdk, "__version__", "unknown"),
"platform": platform.platform(),
"working_directory": cwd
}
}
