SSB MCP Server

from __future__ import annotations import os import json from dataclasses import dataclass from typing import Optional @dataclass class ServerConfig: # Transport: stdio (default), http, sse transport: str = os.getenv("MCP_TRANSPORT", "stdio") host: str = os.getenv("MCP_HOST", "127.0.0.1") port: int = int(os.getenv("MCP_PORT", "3030")) # Base URL and Tenant (for Cloudera AI Agent Studio) base_url: Optional[str] = os.getenv("SSB_BASE_URL") or None tenant: Optional[str] = os.getenv("SSB_TENANT") or None # Knox + SSB knox_gateway_url: str = os.getenv("KNOX_GATEWAY_URL", "") ssb_api_base: Optional[str] = os.getenv("SSB_API_BASE") # Auth options knox_token: Optional[str] = os.getenv("KNOX_TOKEN") or None knox_cookie: Optional[str] = os.getenv("KNOX_COOKIE") or None knox_user: Optional[str] = os.getenv("KNOX_USER") or None knox_password: Optional[str] = os.getenv("KNOX_PASSWORD") or None knox_token_endpoint: Optional[str] = os.getenv("KNOX_TOKEN_ENDPOINT") or None # Optional passcode token (e.g., Livy/Knox) for alternate auth patterns knox_passcode_token: Optional[str] = os.getenv("KNOX_PASSCODE_TOKEN") or None # Direct SSB authentication (when not using Knox) ssb_user: Optional[str] = os.getenv("SSB_USER") or None ssb_password: Optional[str] = os.getenv("SSB_PASSWORD") or None # TLS/HTTP verify_ssl_env: str = os.getenv("KNOX_VERIFY_SSL", "true").lower() ca_bundle: Optional[str] = os.getenv("KNOX_CA_BUNDLE") timeout_seconds: int = int(os.getenv("HTTP_TIMEOUT_SECONDS", "30")) max_retries: int = int(os.getenv("HTTP_MAX_RETRIES", "3")) rate_limit_rps: float = float(os.getenv("HTTP_RATE_LIMIT_RPS", "5")) # Behavior readonly: bool = os.getenv("SSB_READONLY", "true").lower() == "true" allowed_actions_csv: str = os.getenv("SSB_ALLOWED_ACTIONS", "") # CDP-specific proxy headers proxy_context_path: Optional[str] = os.getenv("SSB_PROXY_CONTEXT_PATH") # MVE API credentials (for Materialized View Engine) mve_api_base: Optional[str] = os.getenv("MVE_API_BASE") mve_username: Optional[str] = os.getenv("MVE_USERNAME") mve_password: Optional[str] = os.getenv("MVE_PASSWORD") @classmethod def from_json_file(cls, config_path: str = "config/cloud_ssb_config.json") -> "ServerConfig": """Load configuration from JSON file with environment variable override.""" config = cls() # Start with environment variables try: with open(config_path, 'r') as f: json_config = json.load(f) # Load cloud_ssb section if it exists cloud_config = json_config.get("cloud_ssb", {}) # Override with JSON values if environment variables are not set if not os.getenv("SSB_BASE_URL") and cloud_config.get("base_url"): config.base_url = cloud_config["base_url"] if not os.getenv("SSB_TENANT") and cloud_config.get("tenant"): config.tenant = cloud_config["tenant"] if not os.getenv("KNOX_GATEWAY_URL") and cloud_config.get("knox_gateway_url"): config.knox_gateway_url = cloud_config["knox_gateway_url"] if not os.getenv("SSB_API_BASE") and cloud_config.get("ssb_api_base"): config.ssb_api_base = cloud_config["ssb_api_base"] # If base_url and tenant are set, clear ssb_api_base to use the new logic if config.base_url and config.tenant and not os.getenv("SSB_API_BASE"): config.ssb_api_base = None if not os.getenv("KNOX_TOKEN") and cloud_config.get("jwt_token"): config.knox_token = cloud_config["jwt_token"] if not os.getenv("KNOX_VERIFY_SSL") and cloud_config.get("knox_verify_ssl") is not None: config.verify_ssl_env = str(cloud_config["knox_verify_ssl"]).lower() if not os.getenv("SSB_READONLY") and cloud_config.get("ssb_readonly") is not None: config.readonly = cloud_config["ssb_readonly"] if not os.getenv("HTTP_TIMEOUT_SECONDS") and cloud_config.get("http_timeout_seconds"): config.timeout_seconds = cloud_config["http_timeout_seconds"] if not os.getenv("HTTP_MAX_RETRIES") and cloud_config.get("http_max_retries"): config.max_retries = cloud_config["http_max_retries"] if not os.getenv("HTTP_RATE_LIMIT_RPS") and cloud_config.get("http_rate_limit_rps"): config.rate_limit_rps = cloud_config["http_rate_limit_rps"] # MVE API credentials if not os.getenv("MVE_API_BASE") and cloud_config.get("mve_api_base"): config.mve_api_base = cloud_config["mve_api_base"] if not os.getenv("MVE_USERNAME") and cloud_config.get("mve_username"): config.mve_username = cloud_config["mve_username"] if not os.getenv("MVE_PASSWORD") and cloud_config.get("mve_password"): config.mve_password = cloud_config["mve_password"] # If base_url and tenant are set, clear mve_api_base to use the new logic if config.base_url and config.tenant and not os.getenv("MVE_API_BASE"): config.mve_api_base = None except FileNotFoundError: # If config file doesn't exist, use environment variables only pass except Exception as e: # Log error but continue with environment variables print(f"Warning: Could not load config file {config_path}: {e}") return config def build_verify(self) -> bool | str: if self.ca_bundle: return self.ca_bundle return self.verify_ssl_env not in {"0", "false", "no"} def build_ssb_base(self) -> str: if self.ssb_api_base: return self.ssb_api_base.rstrip("/") # Try to build from base_url and tenant if self.base_url and self.tenant: base_url = self.base_url.rstrip("/") return f"{base_url}/{self.tenant}/cdp-proxy-token/ssb-sse-api/api/v1" # Fallback to knox_gateway_url if self.knox_gateway_url: return f"{self.knox_gateway_url.rstrip('/')}/irb-ssb-test/cdp-proxy-token/ssb-sse-api/api/v1" # Provide more helpful error message import os env_vars = [] if os.getenv("SSB_API_BASE"): env_vars.append("SSB_API_BASE") if os.getenv("SSB_BASE_URL"): env_vars.append("SSB_BASE_URL") if os.getenv("KNOX_GATEWAY_URL"): env_vars.append("KNOX_GATEWAY_URL") if env_vars: raise ValueError(f"Environment variables {', '.join(env_vars)} are set but not being read properly. Please check your MCP server configuration.") else: raise ValueError("SSB_BASE_URL+SSB_TENANT, SSB_API_BASE, or KNOX_GATEWAY_URL must be set. Please configure your MCP server with the appropriate environment variables.") def get_mve_credentials(self) -> tuple[str, str] | None: """Get MVE API credentials as (username, password) tuple.""" if self.mve_username and self.mve_password: return (self.mve_username, self.mve_password) return None def get_mve_api_base(self) -> str | None: """Get MVE API base URL.""" if self.mve_api_base: return self.mve_api_base # Try to build from base_url and tenant if self.base_url and self.tenant: base_url = self.base_url.rstrip("/") return f"{base_url}/{self.tenant}/cdp-proxy-api/ssb-mve-api/api/v1" return None