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

    @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'),
        )

    @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 _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('"').strip("'")
                if key:
                    data[key] = value
        return data