Add basic glab mcp server

4 files changed, 323 insertions, 0 deletions

diff --git a/servers/gitlab_glab/src/mcp_server_gitlab_glab/__init__.py b/servers/gitlab_glab/src/mcp_server_gitlab_glab/__init__.py
new file mode 100644
index 0000000..cd431bb
--- /dev/null
+++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/__init__.py
@@ -0,0 +1,7 @@
+"""GitLab CLI MCP Server package.
+
+This package provides an MCP server that integrates with GitLab through the GitLab CLI
+(glab).
+"""
+
+__version__ = "0.1.0"
diff --git a/servers/gitlab_glab/src/mcp_server_gitlab_glab/__main__.py b/servers/gitlab_glab/src/mcp_server_gitlab_glab/__main__.py
new file mode 100644
index 0000000..dedd9f8
--- /dev/null
+++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/__main__.py
@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+"""Command-line interface for the GitLab CLI MCP server."""
+
+from .cli import run_server
+
+if __name__ == "__main__":
+    run_server()
diff --git a/servers/gitlab_glab/src/mcp_server_gitlab_glab/cli.py b/servers/gitlab_glab/src/mcp_server_gitlab_glab/cli.py
new file mode 100644
index 0000000..69173f7
--- /dev/null
+++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/cli.py
@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""Common CLI functionality for the GitLab CLI MCP server."""
+
+import argparse
+import asyncio
+import logging
+import os
+import sys
+
+from .server import main
+
+logger = logging.getLogger("mcp_gitlab_glab_server")
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments.
+
+    Returns:
+        The parsed command-line arguments.
+    """
+    parser = argparse.ArgumentParser(description="GitLab CLI MCP Server")
+    parser.add_argument(
+        "--transport",
+        choices=["stdio", "remote"],
+        default="stdio",
+        help="Transport type (stdio or remote)",
+    )
+    parser.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="Host to bind to for remote transport (default: 127.0.0.1)",
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=8080,
+        help="Port to bind to for remote transport (default: 8080)",
+    )
+    parser.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+        help="Set the logging level (default: INFO)",
+    )
+    return parser.parse_args()
+
+
+def validate_args(args: argparse.Namespace) -> argparse.Namespace:
+    """Validate command-line arguments.
+
+    Args:
+        args: The command-line arguments to validate.
+
+    Returns:
+        The validated command-line arguments.
+    """
+    if (
+        args.transport == "remote"
+        and args.port < 1024
+        and not sys.platform.startswith("win")
+    ):
+        logger.warning(
+            "Using a port below 1024 may require root privileges on Unix-like systems."
+        )
+    return args
+
+
+def setup_logging(level: str, transport: str) -> None:
+    """Set up logging configuration.
+
+    Args:
+        level: The base logging level from command line args.
+        transport: The transport type being used (stdio or remote).
+    """
+    # Create logs directory if it doesn't exist
+    os.makedirs("logs", exist_ok=True)
+
+    # Configure file handler for all logs
+    file_handler = logging.FileHandler("logs/mcp_server.log")
+    file_handler.setLevel(getattr(logging, level))
+    file_handler.setFormatter(
+        logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+    )
+
+    # Configure console handler with appropriate level
+    console_handler = logging.StreamHandler()
+    # Use WARNING level for stdio transport to reduce interference
+    if transport == "stdio":
+        console_handler.setLevel(logging.WARNING)
+    else:
+        console_handler.setLevel(getattr(logging, level))
+    console_handler.setFormatter(
+        logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+    )
+
+    # Configure root logger
+    root_logger = logging.getLogger()
+    root_logger.setLevel(getattr(logging, level))  # Base level for all handlers
+    root_logger.handlers = []  # Remove any existing handlers
+    root_logger.addHandler(file_handler)
+    root_logger.addHandler(console_handler)
+
+
+def run_server() -> None:
+    """Run the server with CLI arguments."""
+    args = validate_args(parse_args())
+    setup_logging(args.log_level, args.transport)
+
+    try:
+        asyncio.run(main(args.transport, args.host, args.port))
+    except KeyboardInterrupt:
+        logger.info("Server stopped by user")
+    except Exception as e:
+        logger.error(f"Error running server: {e}")
+        sys.exit(1)
diff --git a/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py
new file mode 100644
index 0000000..38a9eb6
--- /dev/null
+++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py
@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""GitLab CLI MCP Server implementation.
+
+This module provides an MCP server that integrates with GitLab through the GitLab CLI
+(glab).
+"""
+
+import json
+import logging
+import os
+import subprocess
+import sys
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+# reconfigure UnicodeEncodeError prone default (i.e. windows-1252) to utf-8
+if sys.platform == "win32" and os.environ.get("PYTHONIOENCODING") is None:
+    sys.stdin.reconfigure(encoding="utf-8")
+    sys.stdout.reconfigure(encoding="utf-8")
+    sys.stderr.reconfigure(encoding="utf-8")
+
+logger = logging.getLogger("mcp_gitlab_glab_server")
+
+
+class GitLabServer:
+    """GitLab server implementation using the GitLab CLI (glab)."""
+
+    def __init__(self) -> None:
+        """Initialize the GitLab server."""
+        self.auth_message = (
+            "Authentication required. Please run 'glab auth login' to authenticate."
+        )
+
+    def execute_glab_command(self, args: list[str]) -> tuple[bool, Any]:
+        """Execute a glab command and return the result.
+
+        Args:
+            args: List of command arguments to pass to glab.
+
+        Returns:
+            A tuple containing:
+                - A boolean indicating success (True) or failure (False)
+                - The command output (if successful) or an error message (if failed)
+        """
+        try:
+            result = subprocess.run(
+                ["glab"] + args,
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+
+            if result.returncode != 0:
+                error_msg = result.stderr.strip()
+                logger.error(f"glab command failed: {error_msg}")
+                
+                # Check for authentication errors
+                if "authentication required" in error_msg.lower():
+                    return False, {"error": self.auth_message}
+                
+                return False, {"error": error_msg}
+
+            # For API commands, parse JSON output
+            if args and args[0] == "api":
+                try:
+                    return True, json.loads(result.stdout)
+                except json.JSONDecodeError:
+                    logger.error("Failed to parse JSON response")
+                    return False, {"error": "Failed to parse JSON response"}
+
+            return True, result.stdout.strip()
+        except FileNotFoundError:
+            logger.error("glab command not found")
+            return False, {
+                "error": "glab command not found. Please install GitLab CLI."
+            }
+        except Exception as e:
+            logger.error(f"Command execution failed: {str(e)}")
+            return False, {"error": f"Command execution failed: {str(e)}"}
+
+    def check_availability(self) -> dict[str, Any]:
+        """Check if the glab CLI tool is available and accessible.
+
+        Returns:
+            A dictionary containing availability status and version information.
+        """
+        success, result = self.execute_glab_command(["--version"])
+        
+        if success:
+            return {
+                "available": True,
+                "version": result,
+            }
+        else:
+            return {
+                "available": False,
+                "error": result.get("error", "Unknown error"),
+            }
+
+    def find_project(self, project_name: str) -> dict[str, Any]:
+        """Find a GitLab project by name.
+
+        Args:
+            project_name: The name of the project to search for.
+
+        Returns:
+            A dictionary containing project information if found, or an error message.
+        """
+        success, result = self.execute_glab_command(
+            ["api", f"/projects?search={project_name}"]
+        )
+        
+        if not success:
+            return result
+        
+        # Check if any projects were found
+        if isinstance(result, list) and len(result) > 0:
+            # Return the first matching project
+            project = result[0]
+            return {
+                "id": project.get("id"),
+                "name": project.get("name"),
+                "path_with_namespace": project.get("path_with_namespace"),
+                "web_url": project.get("web_url"),
+                "description": project.get("description"),
+            }
+        else:
+            return {"error": f"Project '{project_name}' not found"}
+
+
+def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
+    """Create and configure the FastMCP server.
+
+    Args:
+        host: The host to bind to for remote transport.
+        port: The port to bind to for remote transport.
+
+    Returns:
+        A configured FastMCP server instance.
+    """
+    # Create a FastMCP server with host and port settings
+    mcp = FastMCP("GitLab CLI", host=host, port=port)
+    
+    # Create a GitLabServer instance
+    gitlab = GitLabServer()
+    
+    # Add check_glab_availability tool
+    @mcp.tool()
+    def check_glab_availability() -> dict[str, Any]:
+        """Check if the glab CLI tool is available and accessible.
+
+        Returns:
+            A dictionary containing availability status and version information.
+        """
+        return gitlab.check_availability()
+    
+    # Add find_project tool
+    @mcp.tool()
+    def find_project(project_name: str) -> dict[str, Any]:
+        """Find a GitLab project by name and return its ID.
+
+        Args:
+            project_name: The name of the project to search for.
+
+        Returns:
+            A dictionary containing project information if found, or an error message.
+        """
+        return gitlab.find_project(project_name)
+    
+    return mcp
+
+
+async def main(transport_type: str, host: str, port: int) -> None:
+    """Start the server with the specified transport.
+
+    Args:
+        transport_type: The transport type to use ("stdio" or "remote").
+        host: The host to bind to for remote transport.
+        port: The port to bind to for remote transport.
+    """
+    logger.info("Starting MCP GitLab CLI Server")
+    logger.info(f"Starting GitLab CLI MCP Server with {transport_type} transport")
+    
+    # Create the server with host and port
+    mcp = create_server(host=host, port=port)
+    
+    # Run the server with the appropriate transport
+    if transport_type == "stdio":
+        logger.info("Server running with stdio transport")
+        await mcp.run_stdio_async()
+    else:  # remote transport
+        logger.info(f"Server running with remote transport on {host}:{port}")
+        await mcp.run_sse_async()