Merge pull request #11 from kevinbackhouse/use-path-utils

Use seclab_taskflow_agent.path_utils

Author: kevinbackhouse

src/seclab_taskflows/mcp_servers/codeql_python/mcp_server.py

@@ -3,14 +3,7 @@
 
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_codeql_python.log',
-    filemode='a'
-)
 from seclab_taskflow_agent.mcp_servers.codeql.client import run_query, _debug_log
-# from seclab_taskflow_agent.path_utils import mcp_data_dir
 
 from pydantic import Field
 #from mcp.server.fastmcp import FastMCP, Context
@@ -23,14 +16,20 @@
 from sqlalchemy.orm import Session
 import subprocess
 import importlib.resources
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
 from .codeql_sqlite_models import Base, Source
 from ..utils import process_repo
 
-MEMORY = Path(os.getenv('DATA_DIR', default='/app/data'))
-CODEQL_DBS_BASE_PATH = Path(os.getenv('CODEQL_DBS_BASE_PATH', default='/app/data'))
-# MEMORY =  mcp_data_dir('seclab-taskflows', 'codeql', 'DATA_DIR')
-# CODEQL_DBS_BASE_PATH =  mcp_data_dir('seclab-taskflows', 'codeql', 'CODEQL_DBS_BASE_PATH')
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_codeql_python.log'),
+    filemode='a'
+)
+
+MEMORY = mcp_data_dir('seclab-taskflows', 'codeql', 'DATA_DIR')
+CODEQL_DBS_BASE_PATH = mcp_data_dir('seclab-taskflows', 'codeql', 'CODEQL_DBS_BASE_PATH')
 
 mcp = FastMCP("CodeQL-Python")
 

src/seclab_taskflows/mcp_servers/gh_actions.py

@@ -2,13 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_gh_actions.log',
-    filemode='a'
-)
-
 from fastmcp import FastMCP
 from pydantic import Field
 import httpx
@@ -19,7 +12,14 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 from pathlib import Path
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_gh_actions.log'),
+    filemode='a'
+)
 
 class Base(DeclarativeBase):
     pass
@@ -49,7 +49,7 @@ def __repr__(self):
 if not GITHUB_PERSONAL_ACCESS_TOKEN:
     GITHUB_PERSONAL_ACCESS_TOKEN = os.getenv('COPILOT_TOKEN')
 
-ACTIONS_DB_DIR = Path(os.getenv('ACTIONS_DB_DIR', default='/app/my_data'))
+ACTIONS_DB_DIR = mcp_data_dir('seclab-taskflows', 'gh_actions', 'ACTIONS_DB_DIR')
 
 engine = create_engine(f'sqlite:///{os.path.abspath(ACTIONS_DB_DIR)}/actions.db', echo=False)
 Base.metadata.create_all(engine, tables = [WorkflowUses.__table__])

src/seclab_taskflows/mcp_servers/gh_code_scanning.py

@@ -2,12 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_gh_code_scanning.log',
-    filemode='a'
-)
 from fastmcp import FastMCP
 from pydantic import Field
 import httpx
@@ -20,18 +14,25 @@
 import zipfile
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
 from .alert_results_models import AlertResults, AlertFlowGraph, Base
 
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_gh_code_scanning.log'),
+    filemode='a'
+)
+
 mcp = FastMCP("GitHubCodeScanning")
 
 GITHUB_PERSONAL_ACCESS_TOKEN = os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN', default='')
 if not GITHUB_PERSONAL_ACCESS_TOKEN:
     GITHUB_PERSONAL_ACCESS_TOKEN = os.getenv('COPILOT_TOKEN')
 
-CODEQL_DBS_BASE_PATH = Path(os.getenv('CODEQL_DBS_BASE_PATH', default='/app/my_data'))
-
-ALERT_RESULTS_DIR = Path(os.getenv('ALERT_RESULTS_DIR', default='/app/my_data'))
+CODEQL_DBS_BASE_PATH = mcp_data_dir('seclab-taskflows', 'codeql', 'CODEQL_DBS_BASE_PATH')
+ALERT_RESULTS_DIR = mcp_data_dir('seclab-taskflows', 'gh_code_scanning', 'ALERT_RESULTS_DIR')
 
 def parse_alert(alert: dict) -> dict:
     """Parse the alert dictionary to extract relevant information."""

src/seclab_taskflows/mcp_servers/gh_file_viewer.py

@@ -2,13 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_gh_file_viewer.log',
-    filemode='a'
-)
-
 from fastmcp import FastMCP
 from pydantic import Field
 import httpx
@@ -22,6 +15,14 @@
 import aiofiles
 import zipfile
 import tempfile
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_gh_file_viewer.log'),
+    filemode='a'
+)
 
 class Base(DeclarativeBase):
     pass
@@ -46,7 +47,7 @@ def __repr__(self):
 if not GITHUB_PERSONAL_ACCESS_TOKEN:
     GITHUB_PERSONAL_ACCESS_TOKEN = os.getenv('COPILOT_TOKEN')
 
-SEARCH_RESULT_DIR = Path(os.getenv('SEARCH_RESULTS_DIR', default='/app/my_data'))
+SEARCH_RESULT_DIR = mcp_data_dir('seclab-taskflows', 'gh_file_viewer', 'SEARCH_RESULTS_DIR')
 
 engine = create_engine(f'sqlite:///{os.path.abspath(SEARCH_RESULT_DIR)}/search_result.db', echo=False)
 Base.metadata.create_all(engine, tables = [SearchResults.__table__])

src/seclab_taskflows/mcp_servers/ghsa.py

@@ -1,17 +1,19 @@
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_ghsa.log',
-    filemode='a'
-)
 
 from fastmcp import FastMCP
 from pydantic import Field
 import re
 import json
 from urllib.parse import urlparse, parse_qs
 from .gh_code_scanning import call_api
+from seclab_taskflow_agent.path_utils import log_file_name
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_ghsa.log'),
+    filemode='a'
+)
 
 mcp = FastMCP("GitHubRepoAdvisories")
 

src/seclab_taskflows/mcp_servers/local_file_viewer.py

@@ -2,13 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_local_file_viewer.log',
-    filemode='a'
-)
-
 from fastmcp import FastMCP
 from pydantic import Field
 import httpx
@@ -18,11 +11,18 @@
 import aiofiles
 import zipfile
 import tempfile
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_local_file_viewer.log'),
+    filemode='a'
+)
 
 mcp = FastMCP("LocalFileViewer")
 
-LOCAL_GH_DIR = Path(os.getenv('LOCAL_GH_DIR', default='/app/my_data'))
+LOCAL_GH_DIR = mcp_data_dir('seclab-taskflows', 'local_file_viewer', 'LOCAL_GH_DIR')
 
 def is_subdirectory(directory, potential_subdirectory):
     directory_path = Path(directory)

src/seclab_taskflows/mcp_servers/local_gh_resources.py

@@ -2,13 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_local_gh_resources.log',
-    filemode='a'
-)
-
 from fastmcp import FastMCP
 from pydantic import Field
 import httpx
@@ -18,15 +11,22 @@
 import aiofiles
 import zipfile
 import tempfile
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_local_gh_resources.log'),
+    filemode='a'
+)
 
 mcp = FastMCP("LocalGHResources")
 
 GITHUB_PERSONAL_ACCESS_TOKEN = os.getenv('GITHUB_PERSONAL_ACCESS_TOKEN')
 if not GITHUB_PERSONAL_ACCESS_TOKEN:
     GITHUB_PERSONAL_ACCESS_TOKEN = os.getenv('COPILOT_TOKEN')
 
-LOCAL_GH_DIR = Path(os.getenv('LOCAL_GH_DIR', default='/app/my_data'))
+LOCAL_GH_DIR = mcp_data_dir('seclab-taskflows', 'local_gh_resources', 'LOCAL_GH_DIR')
 
 def is_subdirectory(directory, potential_subdirectory):
     directory_path = Path(directory)

src/seclab_taskflows/mcp_servers/repo_context.py

@@ -2,13 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_repo_context.log',
-    filemode='a'
-)
-
 from fastmcp import FastMCP
 from pydantic import Field
 import httpx
@@ -19,11 +12,19 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 from pathlib import Path
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
 from .repo_context_models import Application, EntryPoint, UserAction, WebEntryPoint, ApplicationIssue, AuditResult, Base
 from .utils import process_repo
 
-MEMORY = Path(os.getenv('REPO_CONTEXT_DIR', default='/app/my_data'))
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_repo_context.log'),
+    filemode='a'
+)
+
+MEMORY = mcp_data_dir('seclab-taskflows', 'repo_context', 'REPO_CONTEXT_DIR')
 
 def app_to_dict(result):
     return {

src/seclab_taskflows/mcp_servers/report_alert_state.py

@@ -2,13 +2,6 @@
 # SPDX-License-Identifier: MIT
 
 import logging
-logging.basicConfig(
-    level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    filename='logs/mcp_report_alert_state.log',
-    filemode='a'
-)
-
 from fastmcp import FastMCP
 import json
 from pathlib import Path
@@ -17,9 +10,17 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 from pydantic import Field
+from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name
 
 from .alert_results_models import AlertResults, AlertFlowGraph, Base
 
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    filename=log_file_name('mcp_report_alert_state.log'),
+    filemode='a'
+)
+
 def result_to_dict(result):
     return {
         "canonical_id": result.canonical_id,
@@ -57,7 +58,7 @@ def remove_line_numbers(location: str) -> str:
     return ':'.join(parts[:-4])
 
 
-MEMORY = Path(os.getenv('ALERT_RESULTS_DIR', default='/app/my_data'))
+MEMORY = mcp_data_dir('seclab-taskflows', 'report_alert_state', 'ALERT_RESULTS_DIR')
 
 class ReportAlertStateBackend:
     def __init__(self, memcache_state_dir: str):