diff --git a/pyproject.toml b/pyproject.toml index 5b5ac7a..4b4790c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,6 +72,7 @@ dependencies = [ "parse==1.20.2", "parso==0.8.4", "pathable==0.4.4", + "platformdirs==4.5.0", "pluggy==1.6.0", "pycparser==2.23", "pydantic==2.11.7", diff --git a/src/seclab_taskflow_agent/mcp_servers/codeql/mcp_server.py b/src/seclab_taskflow_agent/mcp_servers/codeql/mcp_server.py index d692bdc..331ac67 100644 --- a/src/seclab_taskflow_agent/mcp_servers/codeql/mcp_server.py +++ b/src/seclab_taskflow_agent/mcp_servers/codeql/mcp_server.py @@ -20,10 +20,11 @@ import re from urllib.parse import urlparse, unquote import zipfile +from seclab_taskflow_agent.path_utils import mcp_data_dir mcp = FastMCP("CodeQL") -CODEQL_DBS_BASE_PATH = Path(os.getenv('CODEQL_DBS_BASE_PATH', default='/workspaces')) +CODEQL_DBS_BASE_PATH = mcp_data_dir('seclab-taskflow-agent', 'codeql', 'CODEQL_DBS_BASE_PATH') # tool name -> templated query lookup for supported languages TEMPLATED_QUERY_PATHS = { diff --git a/src/seclab_taskflow_agent/mcp_servers/logbook/logbook.py b/src/seclab_taskflow_agent/mcp_servers/logbook/logbook.py index 67b8c8f..8c8bec0 100644 --- a/src/seclab_taskflow_agent/mcp_servers/logbook/logbook.py +++ b/src/seclab_taskflow_agent/mcp_servers/logbook/logbook.py @@ -12,14 +12,13 @@ from fastmcp import FastMCP # move to FastMCP 2.0 import json from pathlib import Path -import os +from seclab_taskflow_agent.path_utils import mcp_data_dir mcp = FastMCP("Logbook") LOG = {} -LOGBOOK = Path(__file__).parent.resolve() / Path(os.getenv('LOGBOOK_STATE_DIR', default='./')) / Path("logbook.json") - +LOGBOOK = mcp_data_dir('seclab-taskflow-agent', 'logbook', 'LOGBOOK_STATE_DIR') / Path("logbook.json") def ensure_log(): global LOG diff --git a/src/seclab_taskflow_agent/mcp_servers/memcache/memcache.py b/src/seclab_taskflow_agent/mcp_servers/memcache/memcache.py index 1a527fb..f8b706c 100644 --- a/src/seclab_taskflow_agent/mcp_servers/memcache/memcache.py +++ b/src/seclab_taskflow_agent/mcp_servers/memcache/memcache.py @@ -16,6 +16,7 @@ from typing import Any from .memcache_backend.dictionary_file import MemcacheDictionaryFileBackend from .memcache_backend.sqlite import SqliteBackend +from seclab_taskflow_agent.path_utils import mcp_data_dir mcp = FastMCP("Memcache") @@ -24,10 +25,7 @@ 'sqlite': SqliteBackend, } -# if MEMCACHE_STATE_DIR contains an absolute path we WANT the user to be able -# to override the relative path in that case this path join will return -# /MEMCACHE_STATE_DIR/memory.json -MEMORY = Path(__file__).parent.resolve() / Path(os.getenv('MEMCACHE_STATE_DIR', default='./')) +MEMORY = mcp_data_dir('seclab-taskflow-agent', 'memcache', 'MEMCACHE_STATE_DIR') BACKEND = os.getenv('MEMCACHE_BACKEND', default='sqlite') backend = backends.get(BACKEND)(str(MEMORY)) diff --git a/src/seclab_taskflow_agent/path_utils.py b/src/seclab_taskflow_agent/path_utils.py new file mode 100644 index 0000000..1175191 --- /dev/null +++ b/src/seclab_taskflow_agent/path_utils.py @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: 2025 GitHub +# SPDX-License-Identifier: MIT + +import platformdirs +import os +from pathlib import Path + + +def mcp_data_dir(packagename: str, mcpname: str, env_override: str | None) -> Path: + """ + Create a directory for an MCP to store its data. + + Parameters: + packagename (str): The name of the package. Used as a subdirectory under the data directory. + mcpname (str): The name of the MCP server. Used as a subdirectory under the package directory. + env_override (str | None): The name of an environment variable that, if set, overrides the default data directory location. If None, the default location is used. + + Returns: + Path: The path to the created data directory for the MCP server. + """ + if env_override: + p = os.getenv(env_override) + if p: + return Path(p) + # Use [platformdirs](https://pypi.org/project/platformdirs/) to + # choose an appropriate location. + d = platformdirs.user_data_dir(appname="seclab-taskflow-agent", + appauthor="GitHubSecurityLab", + ensure_exists=True) + # Each MCP server gets its own sub-directory + p = Path(d).joinpath(packagename).joinpath(mcpname) + p.mkdir(parents=True, exist_ok=True) + return p