diff options
Diffstat (limited to 'servers/gitlab_glab/src/mcp_server_gitlab_glab')
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() |