path: root/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py

#!/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 re
import subprocess
import sys
import tempfile
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], working_directory: str
    ) -> tuple[bool, Any]:
        """Execute a glab command and return the result.

        Args:
            args: List of command arguments to pass to glab.
            working_directory: The directory to execute the command in.

        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,
                cwd=working_directory,
            )

            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, working_directory: str) -> dict[str, Any]:
        """Check if the glab CLI tool is available and accessible.

        Args:
            working_directory: The directory to execute the command in.

        Returns:
            A dictionary containing availability status and version information.
        """
        success, result = self.execute_glab_command(["--version"], working_directory)

        if success:
            return {
                "available": True,
                "version": result,
            }
        else:
            return {
                "available": False,
                "error": result.get("error", "Unknown error"),
            }

    def find_project(
        self, project_name: str, working_directory: str
    ) -> dict[str, Any] | list[dict[str, Any]]:
        """Find GitLab projects by name.

        Args:
            project_name: The name of the project to search for.
            working_directory: The directory to execute the command in.

        Returns:
            A list of dictionaries containing project information if found,
            or an error message.
        """
        success, result = self.execute_glab_command(
            [
                "api",
                f"/projects?search_namespaces=true&search={project_name}",
            ],
            working_directory
        )

        if not success:
            return result

        # Check if any projects were found
        if isinstance(result, list) and len(result) > 0:
            # Return all matching projects
            projects = []
            for project in result:
                projects.append(
                    {
                        "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"),
                    }
                )
            return projects
        else:
            return {"error": f"Project '{project_name}' not found"}

    def search_issues(
        self,
        working_directory: str,
        author: str | None = None,
        assignee: str | None = None,
        closed: bool = False,
        confidential: bool = False,
        group: str | None = None,
        issue_type: str | None = None,
        iteration: int | None = None,
        label: list[str] | None = None,
        milestone: str | None = None,
        not_assignee: str | None = None,
        not_author: str | None = None,
        not_label: list[str] | None = None,
        page: int = 1,
        per_page: int = 30,
        project: str | None = None,
    ) -> dict[str, Any]:
        """Search for GitLab issues with various filters.
        
        Args:
            working_directory: The directory to execute the command in.
            author: Filter issue by author username.
            assignee: Filter issue by assignee username.
            closed: Get only closed issues.
            confidential: Filter by confidential issues.
            group: Select a group or subgroup.
            issue_type: Filter issue by its type (issue, incident, test_case).
            iteration: Filter issue by iteration ID.
            label: Filter issue by label names.
            milestone: Filter issue by milestone ID or title.
            not_assignee: Filter issue by not being assigned to username.
            not_author: Filter issue by not being by author username.
            not_label: Filter issue by lack of label names.
            page: Page number.
            per_page: Number of items to list per page.
            project: Select another repository.
            
        Returns:
            A dictionary containing a list of issues with their IDs, titles, and links.
        """
        # Build command arguments
        args = ["issue", "list", "-O", "json"]
        
        # Add filter arguments
        if author:
            args.extend(["--author", author])
        if assignee:
            args.extend(["-a", assignee])
        if closed:
            args.append("-c")
        if confidential:
            args.append("-C")
        if group:
            args.extend(["-g", group])
        if issue_type:
            args.extend(["-t", issue_type])
        if iteration:
            args.extend(["-i", str(iteration)])
        if label:
            for lbl in label:
                args.extend(["-l", lbl])
        if milestone:
            args.extend(["-m", milestone])
        if not_assignee:
            args.extend(["--not-assignee", not_assignee])
        if not_author:
            args.extend(["--not-author", not_author])
        if not_label:
            for lbl in not_label:
                args.extend(["--not-label", lbl])
        if page != 1:
            args.extend(["-p", str(page)])
        if per_page != 30:
            args.extend(["-P", str(per_page)])
        if project:
            args.extend(["-R", project])
        
        # Execute the command
        success, result = self.execute_glab_command(args, working_directory)
        
        if not success:
            return result
        
        # If the result is already a list (JSON parsed), process it
        if isinstance(result, list):
            issues = []
            for issue in result:
                issues.append({
                    "id": issue.get("id"),
                    "iid": issue.get("iid"),
                    "title": issue.get("title"),
                    "web_url": issue.get("web_url"),
                    "state": issue.get("state"),
                    "created_at": issue.get("created_at"),
                    "updated_at": issue.get("updated_at"),
                })
            return {"issues": issues}
        else:
            # This shouldn't happen with -O json, but handle it just in case
            return {"error": "Failed to parse issues list"}

    def create_issue(
        self,
        title: str,
        description: str,
        working_directory: str,
        labels: list[str] | None = None,
        assignee: list[str] | None = None,
        milestone: str | None = None,
        epic_id: int | None = None,
        project: str | None = None,
    ) -> dict[str, Any]:
        """Create a new GitLab issue.

        Args:
            title: The issue title.
            description: The issue description.
            labels: Optional list of labels to apply to the issue.
            assignee: Optional list of usernames to assign the issue to.
            milestone: Optional milestone title or ID.
            epic_id: Optional ID of the epic to add the issue to.
            project: Optional project name or path.
            working_directory: The directory to execute the command in.

        Returns:
            A dictionary containing the issue URL or an error message.
        """
        # Build command arguments
        args = ["issue", "create", "-y", "-t", title, "-d", description]

        # Add optional arguments if provided
        if labels:
            args.extend(["-l", ",".join(labels)])
        if assignee:
            for user in assignee:
                args.extend(["-a", user])
        if milestone:
            args.extend(["-m", milestone])
        if epic_id:
            args.extend(["--epic", str(epic_id)])
        if project:
            args.extend(["-R", project])

        # Execute the command
        success, result = self.execute_glab_command(args, working_directory)

        if not success:
            return result

        # Parse the output to extract the issue URL
        url_pattern = re.compile(r'https?://[^\s]+/issues/\d+')
        match = url_pattern.search(result)
        if match:
            return {"url": match.group(0)}
        else:
            # Log the full output for debugging
            logger.error(f"Failed to extract issue URL from output: {result}")
            return {"error": "Failed to extract issue URL from command output"}

    def filter_diff_content(
        self,
        diff_content: str,
        exclude_extensions: list[str],
    ) -> str:
        """Filter out files with specified extensions from diff content.

        Args:
            diff_content: The original diff content.
            exclude_extensions: List of file extensions to exclude 
                (e.g., [".lock", ".log"]).

        Returns:
            Filtered diff content with excluded files removed.
        """
        if not exclude_extensions:
            return diff_content

        lines = diff_content.split('\n')
        filtered_lines = []
        skip_block = False

        for line in lines:
            if line.startswith("diff --git a/"):
                parts = line.split()
                if len(parts) >= 4:
                    file_a = parts[2]
                    file_b = parts[3]
                    # Check if either file should be excluded
                    should_exclude = False
                    for ext in exclude_extensions:
                        # Handle both exact extension matches and lock file patterns
                        if ext == ".lock":
                            # Special handling for lock files 
                            # (package-lock.json, yarn.lock, etc.)
                            if (file_a.endswith('.lock') or 
                                file_b.endswith('.lock') or
                                'lock.' in file_a or 'lock.' in file_b or
                                file_a.endswith('-lock.json') or 
                                file_b.endswith('-lock.json')):
                                should_exclude = True
                                break
                        else:
                            # Standard extension matching
                            if file_a.endswith(ext) or file_b.endswith(ext):
                                should_exclude = True
                                break
                    
                    skip_block = should_exclude
                else:
                    skip_block = False

            if not skip_block:
                filtered_lines.append(line)

        return '\n'.join(filtered_lines)

    def get_mr_diff(
        self,
        working_directory: str,
        mr_id: str | None = None,
        color: str = "never",
        raw: bool = False,
        repo: str | None = None,
        max_size_kb: int = 100,
        filter_extensions: list[str] | None = None,
    ) -> dict[str, Any]:
        """Get the diff for a merge request.

        Args:
            working_directory: The directory to execute the command in.
            mr_id: The merge request ID or branch name. If None, uses current branch.
            color: Use color in diff output: always, never, auto (default: never).
            raw: Use raw diff format that can be piped to commands.
            repo: Select another repository (OWNER/REPO or GROUP/NAMESPACE/REPO format).
            max_size_kb: Maximum size in KB before saving to temporary file 
                (default: 100).
            filter_extensions: List of file extensions to exclude from diff 
                (default: [".lock", ".log"]).

        Returns:
            A dictionary containing the diff content or a path to temporary file 
            if too large.
        """
        # Set default filter extensions if not provided
        if filter_extensions is None:
            filter_extensions = [".lock", ".log"]
        
        # Build command arguments
        args = ["mr", "diff"]
        
        # Add MR ID or branch if specified
        if mr_id:
            args.append(mr_id)
        
        # Add color option
        if color in ["always", "never", "auto"]:
            args.extend(["--color", color])
        
        # Add raw option
        if raw:
            args.append("--raw")
        
        # Add repo option
        if repo:
            args.extend(["-R", repo])
        
        # Execute the command
        success, result = self.execute_glab_command(args, working_directory)
        
        if not success:
            return result
        
        # Apply filtering to remove unwanted file extensions
        diff_content = self.filter_diff_content(result, filter_extensions)
        
        # Check if the diff is too large
        diff_size_kb = len(diff_content.encode('utf-8')) / 1024
        
        if diff_size_kb > max_size_kb:
            try:
                # Create a temporary file to store the large diff
                with tempfile.NamedTemporaryFile(
                    mode='w',
                    suffix='.diff',
                    prefix='mr_diff_',
                    delete=False,
                    encoding='utf-8'
                ) as temp_file:
                    temp_file.write(diff_content)
                    temp_path = temp_file.name
                
                return {
                    "diff_too_large": True,
                    "size_kb": round(diff_size_kb, 2),
                    "max_size_kb": max_size_kb,
                    "temp_file_path": temp_path,
                    "message": (
                        f"Diff is too large ({diff_size_kb:.2f} KB > "
                        f"{max_size_kb} KB). Content saved to temporary file: "
                        f"{temp_path}"
                    )
                }
            except Exception as e:
                logger.error(f"Failed to create temporary file: {str(e)}")
                return {
                    "error": (
                        f"Diff is too large ({diff_size_kb:.2f} KB) and failed to "
                        f"create temporary file: {str(e)}"
                    )
                }
        
        return {
            "diff": diff_content,
            "size_kb": round(diff_size_kb, 2),
            "temp_file_path": None
        }

    def run_ci_pipeline(
        self,
        working_directory: str,
        branch: str | None = None,
        variables: list[str] | None = None,
        variables_env: list[str] | None = None,
        variables_file: list[str] | None = None,
        variables_from: str | None = None,
        web_mode: bool = False,
        repo: str | None = None,
    ) -> dict[str, Any]:
        """Run a CI/CD pipeline on GitLab.

        Args:
            working_directory: The directory to execute the command in.
            branch: Create pipeline on branch/ref. If None, uses current branch.
            variables: Pass variables to pipeline in format key:value.
            variables_env: Pass environment variables to pipeline in format key:value.
            variables_file: Pass file contents as file variables in format key:filename.
            variables_from: JSON file containing variables for pipeline execution.
            web_mode: Run pipeline in web mode (overrides CI_PIPELINE_SOURCE to 'web').
            repo: Select another repository (OWNER/REPO or GROUP/NAMESPACE/REPO format).

        Returns:
            A dictionary containing the pipeline information or an error message.
        """
        # Build command arguments
        args = ["ci", "run"]
        
        # Handle branch - if not provided, get current branch
        if branch is None:
            try:
                # Get current branch using git command
                result = subprocess.run(
                    ["git", "branch", "--show-current"],
                    capture_output=True,
                    text=True,
                    check=False,
                    cwd=working_directory,
                )
                if result.returncode == 0 and result.stdout.strip():
                    branch = result.stdout.strip()
                    logger.info(f"Using current branch: {branch}")
                else:
                    logger.warning(
                        "Could not determine current branch, proceeding without -b flag"
                    )
            except Exception as e:
                logger.warning(f"Could not determine current branch: {str(e)}")
        
        # Add branch if available
        if branch:
            args.extend(["-b", branch])
        
        # Add variables
        if variables:
            for var in variables:
                args.extend(["--variables", var])
        
        # Add environment variables
        if variables_env:
            for var in variables_env:
                args.extend(["--variables-env", var])
        
        # Add file variables
        if variables_file:
            for var in variables_file:
                args.extend(["--variables-file", var])
        
        # Add variables from file
        if variables_from:
            args.extend(["-f", variables_from])
        
        # Add web mode - override CI_PIPELINE_SOURCE to 'web'
        if web_mode:
            args.extend(["--variables-env", "CI_PIPELINE_SOURCE:web"])
        
        # Add repo
        if repo:
            args.extend(["-R", repo])
        
        # Execute the command
        success, result = self.execute_glab_command(args, working_directory)
        
        if not success:
            return result
        
        # Parse the output to extract pipeline information
        # The output typically contains pipeline URL and ID
        output_lines = result.strip().split('\n') if isinstance(result, str) else []
        
        pipeline_info = {
            "success": True,
            "output": result,
            "branch": branch,
            "web_mode": web_mode
        }
        
        # Try to extract pipeline URL if present
        for line in output_lines:
            if "https://" in line and "/pipelines/" in line:
                # Extract just the URL from the line
                import re
                url_match = re.search(r'https://[^\s]+/pipelines/\d+', line)
                if url_match:
                    pipeline_info["pipeline_url"] = url_match.group(0)
                break
        
        return pipeline_info
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(working_directory: str) -> dict[str, Any]:
        """Check if the glab CLI tool is available and accessible.

        Args:
            working_directory: The directory to execute the command in.

        Returns:
            A dictionary containing availability status and version information.
        """
        return gitlab.check_availability(working_directory)

    # Add find_project tool
    @mcp.tool()
    def find_project(
        project_name: str, working_directory: str
    ) -> dict[str, Any] | list[dict[str, Any]]:
        """Find GitLab projects by name.

        Args:
            project_name: The name of the project to search for.
            working_directory: The directory to execute the command in.

        Returns:
            A list of dictionaries containing project information if found,
            or an error message.
        """
        return gitlab.find_project(project_name, working_directory)

    # Add create_issue tool
    @mcp.tool()
    def search_issues(
        working_directory: str,
        author: str | None = None,
        assignee: str | None = None,
        closed: bool = False,
        confidential: bool = False,
        group: str | None = None,
        issue_type: str | None = None,
        iteration: int | None = None,
        label: list[str] | None = None,
        milestone: str | None = None,
        not_assignee: str | None = None,
        not_author: str | None = None,
        not_label: list[str] | None = None,
        page: int = 1,
        per_page: int = 30,
        project: str | None = None,
    ) -> dict[str, Any]:
        """Search for GitLab issues with various filters.
        
        Args:
            working_directory: The directory to execute the command in.
            author: Filter issue by author username.
            assignee: Filter issue by assignee username.
            closed: Get only closed issues.
            confidential: Filter by confidential issues.
            group: Select a group or subgroup.
            issue_type: Filter issue by its type (issue, incident, test_case).
            iteration: Filter issue by iteration ID.
            label: Filter issue by label names.
            milestone: Filter issue by milestone ID or title.
            not_assignee: Filter issue by not being assigned to username.
            not_author: Filter issue by not being by author username.
            not_label: Filter issue by lack of label names.
            page: Page number.
            per_page: Number of items to list per page.
            project: Select another repository.
            
        Returns:
            A dictionary containing a list of issues with their IDs, titles, and links.
        """
        return gitlab.search_issues(
            working_directory=working_directory,
            author=author,
            assignee=assignee,
            closed=closed,
            confidential=confidential,
            group=group,
            issue_type=issue_type,
            iteration=iteration,
            label=label,
            milestone=milestone,
            not_assignee=not_assignee,
            not_author=not_author,
            not_label=not_label,
            page=page,
            per_page=per_page,
            project=project,
        )

    @mcp.tool()
    def create_issue(
        title: str,
        description: str,
        working_directory: str,
        labels: list[str] | None = None,
        assignee: list[str] | None = None,
        milestone: str | None = None,
        epic_id: int | None = None,
        project: str | None = None,
    ) -> dict[str, Any]:
        """Create a new GitLab issue.

        Args:
            title: The issue title.
            description: The issue description.
            labels: Optional list of labels to apply to the issue.
            assignee: Optional list of usernames to assign the issue to.
            milestone: Optional milestone title or ID.
            epic_id: Optional ID of the epic to add the issue to.
            project: Optional project name or path.
            working_directory: The directory to execute the command in.

        Returns:
            A dictionary containing the issue URL or an error message.
        """
        return gitlab.create_issue(
            title=title,
            description=description,
            working_directory=working_directory,
            labels=labels,
            assignee=assignee,
            milestone=milestone,
            epic_id=epic_id,
            project=project,
        )

    @mcp.tool()
    def get_mr_diff(
        working_directory: str,
        mr_id: str | None = None,
        color: str = "never",
        raw: bool = False,
        repo: str | None = None,
        max_size_kb: int = 100,
        filter_extensions: list[str] | None = None,
    ) -> dict[str, Any]:
        """Get the diff for a merge request.

        Args:
            working_directory: The directory to execute the command in.
            mr_id: The merge request ID or branch name. If None, uses current branch.
            color: Use color in diff output: always, never, auto (default: never).
            raw: Use raw diff format that can be piped to commands.
            repo: Select another repository (OWNER/REPO or GROUP/NAMESPACE/REPO format).
            max_size_kb: Maximum size in KB before saving to temporary file 
                (default: 100).
            filter_extensions: List of file extensions to exclude from diff 
                (default: [".lock", ".log"]).

        Returns:
            A dictionary containing the diff content or a path to temporary file 
            if too large.
        """
        return gitlab.get_mr_diff(
            working_directory=working_directory,
            mr_id=mr_id,
            color=color,
            raw=raw,
            repo=repo,
            max_size_kb=max_size_kb,
            filter_extensions=filter_extensions,
        )

    @mcp.tool()
    def run_ci_pipeline(
        working_directory: str,
        branch: str | None = None,
        variables: list[str] | None = None,
        variables_env: list[str] | None = None,
        variables_file: list[str] | None = None,
        variables_from: str | None = None,
        web_mode: bool = False,
        repo: str | None = None,
    ) -> dict[str, Any]:
        """Run a CI/CD pipeline on GitLab.

        Args:
            working_directory: The directory to execute the command in.
            branch: Create pipeline on branch/ref. If None, uses current branch.
            variables: Pass variables to pipeline in format key:value.
            variables_env: Pass environment variables to pipeline in format key:value.
            variables_file: Pass file contents as file variables in format key:filename.
            variables_from: JSON file containing variables for pipeline execution.
            web_mode: Run pipeline in web mode (overrides CI_PIPELINE_SOURCE to 'web').
            repo: Select another repository (OWNER/REPO or GROUP/NAMESPACE/REPO format).

        Returns:
            A dictionary containing the pipeline information or an error message.
        """
        return gitlab.run_ci_pipeline(
            working_directory=working_directory,
            branch=branch,
            variables=variables,
            variables_env=variables_env,
            variables_file=variables_file,
            variables_from=variables_from,
            web_mode=web_mode,
            repo=repo,
        )
    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()