feature(mcp_server): POC implementation of mcp that can hook into Argus

argus/client/base.py

@@ -20,6 +20,7 @@ class ArgusClientError(Exception):
 
 class ArgusClient:
     schema_version: str | None = None
+    api_prefix: str = "/client"
 
     class Routes():
         # pylint: disable=too-few-public-methods
@@ -74,7 +75,7 @@ def get_url_for_endpoint(self, endpoint: str, location_params: dict[str, str] |
         if self.verify_location_params(endpoint, location_params):
             for param, value in location_params.items():
                 endpoint = endpoint.replace(f"${param}", str(value))
-        return f"{self._base_url}/api/{self._api_ver}/client{endpoint}"
+        return f"{self._base_url}/api/{self._api_ver}{self.api_prefix}{endpoint}"
 
     @property
     def generic_body(self) -> dict:

mcp_install_with_requirements.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+
+# Initialize the base command
+cmd="mcp install"
+
+# Read requirements.in and process each line
+while IFS= read -r line || [ -n "$line" ]; do
+    # Skip empty lines and comments
+    [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+
+    # Get the full package specification, but remove any trailing comments
+    package_spec=$(echo "$line" | sed -E 's/[[:space:]]*#.*$//' | tr -d ' ')
+
+    # Skip empty lines after comment removal
+    [[ -z "$package_spec" ]] && continue
+
+    # Only include specific packages
+    if [[ "$package_spec" =~ ^(boto3|boto3-stubs|requests|paramiko) ]]; then
+        # Append --with flag with quoted package specification
+        cmd="$cmd --with $package_spec"
+    fi
+done < requirements.in
+
+cmd="$cmd --env-var UV_PYTHON=3.10 mcp_server.py"
+
+# Execute the command
+echo "Executing: $cmd"
+exec $cmd

mcp_server.py

@@ -0,0 +1,69 @@
+from functools import lru_cache
+from uuid import UUID
+
+from mcp.server.fastmcp import FastMCP
+from argus.client.sct.client import ArgusSCTClient
+
+from sdcm.keystore import KeyStore
+
+# Create an MCP server
+mcp = FastMCP("Argus")
+
+
+@lru_cache
+def get_argus_client():
+    creds = KeyStore().get_argus_rest_credentials()
+    # TODO: add headers
+    return ArgusSCTClient(auth_token=creds["token"], base_url=creds["baseUrl"], run_id=UUID("e38b303f-df9b-4aac-b9d8-930cfd45306b"))
+
+
+@mcp.tool()
+def get_sct_run_status(test_id: str) -> str:
+    """Get the specific sct run status from argus"""
+    client = get_argus_client()
+    return client.get_status(run_id=UUID(test_id))
+
+
+@mcp.tool()
+def get_sct_run_information(test_id: str) -> dict:
+    """Get the specific sct run information from argus"""
+    client = get_argus_client()
+    results = client.get_run(run_id=UUID(test_id))
+    return dict(**results)
+
+
+@mcp.tool()
+def search_sct_run(query: str) -> list[dict]:
+    """Search for SCT (Scylla Cluster Tests) runs using Argus search functionality.
+
+    This function allows searching through SCT test runs using various query patterns.
+    The search is performed against the Argus database which stores all test results.
+
+    Args:
+        query (str): Search query string. Supports various search patterns:
+            - Simple text search: "gemini", "alternator"
+            - Version search: "release:2024.1"
+            - Group search: "group:alternator"
+            - Test Id: "e38b303f-df9b-4aac-b9d8-930cfd45306b"
+
+    Returns:
+        list: A list containing search results with test runs matching the query.
+            Each result includes test ID, name, status, and other metadata.
+
+    Examples:
+        >>> # Search for all Gemini tests
+        >>> search_sct_run("gemini")
+
+        >>> # Search for specific version tests
+        >>> search_sct_run("release:2024.1")
+
+        >>> # Search by multiple criteria
+        >>> search_sct_run("gemini release:2024.1")
+
+        >>> # Search for specific test group
+        >>> search_sct_run("group:alternator")
+    """
+    client = get_argus_client()
+    client.api_prefix = '/planning'
+    results = client.get('/search', params={'query': query}, location_params={})
+    return results.json()['response']['hits'][:20]

pyproject.toml

@@ -1,3 +1,11 @@
+[project]
+name = 'scylla-cluster-test'
+version = '0.1'
+requires-python = "==3.10.*"
+
+[tool.setuptools]
+py-modules = []
+
 [tool.ruff]
 lint.select = [
     "BLE",

requirements.in

@@ -54,3 +54,4 @@ PyGithub==2.1.1
 gimme-aws-creds==2.8.0
 confluent-kafka==2.5.3
 fastavro==1.9.7
+mcp==1.6.0