IDA-MCP

"""IDA-MCP 配置管理模块。 读取 config.conf 配置文件，提供所有可配置项的访问。 配置项 ==================== 协调器配置: - coordinator_host: 协调器监听地址 (默认 127.0.0.1) - coordinator_port: 协调器端口 (默认 11337) HTTP 代理配置: - http_host: HTTP 代理监听地址 (默认 127.0.0.1) - http_port: HTTP 代理端口 (默认 11338) - http_path: MCP 端点路径 (默认 /mcp) IDA 实例配置: - ida_host: IDA 实例 SSE 服务器监听地址 (默认 127.0.0.1) - ida_default_port: IDA 实例 MCP 端口起始值 (默认 10000) 通用配置: - request_timeout: 请求超时时间 (默认 30 秒) - debug: 是否启用调试日志 (默认 false) """ from __future__ import annotations import os from typing import Any, Dict # 配置文件路径 _CONFIG_DIR = os.path.dirname(os.path.abspath(__file__)) _CONFIG_FILE = os.path.join(_CONFIG_DIR, "config.conf") # 默认配置 _DEFAULT_CONFIG = { # 协调器配置 "coordinator_host": "127.0.0.1", "coordinator_port": 11337, # HTTP 代理配置 "http_host": "127.0.0.1", "http_port": 11338, "http_path": "/mcp", # IDA 实例配置 "ida_host": "127.0.0.1", "ida_default_port": 10000, # 通用配置 "request_timeout": 30, "debug": False, } # 缓存的配置 _cached_config: Dict[str, Any] | None = None def _parse_value(value: str) -> Any: """解析配置值，支持字符串、整数、布尔值。""" value = value.strip() # 去除引号 if (value.startswith('"') and value.endswith('"')) or \ (value.startswith("'") and value.endswith("'")): return value[1:-1] # 布尔值 if value.lower() in ("true", "yes", "on", "1"): return True if value.lower() in ("false", "no", "off", "0"): return False # 整数 try: return int(value) except ValueError: pass # 浮点数 try: return float(value) except ValueError: pass return value def load_config(reload: bool = False) -> Dict[str, Any]: """加载配置文件。 参数: reload: 是否强制重新加载 返回: 配置字典 """ global _cached_config if _cached_config is not None and not reload: return _cached_config config = dict(_DEFAULT_CONFIG) if os.path.exists(_CONFIG_FILE): try: with open(_CONFIG_FILE, "r", encoding="utf-8") as f: for line in f: line = line.strip() # 跳过空行和注释 if not line or line.startswith("#"): continue # 解析 key = value if "=" in line: key, value = line.split("=", 1) key = key.strip() # 去除行尾注释 if "#" in value: value = value.split("#", 1)[0] value = _parse_value(value) config[key] = value except Exception: pass _cached_config = config return config # ============================================================================ # 协调器配置访问函数 # ============================================================================ def get_coordinator_host() -> str: """获取协调器监听地址。""" config = load_config() return str(config.get("coordinator_host", "127.0.0.1")) def get_coordinator_port() -> int: """获取协调器端口。""" config = load_config() return int(config.get("coordinator_port", 11337)) def get_coordinator_url() -> str: """获取协调器连接 URL（0.0.0.0 转换为 127.0.0.1）。""" host = get_coordinator_host() # 0.0.0.0 只能用于监听，不能作为连接目标 if host == "0.0.0.0": host = "127.0.0.1" port = get_coordinator_port() return f"http://{host}:{port}" # ============================================================================ # HTTP 代理配置访问函数 # ============================================================================ def get_http_host() -> str: """获取 HTTP 代理监听地址。""" config = load_config() return str(config.get("http_host", "127.0.0.1")) def get_http_port() -> int: """获取 HTTP 代理监听端口。""" config = load_config() return int(config.get("http_port", 11338)) def get_http_path() -> str: """获取 HTTP MCP 端点路径。""" config = load_config() return str(config.get("http_path", "/mcp")) def get_http_url() -> str: """获取完整的 HTTP 代理 URL。""" host = get_http_host() port = get_http_port() path = get_http_path() return f"http://{host}:{port}{path}" # ============================================================================ # IDA 实例配置访问函数 # ============================================================================ def get_ida_host() -> str: """获取 IDA 实例 SSE 服务器监听地址。""" config = load_config() return str(config.get("ida_host", "127.0.0.1")) def get_ida_default_port() -> int: """获取 IDA 实例默认端口起始值。""" config = load_config() return int(config.get("ida_default_port", 10000)) # ============================================================================ # 通用配置访问函数 # ============================================================================ def get_request_timeout() -> int: """获取请求超时时间（秒）。""" config = load_config() return int(config.get("request_timeout", 30)) def is_debug_enabled() -> bool: """是否启用调试日志。""" config = load_config() return bool(config.get("debug", False))