mcp/config.py

from dataclasses import dataclass
import os


@dataclass
class GatewayConfig:
    auth_base_url: str
    tools_base_url: str
    client_type: str = 'workbuddy'
    timeout_seconds: int = 10
    refresh_skew_seconds: int = 120
    log_level: str = 'info'
    token_store_path: str = '.mcp_token.json'
    token_store_type: str = 'file'
    redis_host: str = '127.0.0.1'
    redis_port: int = 6379
    redis_db: int = 0
    redis_password: str = ''
    redis_prefix: str = 'fms:mcp:workbuddy:'
    session_key: str = ''
    gateway_mode: str = 'local'
    gateway_session_ttl_seconds: int = 2592000
    rate_limit_enabled: bool = True
    rate_limit_max_requests: int = 60
    rate_limit_window_seconds: int = 60
    max_in_flight_per_tool: int = 2

    @classmethod
    def from_env(cls, env=None, dotenv_path=''):
        primary_env = os.environ if env is None else env
        dotenv_path = str(dotenv_path or '').strip() or os.path.join(os.path.dirname(__file__), '.env')
        dotenv_env = cls._load_dotenv(dotenv_path)
        dotenv_shared_base = (dotenv_env.get('FMS_API_BASE') or '').rstrip('/')
        env_shared_base = (primary_env.get('FMS_API_BASE') or '').rstrip('/')
        auth_base_url = cls._resolve_base_url(
            dotenv_env,
            primary_env,
            dotenv_shared_base,
            env_shared_base,
            'MCP_AUTH_BASE_URL',
            'FMS_AUTH_BASE',
        )
        tools_base_url = cls._resolve_base_url(
            dotenv_env,
            primary_env,
            dotenv_shared_base,
            env_shared_base,
            'MCP_TOOLS_BASE_URL',
            'FMS_TOOLS_BASE',
        )
        timeout_seconds = cls._resolve_timeout_seconds(dotenv_env, primary_env)
        session_key = cls._pick(dotenv_env, primary_env, 'MCP_SESSION_KEY', 'FMS_SESSION_KEY')
        gateway_mode = (cls._pick(dotenv_env, primary_env, 'FMS_GATEWAY_MODE', 'MCP_GATEWAY_MODE') or 'local').lower()
        redis_prefix = cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PREFIX', 'FMS_REDIS_PREFIX')
        if not redis_prefix:
            redis_prefix = 'fms:mcp:gateway:' if gateway_mode == 'public' else 'fms:mcp:workbuddy:'
        return cls(
            auth_base_url=auth_base_url,
            tools_base_url=tools_base_url,
            client_type=cls._pick(dotenv_env, primary_env, 'MCP_CLIENT_TYPE', 'FMS_CLIENT_TYPE') or 'workbuddy',
            timeout_seconds=timeout_seconds,
            refresh_skew_seconds=int(cls._pick(dotenv_env, primary_env, 'MCP_REFRESH_SKEW_SECONDS', 'FMS_REFRESH_SKEW_SECONDS') or '120'),
            log_level=(cls._pick(dotenv_env, primary_env, 'MCP_LOG_LEVEL', 'FMS_LOG_LEVEL') or 'info').lower(),
            token_store_path=cls._pick(dotenv_env, primary_env, 'MCP_TOKEN_STORE_PATH', 'FMS_TOKEN_STORE_PATH') or '.mcp_token.json',
            token_store_type=(cls._pick(dotenv_env, primary_env, 'MCP_TOKEN_STORE', 'FMS_TOKEN_STORE') or 'file').lower(),
            redis_host=cls._pick(dotenv_env, primary_env, 'MCP_REDIS_HOST', 'FMS_REDIS_HOST') or '127.0.0.1',
            redis_port=int(cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PORT', 'FMS_REDIS_PORT') or '6379'),
            redis_db=int(cls._pick(dotenv_env, primary_env, 'MCP_REDIS_DB', 'FMS_REDIS_DB') or '0'),
            redis_password=cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PASSWORD', 'FMS_REDIS_PASSWORD') or '',
            redis_prefix=redis_prefix,
            session_key=session_key or cls._build_default_session_key(primary_env),
            gateway_mode=gateway_mode,
            gateway_session_ttl_seconds=int(cls._pick(dotenv_env, primary_env, 'FMS_GATEWAY_SESSION_TTL_SECONDS', 'MCP_GATEWAY_SESSION_TTL_SECONDS') or '2592000'),
            rate_limit_enabled=cls._parse_bool(cls._pick(dotenv_env, primary_env, 'FMS_RATE_LIMIT_ENABLED', 'MCP_RATE_LIMIT_ENABLED'), default=True),
            rate_limit_max_requests=cls._parse_int(cls._pick(dotenv_env, primary_env, 'FMS_RATE_LIMIT_MAX_REQUESTS', 'MCP_RATE_LIMIT_MAX_REQUESTS'), default=60),
            rate_limit_window_seconds=cls._parse_int(cls._pick(dotenv_env, primary_env, 'FMS_RATE_LIMIT_WINDOW_SECONDS', 'MCP_RATE_LIMIT_WINDOW_SECONDS'), default=60),
            max_in_flight_per_tool=cls._parse_int(cls._pick(dotenv_env, primary_env, 'FMS_MAX_IN_FLIGHT_PER_TOOL', 'MCP_MAX_IN_FLIGHT_PER_TOOL'), default=2),
        )

    @staticmethod
    def _resolve_base_url(preferred_env, fallback_env, preferred_shared_base, fallback_shared_base, *specific_keys):
        for key in specific_keys:
            value = preferred_env.get(key)
            if value:
                return value.rstrip('/')
        if preferred_shared_base:
            return preferred_shared_base
        for key in specific_keys:
            value = fallback_env.get(key)
            if value:
                return value.rstrip('/')
        return fallback_shared_base

    @staticmethod
    def _resolve_timeout_seconds(preferred_env, fallback_env):
        if preferred_env.get('MCP_TIMEOUT_SECONDS'):
            return int(preferred_env['MCP_TIMEOUT_SECONDS'])
        if preferred_env.get('FMS_TIMEOUT_SECONDS'):
            return int(preferred_env['FMS_TIMEOUT_SECONDS'])
        if preferred_env.get('FMS_TIMEOUT_MS'):
            return max(1, int(preferred_env['FMS_TIMEOUT_MS']) // 1000)
        if fallback_env.get('MCP_TIMEOUT_SECONDS'):
            return int(fallback_env['MCP_TIMEOUT_SECONDS'])
        if fallback_env.get('FMS_TIMEOUT_SECONDS'):
            return int(fallback_env['FMS_TIMEOUT_SECONDS'])
        if fallback_env.get('FMS_TIMEOUT_MS'):
            return max(1, int(fallback_env['FMS_TIMEOUT_MS']) // 1000)
        return 10

    @staticmethod
    def _pick(preferred_env, fallback_env, *keys):
        for key in keys:
            value = preferred_env.get(key)
            if value:
                return value
        for key in keys:
            value = fallback_env.get(key)
            if value:
                return value
        return ''

    @staticmethod
    def _strip_comment(value):
        """Strip inline comment from a raw env value (space+# pattern)."""
        pos = str(value or '').find(' #')
        return value[:pos].strip() if pos >= 0 else value

    @staticmethod
    def _parse_int(raw, default):
        """Parse int from env value, tolerating inline comments from OS env vars."""
        value = GatewayConfig._strip_comment(str(raw or '').strip())
        if not value:
            return default
        try:
            return int(value)
        except ValueError:
            return default

    @staticmethod
    def _parse_bool(raw, default=True):
        """Parse bool from env value; empty / unset → default."""
        value = GatewayConfig._strip_comment(str(raw or '').strip()).lower()
        if not value:
            return default
        return value not in ('0', 'false', 'no', 'off')

    @staticmethod
    def _build_default_session_key(env):
        computer = env.get('COMPUTERNAME') or env.get('HOSTNAME') or os.environ.get('COMPUTERNAME') or os.environ.get('HOSTNAME') or 'unknown-computer'
        user = env.get('USERNAME') or env.get('USER') or os.environ.get('USERNAME') or os.environ.get('USER') or 'unknown-user'
        domain = env.get('USERDOMAIN') or os.environ.get('USERDOMAIN') or ''
        parts = [domain, computer, user] if domain else [computer, user]
        return ':'.join(str(part).strip().replace(' ', '_') for part in parts if str(part).strip())

    @staticmethod
    def _load_dotenv(dotenv_path):
        data = {}
        if not dotenv_path or not os.path.exists(dotenv_path):
            return data
        with open(dotenv_path, 'r', encoding='utf-8') as file:
            for raw_line in file:
                line = raw_line.strip()
                if not line or line.startswith('#') or '=' not in line:
                    continue
                key, value = line.split('=', 1)
                key = key.strip()
                value = value.strip()
                # Strip inline comments for unquoted values (e.g. KEY=123  # comment)
                if value and value[0] not in ('"', "'"):
                    comment_pos = value.find(' #')
                    if comment_pos >= 0:
                        value = value[:comment_pos].strip()
                else:
                    value = value.strip('"').strip("'")
                if key:
                    data[key] = value
        return data