diff --git a/scripts/infra-mcp/pyproject.toml b/scripts/infra-mcp/pyproject.toml index 0fa4e84c..d81c97db 100644 --- a/scripts/infra-mcp/pyproject.toml +++ b/scripts/infra-mcp/pyproject.toml @@ -1,8 +1,10 @@ [project] name = "infra-mcp" version = "0.1.0" -description = "MCP server for task runner integration" +description = "MCP server for task runner integration, container operations, and infrastructure management" +readme = "README.md" requires-python = ">=3.13" +license = { text = "MIT" } dependencies = [ "fastmcp>=2.10.0,<3", "requests>=2.32.4,<3", diff --git a/scripts/infra-mcp/server.py b/scripts/infra-mcp/server.py index ce71ef22..850ab178 100755 --- a/scripts/infra-mcp/server.py +++ b/scripts/infra-mcp/server.py @@ -8,6 +8,7 @@ import io import logging import os +import signal import sys from fastmcp import FastMCP @@ -20,11 +21,17 @@ from tools.get_container_categories import ContainerCategoryFinder from tools.get_container_tags import ContainerTagFinder from tools.get_dashboard_groups import DashboardGroupFinder +from utils.constants import DEFAULT_CONTAINER_ARCHITECTURE, DEFAULT_SAME_HASH_LIMIT, DEFAULT_TAG_LIMIT from utils.git import get_git_root +from utils.models import ContainerTagFinderArgs from utils.security import validate_url_for_ssrf -# Configure logging -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") +# Configure logging from environment +LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO") +logging.basicConfig( + level=getattr(logging, LOG_LEVEL), + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) logger = logging.getLogger("infra-mcp") logger.info("Starting Infra MCP server") @@ -36,6 +43,7 @@ @mcp.custom_route("/healthz", methods=["GET"]) async def health_check(_request: Request) -> PlainTextResponse: + """Health check endpoint for monitoring server status.""" return PlainTextResponse("OK") @@ -96,7 +104,7 @@ def get_container_categories() -> list[str]: @mcp.tool(name="list-container-tags") -def list_container_tags(image: str, limit: int = 10) -> list[str]: +def list_container_tags(image: str, limit: int = DEFAULT_TAG_LIMIT) -> list[str]: """ List recent tags for a container image. @@ -109,26 +117,23 @@ def list_container_tags(image: str, limit: int = 10) -> list[str]: """ tag_finder = ContainerTagFinder() try: - # Create a namespace to simulate command line args - class Args: - pass - - args = Args() - args.image = image - args.architecture = "linux/amd64" - args.limit = limit - args.quiet = True - args.registry = None + args = ContainerTagFinderArgs( + image=image, + architecture=DEFAULT_CONTAINER_ARCHITECTURE, + limit=limit, + quiet=True, + registry=None, + ) tags, _, _, _ = tag_finder.get_image_tags(args) return [tag["name"] for tag in tags[:limit]] if tags else [] except Exception: - logger.exception("list-container-tags failed for image=%r", image) + logger.exception(f"list-container-tags failed for image={image!r}") return [] @mcp.tool(name="list-same-hash-container-tags") -def list_same_hash_container_tags(image: str, tag: str | None = None, limit: int = 100) -> list[str]: +def list_same_hash_container_tags(image: str, tag: str | None = None, limit: int = DEFAULT_SAME_HASH_LIMIT) -> list[str]: """ List tags that have the same hash as a specified container tag. @@ -142,28 +147,25 @@ def list_same_hash_container_tags(image: str, tag: str | None = None, limit: int """ tag_finder = ContainerTagFinder() try: - # Create a namespace to simulate command line args - class Args: - pass - - args = Args() - args.image = image - args.tag = tag - args.architecture = "linux/amd64" - args.limit = limit - args.quiet = True - args.registry = None + args = ContainerTagFinderArgs( + image=image, + tag=tag, + architecture=DEFAULT_CONTAINER_ARCHITECTURE, + limit=limit, + quiet=True, + registry=None, + ) # Get same hash tags but don't output to stdout same_hash_tags = tag_finder.list_same_hash_tags(args, suppress_output=True) return [tag["name"] for tag in same_hash_tags] if same_hash_tags else [] except Exception: - logger.exception("list-same-hash-container-tags failed for image=%r tag=%r", image, tag) + logger.exception(f"list-same-hash-container-tags failed for image={image!r} tag={tag!r}") return [] @mcp.tool(name="get-most-specific-container-tag") -def get_most_specific_container_tag(image: str, tag: str | None = None, limit: int = 100) -> str: +def get_most_specific_container_tag(image: str, tag: str | None = None, limit: int = DEFAULT_SAME_HASH_LIMIT) -> str: """ Find the most specific container version tag from tags with the same hash. @@ -177,35 +179,49 @@ def get_most_specific_container_tag(image: str, tag: str | None = None, limit: i """ tag_finder = ContainerTagFinder() try: - # Create a namespace to simulate command line args - class Args: - pass - - args = Args() - args.image = image - args.tag = tag - args.architecture = "linux/amd64" - args.limit = limit - args.quiet = True - args.registry = None + args = ContainerTagFinderArgs( + image=image, + tag=tag, + architecture=DEFAULT_CONTAINER_ARCHITECTURE, + limit=limit, + quiet=True, + registry=None, + ) # Suppress any prints from the finder with contextlib.redirect_stdout(io.StringIO()): most_specific = tag_finder.get_most_specific_tag(args) same_hash = tag_finder.list_same_hash_tags(args, suppress_output=True) + except Exception: + logger.exception(f"get-most-specific-container-tag failed for image={image!r} tag={tag!r}") + return tag or "latest" + else: if most_specific: return most_specific["name"] - elif same_hash: + if same_hash: return same_hash[0]["name"] - else: - return tag or "latest" - except Exception: - logger.exception("get-most-specific-container-tag failed for image=%r tag=%r", image, tag) return tag or "latest" # --- Configure the FastMCP server --- + +def handle_shutdown(signum: int, _frame: object) -> None: + """Handle shutdown signals gracefully. + + Args: + signum: Signal number received + _frame: Current stack frame (unused) + """ + signal_name = signal.Signals(signum).name + logger.info(f"Received {signal_name}, shutting down gracefully...") + sys.exit(0) + + +# Register signal handlers for graceful shutdown +signal.signal(signal.SIGINT, handle_shutdown) +signal.signal(signal.SIGTERM, handle_shutdown) + try: # Get the repository root path repository_root_path = get_git_root() @@ -227,12 +243,22 @@ class Args: add_container_operation_tools(mcp, repository_root_path) else: logger.info("Container operation tools disabled by environment variable") -except Exception: # noqa: BLE001 - logger.exception("Failed to initialize server") +except FileNotFoundError: + logger.exception("Repository not found") + sys.exit(1) +except RuntimeError: + logger.exception("Git operation failed") + sys.exit(1) +except ImportError: + logger.exception("Failed to import required module") + sys.exit(1) +except Exception: + logger.exception("Unexpected error during server initialization") sys.exit(1) -def main(): +def main() -> None: + """Start the MCP server.""" # Start the server mcp.run() # Or start with parameters: diff --git a/scripts/infra-mcp/tools/collections/container_tools.py b/scripts/infra-mcp/tools/collections/container_tools.py index c09e562b..4600ed9b 100644 --- a/scripts/infra-mcp/tools/collections/container_tools.py +++ b/scripts/infra-mcp/tools/collections/container_tools.py @@ -9,15 +9,23 @@ import subprocess import sys from collections.abc import Callable +from pathlib import Path from fastmcp import FastMCP from fastmcp.tools import Tool +# Import constants from the shared constants module +try: + from ...utils.constants import TASK_COMMAND_TIMEOUT +except ImportError: + # Fallback for standalone execution + TASK_COMMAND_TIMEOUT = 600 + # Configure logging logger = logging.getLogger("infra-mcp") -def get_container_operations(): +def get_container_operations() -> list[dict[str, str]]: """ Get the list of valid container operations @@ -34,7 +42,7 @@ def get_container_operations(): ] -def execute_container_operation(operation: str, service_name: str, repository_root_path: str) -> str: +def execute_container_operation(operation: str, service_name: str, repository_root_path: str | Path) -> str: """ Execute one operation on the specified service and return the output @@ -55,9 +63,10 @@ def execute_container_operation(operation: str, service_name: str, repository_ro if not re.match(r"^[a-zA-Z0-9_/-]+$", service_name): return f"Invalid service name format: {service_name}" + repository_root_str = str(repository_root_path) cmd = [ sys.executable, # Use the current Python interpreter - os.path.join(repository_root_path, "scripts", "labctl.py"), + os.path.join(repository_root_str, "scripts", "labctl.py"), "service", operation, service_name, @@ -65,7 +74,11 @@ def execute_container_operation(operation: str, service_name: str, repository_ro try: result = subprocess.run( # noqa: S603 - cmd, capture_output=True, text=True, check=True + cmd, + capture_output=True, + text=True, + check=True, + timeout=TASK_COMMAND_TIMEOUT, ) except subprocess.CalledProcessError as e: return f"Error running operation: {e.stderr or str(e)}" @@ -73,7 +86,7 @@ def execute_container_operation(operation: str, service_name: str, repository_ro return result.stdout or "(No output)" -def create_operation_function(op: str, repository_root_path: str) -> Callable[[str], str]: +def create_operation_function(op: str, repository_root_path: str | Path) -> Callable[[str], str]: """ Create a function that executes a specific container service operation. @@ -100,7 +113,7 @@ def operation_fn(service_name: str) -> str: return operation_fn -def add_container_operation_tools(mcp_server: FastMCP, repository_root_path: str) -> None: +def add_container_operation_tools(mcp_server: FastMCP, repository_root_path: str | Path) -> None: """ Create and add tools to MCP server for container service operations. diff --git a/scripts/infra-mcp/tools/collections/task_tools.py b/scripts/infra-mcp/tools/collections/task_tools.py index 73a606eb..943fe642 100644 --- a/scripts/infra-mcp/tools/collections/task_tools.py +++ b/scripts/infra-mcp/tools/collections/task_tools.py @@ -8,15 +8,23 @@ import shutil import subprocess from collections.abc import Callable +from pathlib import Path from fastmcp import FastMCP from fastmcp.tools import Tool +# Import constants from the shared constants module +try: + from ...utils.constants import TASK_COMMAND_TIMEOUT +except ImportError: + # Fallback for standalone execution + TASK_COMMAND_TIMEOUT = 600 + # Configure logging logger = logging.getLogger("infra-mcp") -def get_task_list(repository_root_path: str) -> list[dict[str, str]]: +def get_task_list(repository_root_path: str | Path) -> list[dict[str, str]]: """ Get the list of available tasks by running 'task --list-all'. @@ -33,10 +41,11 @@ def get_task_list(repository_root_path: str) -> list[dict[str, str]]: return [] try: result = subprocess.run( # noqa: S603 - [task_bin, "--list-all", "--dir", repository_root_path], + [task_bin, "--list-all", "--dir", str(repository_root_path)], capture_output=True, text=True, check=True, + timeout=TASK_COMMAND_TIMEOUT, ) except subprocess.CalledProcessError: logger.exception("Error getting task list") @@ -58,7 +67,7 @@ def get_task_list(repository_root_path: str) -> list[dict[str, str]]: return tasks -def execute_task(task_name: str, repository_root_path: str) -> str: +def execute_task(task_name: str, repository_root_path: str | Path) -> str: """ Execute a task command and return the output. @@ -76,17 +85,18 @@ def execute_task(task_name: str, repository_root_path: str) -> str: return f"Error executing task {task_name}: 'task' binary not found" try: return subprocess.run( # noqa: S603 - [task_bin, task_name, "--dir", repository_root_path], + [task_bin, task_name, "--dir", str(repository_root_path)], capture_output=True, text=True, check=True, + timeout=TASK_COMMAND_TIMEOUT, ).stdout.strip() except subprocess.CalledProcessError as e: logger.exception(f"Error executing task {task_name}") return f"Error executing task {task_name}: {e.stderr}" -def create_task_function(task_name: str, repository_root_path: str) -> Callable[[], str]: +def create_task_function(task_name: str, repository_root_path: str | Path) -> Callable[[], str]: """ Create a function that executes a specific task. @@ -104,7 +114,7 @@ def task_fn() -> str: return task_fn -def add_task_tools(mcp_server: FastMCP, repository_root_path: str) -> None: +def add_task_tools(mcp_server: FastMCP, repository_root_path: str | Path) -> None: """ Get list of tasks, then create and add tools to MCP server for each task. @@ -116,7 +126,7 @@ def add_task_tools(mcp_server: FastMCP, repository_root_path: str) -> None: for task_info in tasks: task_name = task_info["name"] - tool_name = task_name.replace(":", "--") + tool_name = task_name.replace(":", "-") description = task_info["description"] task_fn = create_task_function(task_name, repository_root_path) diff --git a/scripts/infra-mcp/tools/get_app_icon.py b/scripts/infra-mcp/tools/get_app_icon.py index b6ef994d..d764d9d4 100755 --- a/scripts/infra-mcp/tools/get_app_icon.py +++ b/scripts/infra-mcp/tools/get_app_icon.py @@ -8,6 +8,13 @@ import requests from bs4 import BeautifulSoup +# Import constants from the shared constants module +try: + from ..utils.constants import DEFAULT_REQUEST_TIMEOUT +except ImportError: + # Fallback for standalone execution + DEFAULT_REQUEST_TIMEOUT = 10 + class AppIconFinder: """ @@ -23,13 +30,13 @@ def __init__(self): "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } - def get_app_icon(self, app_name, homepage_url): + def get_app_icon(self, app_name: str, homepage_url: str) -> str: """ Main function to get an application icon. Args: - app_name (str): The name of the application. - homepage_url (str): The URL of the application's homepage. + app_name: The name of the application. + homepage_url: The URL of the application's homepage. Returns: str: Either the icon filename (e.g., "github.png") if found in the dashboard-icons set, @@ -48,7 +55,15 @@ def get_app_icon(self, app_name, homepage_url): # Return default if no favicon found return "default" - def _find_dashboard_icon(self, app_name): + def _find_dashboard_icon(self, app_name: str) -> str | None: + """Find an icon from the dashboard-icons repository. + + Args: + app_name: The name of the application + + Returns: + str | None: The icon filename if found, None otherwise + """ normalized_name = app_name.lower().replace(" ", "-") icon_name = f"{normalized_name}.png" url = f"https://cdn.jsdelivr.net/gh/homarr-labs/dashboard-icons/png/{icon_name}" @@ -56,7 +71,7 @@ def _find_dashboard_icon(self, app_name): response = requests.head( url, headers=self.headers, - timeout=10, + timeout=DEFAULT_REQUEST_TIMEOUT, allow_redirects=True, ) if response.ok: @@ -64,17 +79,79 @@ def _find_dashboard_icon(self, app_name): # Some CDNs/origins may disallow HEAD or require GET. if response.status_code in (403, 405): # Use GET fallback for servers that disallow HEAD; ensure connection is closed. - with requests.get(url, headers=self.headers, timeout=10) as probe: + with requests.get(url, headers=self.headers, timeout=DEFAULT_REQUEST_TIMEOUT) as probe: if probe.ok: return icon_name - else: - return None - else: - return None except requests.RequestException: + pass + return None + + def _parse_html_for_favicon(self, soup: BeautifulSoup, homepage_url: str) -> str | None: + """Parse HTML to find favicon link in meta tags. + + Args: + soup: BeautifulSoup object containing parsed HTML + homepage_url: Base URL for resolving relative links + + Returns: + str | None: Absolute favicon URL if found, None otherwise + """ + # Look for link tags with rel="icon" or rel="shortcut icon" + icon_links = soup.find_all("link", rel=re.compile(r"(shortcut icon|icon|apple-touch-icon)", re.I)) + if not icon_links: return None - def _find_favicon_url(self, homepage_url): + # Sort by preference: apple-touch-icon > icon > shortcut icon + def get_priority(link): + rel_attr = link.get("rel", []) + if isinstance(rel_attr, str): + rel_attr = [rel_attr] + rel_lower = " ".join(rel_attr).lower() + if "apple-touch-icon" in rel_lower: + return 3 + if "icon" in rel_lower and "shortcut" not in rel_lower: + return 2 + return 1 + + icon_links = sorted(icon_links, key=get_priority, reverse=True) + for link in icon_links: + if "href" in link.attrs: + # Make relative URLs absolute + return urljoin(homepage_url, link["href"]) + return None + + def _check_default_favicon(self, homepage_url: str) -> str | None: + """Check for favicon.ico at the default location. + + Args: + homepage_url: Base URL to check + + Returns: + str | None: Absolute URL to favicon.ico if exists, None otherwise + """ + default_favicon = urljoin(homepage_url, "/favicon.ico") + try: + favicon_response = requests.head( + default_favicon, + headers=self.headers, + timeout=5, + allow_redirects=True, + ) + if favicon_response.ok: + return default_favicon + except requests.RequestException: + pass + return None + + def _find_favicon_url(self, homepage_url: str) -> str | None: + """Find a favicon URL by parsing the homepage. + + Args: + homepage_url: The URL of the application's homepage + + Returns: + str | None: The favicon URL if found, None otherwise + """ try: if not homepage_url: return None @@ -82,46 +159,22 @@ def _find_favicon_url(self, homepage_url): # Make sure URL has a scheme homepage_url = homepage_url.strip() if not homepage_url.startswith(("http://", "https://")): - homepage_url = "https://" + homepage_url + homepage_url = f"https://{homepage_url}" # Fetch the homepage - response = requests.get(homepage_url, headers=self.headers, timeout=10) + response = requests.get(homepage_url, headers=self.headers, timeout=DEFAULT_REQUEST_TIMEOUT) response.raise_for_status() # Parse the HTML soup = BeautifulSoup(response.text, "html.parser") - # Look for favicon in different ways - # 1. Check for link tags with rel="icon" or rel="shortcut icon" - icon_links = soup.find_all("link", rel=re.compile(r"(shortcut icon|icon|apple-touch-icon)", re.I)) - if icon_links: - # Sort by preference: apple-touch-icon > icon > shortcut icon - def get_priority(link): - rel_attr = link.get("rel", []) - if isinstance(rel_attr, str): - rel_attr = [rel_attr] - rel_lower = " ".join(rel_attr).lower() - if "apple-touch-icon" in rel_lower: - return 3 - elif "icon" in rel_lower and "shortcut" not in rel_lower: - return 2 - else: - return 1 - - icon_links = sorted(icon_links, key=get_priority, reverse=True) - for link in icon_links: - if "href" in link.attrs: - # Make relative URLs absolute - favicon_url = urljoin(homepage_url, link["href"]) - return favicon_url - - # 2. Check for the default location - default_favicon = urljoin(homepage_url, "/favicon.ico") - favicon_response = requests.head(default_favicon, headers=self.headers, timeout=5, allow_redirects=True) - if favicon_response.ok: - return default_favicon - else: - return None + # Try to find favicon in HTML meta tags + favicon_url = self._parse_html_for_favicon(soup, homepage_url) + if favicon_url: + return favicon_url + + # Check for the default location + return self._check_default_favicon(homepage_url) except Exception as e: print(f"Error finding favicon: {e}", file=sys.stderr) return None diff --git a/scripts/infra-mcp/tools/get_container_categories.py b/scripts/infra-mcp/tools/get_container_categories.py index c4b883a4..d65721d4 100755 --- a/scripts/infra-mcp/tools/get_container_categories.py +++ b/scripts/infra-mcp/tools/get_container_categories.py @@ -4,11 +4,11 @@ import sys from pathlib import Path +# Use relative import for package structure try: - # Try importing with absolute path first (when imported from server.py) - from utils.git import get_git_root -except ModuleNotFoundError: - # If that fails, try relative import (when run as a standalone script) + from ..utils.git import get_git_root +except ImportError: + # When run as standalone script, adjust path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from utils.git import get_git_root @@ -18,7 +18,7 @@ class ContainerCategoryFinder: A class for finding container categories in the docker directory. """ - def __init__(self): + def __init__(self) -> None: """ Initialize the ContainerCategoryFinder. """ @@ -74,9 +74,9 @@ def get_container_categories(self) -> list[str]: raise RuntimeError(f"Error finding container categories: {str(e)}") from None -def main(): +def main() -> None: """ - Run the application: read homepage settings and print dashboard groups. + Parse CLI args, find container categories from the docker directory, and print each path. """ try: finder = ContainerCategoryFinder() diff --git a/scripts/infra-mcp/tools/get_container_tags.py b/scripts/infra-mcp/tools/get_container_tags.py index 9e7d588e..daed7fa2 100755 --- a/scripts/infra-mcp/tools/get_container_tags.py +++ b/scripts/infra-mcp/tools/get_container_tags.py @@ -8,6 +8,14 @@ import requests +# Import constants from the shared constants module +try: + from ..utils.constants import MAX_TAGS_FETCH_LIMIT, REGISTRY_REQUEST_TIMEOUT +except ImportError: + # Fallback for standalone execution + REGISTRY_REQUEST_TIMEOUT = 30 + MAX_TAGS_FETCH_LIMIT = 1000 + class ContainerTagFinder: """ @@ -235,6 +243,57 @@ def get_docker_hub_tags( else: return tag_data + def _fetch_manifest_for_tag( + self, + registry_url: str, + image_name: str, + tag: str, + architecture: str, + ) -> dict[str, Any]: + """Fetch manifest information for a specific tag. + + Args: + registry_url: URL of the registry + image_name: Name of the image + tag: Tag to fetch manifest for + architecture: Architecture to filter by (e.g., 'linux/amd64') + + Returns: + dict: Tag data with name, last_updated, and digest fields + """ + manifest_url = f"{registry_url}/v2/{image_name}/manifests/{tag}" + try: + # Try to get the manifest to extract creation time + headers = {"Accept": "application/vnd.docker.distribution.manifest.v2+json"} + manifest_response = requests.get(manifest_url, headers=headers, timeout=REGISTRY_REQUEST_TIMEOUT) + manifest_response.raise_for_status() + + # Get digest from the manifest + manifest = manifest_response.json() + + # For multi-arch images, we need to find the digest for the specific architecture + digest = None + # Try to parse the architecture from the manifest if it's a multi-arch image + if "manifests" in manifest: + os_part, arch_part = self._parse_arch(architecture) + for m in manifest.get("manifests", []): + platform = m.get("platform", {}) + if platform.get("architecture") == arch_part and platform.get("os") == os_part: + digest = m.get("digest") + break + else: + # If it's not a multi-arch manifest, just use the digest directly + digest = manifest_response.headers.get("Docker-Content-Digest") + + # Most registry implementations don't expose creation time directly in the API + # We'll use the response headers as a rough proxy for recency + last_modified = manifest_response.headers.get("Last-Modified") + except requests.exceptions.RequestException: + # If we can't get detailed info, just use the tag name + last_modified = None + digest = None + return {"name": tag, "last_updated": last_modified, "digest": digest} + def get_registry_tags( self, registry_url: str, @@ -257,46 +316,17 @@ def get_registry_tags( """ url: str = f"{registry_url}/v2/{image_name}/tags/list" try: - response = requests.get(url, timeout=30) + response = requests.get(url, timeout=REGISTRY_REQUEST_TIMEOUT) response.raise_for_status() data = response.json() tags: list[str] = data.get("tags", []) # For Docker Registry API v2, we need to make additional requests to get manifest and timestamps tag_data: list[dict[str, Any]] = [] - for tag in tags[:100]: # Limit the number of additional requests - manifest_url = f"{registry_url}/v2/{image_name}/manifests/{tag}" - try: - # Try to get the manifest to extract creation time - headers = {"Accept": "application/vnd.docker.distribution.manifest.v2+json"} - manifest_response = requests.get(manifest_url, headers=headers, timeout=30) - manifest_response.raise_for_status() - - # Get digest from the manifest - manifest = manifest_response.json() - - # For multi-arch images, we need to find the digest for the specific architecture - digest = None - # Try to parse the architecture from the manifest if it's a multi-arch image - if "manifests" in manifest: - for m in manifest.get("manifests", []): - if ( - m.get("platform", {}).get("architecture") == architecture.split("/")[1] - and m.get("platform", {}).get("os") == architecture.split("/", maxsplit=1)[0] - ): - digest = m.get("digest") - break - else: - # If it's not a multi-arch manifest, just use the digest directly - digest = manifest_response.headers.get("Docker-Content-Digest") - - # Most registry implementations don't expose creation time directly in the API - # We'll use the response headers as a rough proxy for recency - last_modified = manifest_response.headers.get("Last-Modified") - tag_data.append({"name": tag, "last_updated": last_modified, "digest": digest}) - except requests.exceptions.RequestException: - # If we can't get detailed info, just use the tag name - tag_data.append({"name": tag, "last_updated": None, "digest": None}) + # Limit the number of additional requests + for tag in tags[:100]: + tag_info = self._fetch_manifest_for_tag(registry_url, image_name, tag, architecture) + tag_data.append(tag_info) # Sort based on sort_by parameter self._sort_tags(tag_data, sort_by) diff --git a/scripts/infra-mcp/tools/get_dashboard_groups.py b/scripts/infra-mcp/tools/get_dashboard_groups.py index bfbd9d58..9eaf8e9b 100755 --- a/scripts/infra-mcp/tools/get_dashboard_groups.py +++ b/scripts/infra-mcp/tools/get_dashboard_groups.py @@ -1,16 +1,17 @@ #!/usr/bin/env python3 -import os import sys from pathlib import Path import yaml +# Use relative import for package structure try: - # Try importing with absolute path first (when imported from server.py) - from utils.git import get_git_root -except ModuleNotFoundError: - # If that fails, try relative import (when run as a standalone script) + from ..utils.git import get_git_root +except ImportError: + # When run as standalone script, adjust path + import os + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from utils.git import get_git_root @@ -20,7 +21,7 @@ class DashboardGroupFinder: A class for finding dashboard groups from homepage settings. """ - def __init__(self): + def __init__(self) -> None: """ Initialize the DashboardGroupFinder. """ @@ -32,10 +33,10 @@ def get_dashboard_groups(self, settings_file: Path | None = None) -> list[str]: Get a list of dashboard groups from the homepage settings. Args: - settings_file (Optional[Path]): Path to the settings file. If None, uses default. + settings_file: Path to the settings file. If None, uses default. Returns: - List[str]: A list of dashboard group names. + A list of dashboard group names. """ if settings_file is None: settings_file = self.settings_file @@ -52,7 +53,7 @@ def get_dashboard_groups(self, settings_file: Path | None = None) -> list[str]: return sorted(layout.keys()) -def main(): +def main() -> None: """ Run the application: read homepage settings and print dashboard groups. """ diff --git a/scripts/infra-mcp/utils/constants.py b/scripts/infra-mcp/utils/constants.py new file mode 100644 index 00000000..471a51ef --- /dev/null +++ b/scripts/infra-mcp/utils/constants.py @@ -0,0 +1,19 @@ +"""Constants for the infra-mcp server. + +This module centralizes magic numbers and configuration values used throughout the codebase. +""" + +# HTTP Request Timeouts (in seconds) +DEFAULT_REQUEST_TIMEOUT = 10 +REGISTRY_REQUEST_TIMEOUT = 30 + +# Container Tag Limits +MAX_TAGS_FETCH_LIMIT = 1000 +DEFAULT_TAG_LIMIT = 10 +DEFAULT_SAME_HASH_LIMIT = 100 + +# Container Architecture +DEFAULT_CONTAINER_ARCHITECTURE = "linux/amd64" + +# Task Execution +TASK_COMMAND_TIMEOUT = 600 # 10 minutes in seconds diff --git a/scripts/infra-mcp/utils/models.py b/scripts/infra-mcp/utils/models.py new file mode 100644 index 00000000..4225afdc --- /dev/null +++ b/scripts/infra-mcp/utils/models.py @@ -0,0 +1,27 @@ +"""Data models for the infra-mcp server. + +This module contains dataclass definitions for shared data structures used throughout the codebase. +""" + +from dataclasses import dataclass + + +@dataclass +class ContainerTagFinderArgs: + """Arguments for ContainerTagFinder operations. + + Attributes: + image: Container image name (e.g., nginx or registry.example.com/nginx) + architecture: Container architecture (e.g., linux/amd64, linux/arm64) + limit: Maximum number of tags to process + quiet: Whether to suppress output + registry: Optional registry URL for private registries + tag: Optional tag to use as reference (for same-hash operations) + """ + + image: str + architecture: str = "linux/amd64" + limit: int = 10 + quiet: bool = True + registry: str | None = None + tag: str | None = None