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
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 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,
        )

    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()