refactor: AppConfig via dependency injection

Author: alecron

main.py

@@ -14,17 +14,19 @@
 # Register all tools so they attach to the MCP server
 from utils.mcp_server import run_stdio, run_http
 
+# Load environment variables from .env
+load_dotenv()
+
+app_config = get_app_config()
+
 # Set up logging
 logging.basicConfig(
     format="%(asctime)s-%(process)d-%(levelname)s- %(message)s",
-    level=os.environ.get("LOGLEVEL", "ERROR"),
+    level=app_config.log_level(),
 )
 log = logging.getLogger(__name__)
 
-# Load environment variables from .env
-load_dotenv()
 
-app_config = get_app_config()
 
 
 def handle_signals():
@@ -40,7 +42,7 @@ def signal_handler(sig, frame):
 def main():
     # Choose transport: "stdio" or "sse" (HTTP/SSE)
     handle_signals()
-    transport = os.environ.get("MCP_TRANSPORT", app_config["mcp"]["transport"]).lower()
+    transport = os.environ.get("MCP_TRANSPORT", app_config.transport())
     log.info("""
     ▄▖     ▌▘    ▖  ▖▄▖▄▖  ▄▖
     ▚ ▌▌▛▘▛▌▌▛▌  ▛▖▞▌▌ ▙▌  ▚ █▌▛▘▌▌█▌▛▘

tools/cli_scanner/tool.py

@@ -9,38 +9,37 @@
 import subprocess
 from typing import Literal, Optional
 
-from utils.app_config import get_app_config
+from utils.app_config import AppConfig
 
-logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=os.environ.get("LOGLEVEL", "ERROR"))
 
-log = logging.getLogger(__name__)
-
-# Load app config (expects keys: mcp.host, mcp.port, mcp.transport)
-app_config = get_app_config()
 
 
 class CLIScannerTool:
     """
     A class to encapsulate the tools for interacting with the Sysdig CLI Scanner.
     """
 
-    cmd: str = "sysdig-cli-scanner"
-    default_args: list = [
-        "--loglevel=err",
-        "--apiurl=" + app_config["sysdig"]["host"],
-    ]
-    iac_default_args: list = [
-        "--iac",
-        "--group-by=violation",
-        "--recursive",
-    ]
-
-    exit_code_explained: str = """
-        0: Scan evaluation "pass"
-        1: Scan evaluation "fail"
-        2: Invalid parameters
-        3: Internal error
-        """
+    def __init__(self, app_config: AppConfig):
+        self.app_config = app_config
+        logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=app_config.log_level())
+        self.log = logging.getLogger(__name__)
+        self.cmd: str = "sysdig-cli-scanner"
+        self.default_args: list = [
+            "--loglevel=err",
+            "--apiurl=" + app_config.sysdig_endpoint(),
+        ]
+        self.iac_default_args: list = [
+            "--iac",
+            "--group-by=violation",
+            "--recursive",
+        ]
+
+        self.exit_code_explained: str = """
+            0: Scan evaluation "pass"
+            1: Scan evaluation "fail"
+            2: Invalid parameters
+            3: Internal error
+            """
 
     def check_sysdig_cli_installed(self) -> None:
         """
@@ -49,7 +48,7 @@ def check_sysdig_cli_installed(self) -> None:
         try:
             # Attempt to run 'sysdig-cli-scanner --version' to check if it's installed
             result = subprocess.run([self.cmd, "--version"], capture_output=True, text=True, check=True)
-            log.info(f"Sysdig CLI Scanner is installed: {result.stdout.strip()}")
+            self.log.info(f"Sysdig CLI Scanner is installed: {result.stdout.strip()}")
         except subprocess.CalledProcessError as e:
             error: dict = {
                 "error": "Sysdig CLI Scanner is not installed or not in the $PATH. Check the docs to install it here: https://docs.sysdig.com/en/sysdig-secure/install-vulnerability-cli-scanner/#deployment"
@@ -64,14 +63,14 @@ def check_env_credentials(self) -> None:
             EnvironmentError: If the SYSDIG_SECURE_TOKEN or SYSDIG_HOST environment variables are not set.
         """
         sysdig_secure_token = os.environ.get("SYSDIG_SECURE_TOKEN")
-        sysdig_host = os.environ.get("SYSDIG_HOST", app_config["sysdig"]["host"])
+        sysdig_host = os.environ.get("SYSDIG_HOST", self.app_config.sysdig_endpoint())
         if not sysdig_secure_token:
-            log.error("SYSDIG_SECURE_TOKEN environment variable is not set.")
+            self.log.error("SYSDIG_SECURE_TOKEN environment variable is not set.")
             raise EnvironmentError("SYSDIG_SECURE_TOKEN environment variable is not set.")
         else:
             os.environ["SECURE_API_TOKEN"] = sysdig_secure_token  # Ensure the token is set in the environment
         if not sysdig_host:
-            log.error("SYSDIG_HOST environment variable is not set.")
+            self.log.error("SYSDIG_HOST environment variable is not set.")
             raise EnvironmentError("SYSDIG_HOST environment variable is not set.")
 
     def run_sysdig_cli_scanner(
@@ -124,7 +123,7 @@ def run_sysdig_cli_scanner(
 
         # Prepare the command based on the mode
         if mode == "iac":
-            log.info("Running Sysdig CLI Scanner in IaC mode.")
+            self.log.info("Running Sysdig CLI Scanner in IaC mode.")
             extra_iac_args = [
                 f"--group-by={iac_group_by}",
                 f"--severity-threshold={iac_severity_threshold}",
@@ -135,7 +134,7 @@ def run_sysdig_cli_scanner(
             extra_iac_args = [arg for arg in extra_iac_args if arg]
             cmd = [self.cmd] + self.default_args + self.iac_default_args + extra_iac_args + [path_to_scan]
         else:
-            log.info("Running Sysdig CLI Scanner in vulnerability mode.")
+            self.log.info("Running Sysdig CLI Scanner in vulnerability mode.")
             # Default to vulnerability mode
             extra_args = [
                 "--standalone" if standalone else "",
@@ -161,9 +160,9 @@ def run_sysdig_cli_scanner(
                 }
         # Handle non-zero exit codes speically exit code 1
         except subprocess.CalledProcessError as e:
-            log.warning(f"Sysdig CLI Scanner returned non-zero exit code: {e.returncode}")
+            self.log.warning(f"Sysdig CLI Scanner returned non-zero exit code: {e.returncode}")
             if e.returncode in [2, 3]:
-                log.error(f"Sysdig CLI Scanner encountered an error: {e.stderr.strip()}")
+                self.log.error(f"Sysdig CLI Scanner encountered an error: {e.stderr.strip()}")
                 result: dict = {
                     "error": "Error running Sysdig CLI Scanner",
                     "exit_code": e.returncode,

tools/events_feed/tool.py

@@ -20,22 +20,21 @@
 from fastmcp.server.dependencies import get_http_request
 from utils.query_helpers import create_standard_response
 from utils.sysdig.client_config import get_configuration
-from utils.app_config import get_app_config
+from utils.app_config import AppConfig
 from utils.sysdig.api import initialize_api_client
 
-logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=os.environ.get("LOGLEVEL", "ERROR"))
-log = logging.getLogger(__name__)
-
-# Load app config (expects keys: mcp.host, mcp.port, mcp.transport)
-app_config = get_app_config()
-
 
 class EventsFeedTools:
     """
     A class to encapsulate the tools for interacting with the Sysdig Secure Events Feed API.
     This class provides methods to retrieve event information and list runtime events.
     """
 
+    def __init__(self, app_config: AppConfig):
+        self.app_config = app_config
+        logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=self.app_config.log_level())
+        self.log = logging.getLogger(__name__)
+
     def init_client(self, old_api: bool = False) -> SecureEventsApi | OldSysdigApi:
         """
         Initializes the SecureEventsApi client from the request state.
@@ -48,16 +47,16 @@ def init_client(self, old_api: bool = False) -> SecureEventsApi | OldSysdigApi:
         """
         secure_events_api: SecureEventsApi = None
         old_sysdig_api: OldSysdigApi = None
-        transport = os.environ.get("MCP_TRANSPORT", app_config["mcp"]["transport"]).lower()
+        transport = self.app_config.transport()
         if transport in ["streamable-http", "sse"]:
             # Try to get the HTTP request
-            log.debug("Attempting to get the HTTP request to initialize the Sysdig API client.")
+            self.log.debug("Attempting to get the HTTP request to initialize the Sysdig API client.")
             request: Request = get_http_request()
             secure_events_api = request.state.api_instances["secure_events"]
             old_sysdig_api = request.state.api_instances["old_sysdig_api"]
         else:
             # If running in STDIO mode, we need to initialize the API client from environment variables
-            log.debug("Running in STDIO mode, initializing the Sysdig API client from environment variables.")
+            self.log.debug("Running in STDIO mode, initializing the Sysdig API client from environment variables.")
             cfg = get_configuration()
             api_client = initialize_api_client(cfg)
             secure_events_api = SecureEventsApi(api_client)
@@ -167,15 +166,15 @@ def tool_list_runtime_events(
                 to=to_ts, var_from=from_ts, filter=filter_expr, limit=limit, cursor=cursor
             )
             duration_ms = (time.time() - start_time) * 1000
-            log.debug(f"Execution time: {duration_ms:.2f} ms")
+            self.log.debug(f"Execution time: {duration_ms:.2f} ms")
 
             response = create_standard_response(
                 results=api_response,
                 execution_time_ms=duration_ms,
             )
             return response
         except ToolError as e:
-            log.error(f"Exception when calling SecureEventsApi->get_events_v1: {e}\n")
+            self.log.error(f"Exception when calling SecureEventsApi->get_events_v1: {e}\n")
             raise e
 
     # A tool to retrieve all the process-tree information for a specific event.Add commentMore actions
@@ -199,26 +198,39 @@ def tool_get_event_process_tree(self, event_id: str) -> dict:
             # Get process tree
             tree = old_api_client.request_process_tree_trees(event_id)
 
-            # Parse the response
-            branches = create_standard_response(results=branches, execution_time_ms=(time.time() - start_time) * 1000)
-            tree = create_standard_response(results=tree, execution_time_ms=(time.time() - start_time) * 1000)
+            # Parse the response (tolerates empty bodies)
+            branches_std = create_standard_response(results=branches, execution_time_ms=(time.time() - start_time) * 1000)
+            tree_std = create_standard_response(results=tree, execution_time_ms=(time.time() - start_time) * 1000)
 
             execution_time = (time.time() - start_time) * 1000
 
-            response = (
-                {
-                    "branches": branches.get("results", []),
-                    "tree": tree.get("results", []),
-                    "metadata": {
-                        "execution_time_ms": execution_time,
-                        "timestamp": datetime.utcnow().isoformat() + "Z",
-                    },
+            response = {
+                "branches": branches_std.get("results", {}),
+                "tree": tree_std.get("results", {}),
+                "metadata": {
+                    "execution_time_ms": execution_time,
+                    "timestamp": datetime.utcnow().isoformat() + "Z",
                 },
-            )
+            }
 
             return response
+        except ApiException as e:
+            if e.status == 404:
+                # Process tree not available for this event
+                return {
+                    "branches": {},
+                    "tree": {},
+                    "metadata": {
+                        "execution_time_ms": (time.time() - start_time) * 1000,
+                        "timestamp": datetime.utcnow().isoformat() + "Z",
+                        "note": "Process tree not available for this event"
+                    },
+                }
+            else:
+                self.log.error(f"Exception when calling process tree API: {e}")
+                raise ToolError(f"Failed to get process tree: {e}")
         except ToolError as e:
-            log.error(f"Exception when calling Sysdig Sage API to get process tree: {e}")
+            self.log.error(f"Exception when calling Sysdig Sage API to get process tree: {e}")
             raise e
 
     # Prompts

tools/inventory/tool.py

@@ -3,33 +3,29 @@
 """
 
 import logging
-import os
 import time
 from typing import Annotated
 from pydantic import Field
 from fastmcp.server.dependencies import get_http_request
 from fastmcp.exceptions import ToolError
 from starlette.requests import Request
-from sysdig_client import ApiException
 from sysdig_client.api import InventoryApi
 from utils.sysdig.client_config import get_configuration
-from utils.app_config import get_app_config
+from utils.app_config import AppConfig
 from utils.sysdig.api import initialize_api_client
 from utils.query_helpers import create_standard_response
 
-# Configure logging
-logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=os.environ.get("LOGLEVEL", "ERROR"))
-log = logging.getLogger(__name__)
-
-# Load app config (expects keys: mcp.host, mcp.port, mcp.transport)
-app_config = get_app_config()
-
 
 class InventoryTools:
     """
     A class to encapsulate the tools for interacting with the Sysdig Secure Inventory API.
     This class provides methods to list resources and retrieve a single resource by its hash.
     """
+    def __init__(self, app_config: AppConfig):
+        self.app_config = app_config
+        # Configure logging
+        logging.basicConfig(format="%(asctime)s-%(process)d-%(levelname)s- %(message)s", level=self.app_config.log_level())
+        self.log = logging.getLogger(__name__)
 
     def init_client(self) -> InventoryApi:
         """
@@ -40,15 +36,15 @@ def init_client(self) -> InventoryApi:
             InventoryApi: An instance of the InventoryApi client.
         """
         inventory_api: InventoryApi = None
-        transport = os.environ.get("MCP_TRANSPORT", app_config["mcp"]["transport"]).lower()
+        transport = self.app_config.transport()
         if transport in ["streamable-http", "sse"]:
             # Try to get the HTTP request
-            log.debug("Attempting to get the HTTP request to initialize the Sysdig API client.")
+            self.log.debug("Attempting to get the HTTP request to initialize the Sysdig API client.")
             request: Request = get_http_request()
             inventory_api = request.state.api_instances["inventory"]
         else:
             # If running in STDIO mode, we need to initialize the API client from environment variables
-            log.debug("Running in STDIO mode, initializing the Sysdig API client from environment variables.")
+            self.log.debug("Running in STDIO mode, initializing the Sysdig API client from environment variables.")
             cfg = get_configuration()
             api_client = initialize_api_client(cfg)
             inventory_api = InventoryApi(api_client)
@@ -206,5 +202,5 @@ def tool_get_resource(
 
             return response
         except ToolError as e:
-            log.error(f"Exception when calling InventoryApi->get_resource: {e}")
+            self.log.error(f"Exception when calling InventoryApi->get_resource: {e}")
             raise e