# ============================================================================= # app/config.py # .pyfun parser for app/* modules # Universal MCP Hub (Sandboxed) - based on PyFundaments Architecture # Copyright 2026 - Volkan Kücükbudak # Apache License V. 2 + ESOL 1.1 # ============================================================================= # USAGE in any app/* module: # from . import config # cfg = config.get() # providers = cfg["LLM_PROVIDERS"] # ============================================================================= # USAGE # in providers.py # from . import config # active = config.get_active_llm_providers() # → { "anthropic": { "base_url": "...", "env_key": "ANTHROPIC_API_KEY", ... }, ... } # ============================================================================= # in models.py # from . import config # anthropic_models = config.get_models_for_provider("anthropic") # ============================================================================= # in tools.py # from . import config # active_tools = config.get_active_tools() # ============================================================================= import os import logging from typing import Dict, Any, Optional logger = logging.getLogger('app.config') # Path to .pyfun — lives in app/ next to this file PYFUN_PATH = os.path.join(os.path.dirname(__file__), ".pyfun") # Internal cache — loaded once at first get() _cache: Optional[Dict[str, Any]] = None def _parse_value(value: str) -> str: """Strip quotes and inline comments from a value.""" value = value.strip() # Remove inline comment if " #" in value: value = value[:value.index(" #")].strip() # Strip surrounding quotes if value.startswith('"') and value.endswith('"'): value = value[1:-1] return value def _parse() -> Dict[str, Any]: """ Parses the app/.pyfun file into a nested dictionary. Structure: [SECTION] [SUBSECTION] [BLOCK.name] key = "value" [BLOCK.name_END] [SUBSECTION_END] [SECTION_END] Returns nested dict: { "HUB": { "HUB_NAME": "...", ... }, "LLM_PROVIDERS": { "anthropic": { "active": "true", "base_url": "...", ... }, "gemini": { ... }, }, "MODELS": { "claude-opus-4-6": { "provider": "anthropic", ... }, }, ... } """ if not os.path.isfile(PYFUN_PATH): logger.critical(f".pyfun not found at: {PYFUN_PATH}") raise FileNotFoundError(f".pyfun not found at: {PYFUN_PATH}") result: Dict[str, Any] = {} # Parser state section: Optional[str] = None # e.g. "HUB", "PROVIDERS" subsection: Optional[str] = None # e.g. "LLM_PROVIDERS" block_type: Optional[str] = None # e.g. "LLM_PROVIDER", "MODEL", "TOOL" block_name: Optional[str] = None # e.g. "anthropic", "claude-opus-4-6" with open(PYFUN_PATH, "r", encoding="utf-8") as f: for raw_line in f: line = raw_line.strip() # Skip empty lines and full-line comments if not line or line.startswith("#"): continue # Skip file identifier if line.startswith("[PYFUN_FILE"): continue # --- Block END markers (most specific first) --- if line.endswith("_END]") and "." in line: # e.g. [LLM_PROVIDER.anthropic_END] or [MODEL.claude-opus-4-6_END] block_type = None block_name = None continue if line.endswith("_END]") and not "." in line: # e.g. [LLM_PROVIDERS_END], [HUB_END], [MODELS_END] inner = line[1:-1].replace("_END", "") if subsection and inner == subsection: subsection = None elif section and inner == section: section = None continue # --- Block START markers --- if line.startswith("[") and line.endswith("]"): inner = line[1:-1] # Named block: [LLM_PROVIDER.anthropic] or [MODEL.claude-opus-4-6] if "." in inner: parts = inner.split(".", 1) block_type = parts[0] # e.g. LLM_PROVIDER, MODEL, TOOL block_name = parts[1] # e.g. anthropic, claude-opus-4-6 # Determine which top-level key to store under if block_type == "LLM_PROVIDER": result.setdefault("LLM_PROVIDERS", {}) result["LLM_PROVIDERS"].setdefault(block_name, {}) elif block_type == "SEARCH_PROVIDER": result.setdefault("SEARCH_PROVIDERS", {}) result["SEARCH_PROVIDERS"].setdefault(block_name, {}) elif block_type == "WEB_PROVIDER": result.setdefault("WEB_PROVIDERS", {}) result["WEB_PROVIDERS"].setdefault(block_name, {}) elif block_type == "MODEL": result.setdefault("MODELS", {}) result["MODELS"].setdefault(block_name, {}) elif block_type == "TOOL": result.setdefault("TOOLS", {}) result["TOOLS"].setdefault(block_name, {}) continue # Subsection: [LLM_PROVIDERS], [SEARCH_PROVIDERS] etc. if section and not subsection: subsection = inner result.setdefault(inner, {}) continue # Top-level section: [HUB], [PROVIDERS], [MODELS] etc. section = inner result.setdefault(inner, {}) continue # --- Key = Value --- if "=" in line: key, _, val = line.partition("=") key = key.strip() val = _parse_value(val) # Strip provider prefix from key (e.g. "anthropic.base_url" → "base_url") if block_name and key.startswith(f"{block_name}."): key = key[len(block_name) + 1:] # Store in correct location if block_type and block_name: if block_type == "LLM_PROVIDER": result["LLM_PROVIDERS"][block_name][key] = val elif block_type == "SEARCH_PROVIDER": result["SEARCH_PROVIDERS"][block_name][key] = val elif block_type == "WEB_PROVIDER": result["WEB_PROVIDERS"][block_name][key] = val elif block_type == "MODEL": result["MODELS"][block_name][key] = val elif block_type == "TOOL": result["TOOLS"][block_name][key] = val elif section: result[section][key] = val logger.info(f".pyfun loaded. Sections: {list(result.keys())}") return result def load() -> Dict[str, Any]: """Force (re)load of .pyfun — clears cache.""" global _cache _cache = _parse() return _cache def get() -> Dict[str, Any]: """ Returns parsed .pyfun config as nested dict. Loads and caches on first call — subsequent calls return cache. """ global _cache if _cache is None: _cache = _parse() return _cache def get_section(section: str) -> Dict[str, Any]: """ Returns a specific top-level section. Returns empty dict if section not found. """ return get().get(section, {}) def get_llm_providers() -> Dict[str, Any]: """Returns all LLM providers (active and inactive).""" return get().get("LLM_PROVIDERS", {}) def get_active_llm_providers() -> Dict[str, Any]: """Returns only LLM providers where active = 'true'.""" return { name: cfg for name, cfg in get_llm_providers().items() if cfg.get("active", "false").lower() == "true" } def get_search_providers() -> Dict[str, Any]: """Returns all search providers.""" return get().get("SEARCH_PROVIDERS", {}) def get_active_search_providers() -> Dict[str, Any]: """Returns only search providers where active = 'true'.""" return { name: cfg for name, cfg in get_search_providers().items() if cfg.get("active", "false").lower() == "true" } def get_models() -> Dict[str, Any]: """Returns all model definitions.""" return get().get("MODELS", {}) def get_models_for_provider(provider_name: str) -> Dict[str, Any]: """Returns all models for a specific provider.""" return { name: cfg for name, cfg in get_models().items() if cfg.get("provider", "") == provider_name } def get_tools() -> Dict[str, Any]: """Returns all tool definitions.""" return get().get("TOOLS", {}) def get_active_tools() -> Dict[str, Any]: """Returns only tools where active = 'true'.""" return { name: cfg for name, cfg in get_tools().items() if cfg.get("active", "false").lower() == "true" } def get_hub() -> Dict[str, Any]: """Returns [HUB] section.""" return get_section("HUB") def get_limits() -> Dict[str, Any]: """Returns [HUB_LIMITS] section.""" return get_section("HUB_LIMITS") def get_db_sync() -> Dict[str, Any]: """Returns [DB_SYNC] section.""" return get_section("DB_SYNC") def get_debug() -> Dict[str, Any]: """Returns [DEBUG] section.""" return get_section("DEBUG") def is_debug() -> bool: """Returns True if DEBUG = 'ON' in .pyfun.""" return get_debug().get("DEBUG", "OFF").upper() == "ON"