mcp-ntopng

by marcoeg
Verified
"""Environment configuration for the MCP ClntopngickHouse server. This module handles all environment variable configuration with sensible defaults and type conversion. """ from dataclasses import dataclass import os @dataclass class ClickHouseConfig: """Configuration for ntopng ClickHouse database connection settings. This class handles all environment variable configuration related to the ntopng Clickhouse database connection. Required environment variables: NTOPNG_HOST: The hostname of the ClickHouse server NTOPNG_DBUSER: The username for authentication NTOPNG_DBPASSWORD: The password for authentication Optional environment variables (with defaults): NTOPNG_DBPORT: The port number (default: 8443 if secure=True, 8123 if secure=False) NTOPNG_SECURE: Enable HTTPS (default: true) NTOPNG_VERIFY: Verify SSL certificates (default: true) NTOPNG_CONNECT_TIMEOUT: Connection timeout in seconds (default: 30) NTOPNG_SEND_RECEIVE_TIMEOUT: Send/receive timeout in seconds (default: 300) """ def __init__(self): """Initialize the configuration from environment variables.""" self._validate_required_vars() @property def host(self) -> str: """Get the ntopng host.""" return os.environ["NTOPNG_HOST"] @property def port(self) -> int: """Get the ClickHouse port. Defaults to 9440 if secure=True, 9000 if secure=False. Can be overridden by CLICKHOUSE_PORT environment variable. """ if "NTOPNG_DBPORT" in os.environ: return int(os.environ["NTOPNG_DBPORT"]) return 9440 if self.secure else 9000 @property def user(self) -> str: """Get the ClickHouse user.""" return os.environ["NTOPNG_DBUSER"] @property def password(self) -> str: """Get the ClickHouse password.""" return os.environ["NTOPNG_DBPASSWORD"] @property def database(self) -> str: """Get the default database name if set.""" return "ntopng" @property def secure(self) -> bool: """Get whether HTTPS is enabled. Default: False """ return os.getenv("NTOPNG_SECURE", "false").lower() == "true" @property def verify(self) -> bool: """Get whether SSL certificate verification is enabled. Default: True """ return os.getenv("NTOPNG_VERIFY", "true").lower() == "true" @property def connect_timeout(self) -> int: """Get the connection timeout in seconds. Default: 30 """ return int(os.getenv("NTOPNG_CONNECT_TIMEOUT", "30")) @property def send_receive_timeout(self) -> int: """Get the send/receive timeout in seconds. Default: 300 (ClickHouse default) """ return int(os.getenv("NTOPNG_SEND_RECEIVE_TIMEOUT", "300")) def get_client_config(self) -> dict: """Get the configuration dictionary for clickhouse_connect client. Returns: dict: Configuration ready to be passed to clickhouse_connect.get_client() """ config = { "host": self.host, "port": self.port, "user": self.user, "password": self.password, "secure": self.secure, "verify": self.verify, "connect_timeout": self.connect_timeout, "send_receive_timeout": self.send_receive_timeout, "database": "ntopng", "client_name": "mcp_ntopng", } return config def _validate_required_vars(self) -> None: """Validate that all required environment variables are set. Raises: ValueError: If any required environment variable is missing. """ missing_vars = [] for var in ["NTOPNG_HOST", "NTOPNG_DBUSER", "NTOPNG_DBPASSWORD"]: if var not in os.environ: missing_vars.append(var) if missing_vars: raise ValueError( f"Missing required environment variables: {', '.join(missing_vars)}" ) # Global instance for easy access config = ClickHouseConfig()