Coverage for src / qdrant_loader_mcp_server / search / sparse_config.py: 52%

1"""Yaml-loader wrapper that produces a shared :class:`SparseRuntimeConfig`. 

2 

3The dataclass and the validation rules live in 

4``qdrant_loader_core.config``; this module's only job is to read the 

5MCP server's ``MCP_CONFIG`` yaml file and delegate. 

6""" 

7 

8from __future__ import annotations 

9 

10import logging 

11import os 

12from pathlib import Path 

13from typing import Any 

14 

15import yaml 

16from qdrant_loader_core.config import SparseRuntimeConfig 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21def _load_global_section(mcp_config_path: str | None) -> dict[str, Any]: 

22 """Read the ``global`` block from an MCP server yaml config file. 

23 

24 Returns an empty dict on any I/O or parse failure — the config layer 

25 treats that as "no overrides" and falls back to defaults. 

26 """ 

27 if not mcp_config_path: 

28 return {} 

29 path = Path(mcp_config_path) 

30 if not path.exists(): 

31 return {} 

32 try: 

33 with path.open(encoding="utf-8") as f: 

34 data = yaml.safe_load(f) or {} 

35 except Exception as e: 

36 logger.warning("Failed to parse MCP config at %s: %s", path, e) 

37 return {} 

38 if not isinstance(data, dict): 

39 return {} 

40 global_section = data.get("global") 

41 return global_section if isinstance(global_section, dict) else {} 

42 

43 

44def load_sparse_runtime_config( 

45 mcp_config_path: str | None = None, 

46) -> SparseRuntimeConfig: 

47 """Build a :class:`SparseRuntimeConfig` from the MCP server yaml.""" 

48 global_section = _load_global_section(mcp_config_path or os.getenv("MCP_CONFIG")) 

49 return SparseRuntimeConfig.from_global_config(global_section)