path: root/servers/gitlab_python/src/mcp_server_gitlab_python/server.py

import logging
import os
import sys
import yaml
import gitlab
import subprocess
from typing import Any, Optional
from mcp.server.fastmcp import FastMCP

logger = logging.getLogger("mcp_gitlab_python_server")

def get_git_remote_url(working_directory: str) -> Optional[str]:
    try:
        result = subprocess.run([
            "git", "remote", "get-url", "origin"
        ], capture_output=True, text=True, check=False, cwd=working_directory)
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception as e:
        logger.warning(f"Could not get git remote url: {e}")
    return None

def parse_gitlab_url_from_remote(remote_url: str) -> Optional[str]:
    # Handles both SSH and HTTPS remotes
    if remote_url.startswith("git@"):
        # git@gitlab.com:namespace/project.git
        host = remote_url.split('@')[1].split(':')[0]
        return f"https://{host}"
    elif remote_url.startswith("https://"):
        # https://gitlab.com/namespace/project.git
        parts = remote_url.split('/')
        if len(parts) > 2:
            return f"{parts[0]}//{parts[2]}"
    return None

def get_token_from_glab_config(host: str) -> Optional[str]:
    """
    Retrieve the GitLab token for a specific host from the glab-cli config file.

    Args:
        host (str): The GitLab host (e.g., 'gitlab.com').

    Returns:
        Optional[str]: The token if found, otherwise None.
    """
    config_path = os.path.expanduser("~/.config/glab-cli/config.yml")
    if not os.path.exists(config_path):
        return None
    try:
        with open(config_path, "r") as f:
            config = yaml.safe_load(f)
        hosts = config.get('hosts', {})
        # Try direct match
        if host in hosts and 'token' in hosts[host]:
            return hosts[host]['token']
        # Try matching by api_host if present
        for h, data in hosts.items():
            if data.get('api_host') == host and 'token' in data:
                return data['token']
    except Exception as e:
        logger.warning(f"Could not parse glab-cli config: {e}")
    return None


def get_gitlab_settings(working_directory: str) -> tuple[str, str]:
    """
    Determine the GitLab URL and token to use, matching the token to the detected host.

    Args:
        working_directory (str): The working directory to detect the git remote.

    Returns:
        tuple[str, str]: (url, token)
    """
    # URL
    url = os.environ.get("GITLAB_HOST")
    if not url:
        remote_url = get_git_remote_url(working_directory)
        if remote_url:
            url = parse_gitlab_url_from_remote(remote_url)
    if not url:
        url = "https://gitlab.com"  # fallback default
    # Ensure scheme is present
    if not url.startswith("http://") and not url.startswith("https://"):
        url = f"https://{url}"
    # Extract host from URL
    from urllib.parse import urlparse
    parsed = urlparse(url)
    host = parsed.hostname or url.replace("https://", "").replace("http://", "").split("/")[0]
    # Token
    token = os.environ.get("GITLAB_TOKEN")
    if not token:
        token = get_token_from_glab_config(host)
    if not token:
        logger.error(f"No GitLab token found for host '{host}' in env or glab-cli config.")
        raise RuntimeError(f"No GitLab token found for host '{host}' in env or glab-cli config.")
    return url, token

class GitLabPythonServer:
    def __init__(self, working_directory: str):
        url, token = get_gitlab_settings(working_directory)
        self.gl = gitlab.Gitlab(url, private_token=token)
        self.gl.auth()

    def find_project(self, project_name: str) -> list[dict[str, Any]]:
        projects = self.gl.projects.list(search=project_name, all=True)
        return [
            {
                "id": p.id,
                "name": p.name,
                "path_with_namespace": p.path_with_namespace,
                "web_url": p.web_url,
                "description": p.description,
            }
            for p in projects
        ]

    def search_issues(self, project: str, **filters) -> dict[str, Any]:
        try:
            proj = self.gl.projects.get(project)
            issues = proj.issues.list(**filters, all=True)
            return {
                "issues": [
                    {
                        "id": i.id,
                        "iid": i.iid,
                        "title": i.title,
                        "web_url": i.web_url,
                        "state": i.state,
                        "created_at": i.created_at,
                        "updated_at": i.updated_at,
                    }
                    for i in issues
                ]
            }
        except Exception as e:
            return {"error": str(e)}

    def create_issue(self, project: str, title: str, description: str, **kwargs) -> dict[str, Any]:
        try:
            proj = self.gl.projects.get(project)
            issue = proj.issues.create({"title": title, "description": description, **kwargs})
            return {"url": issue.web_url}
        except Exception as e:
            return {"error": str(e)}

    def get_mr_diff(
        self,
        project: str,
        mr_iid: int,
        max_size_kb: int = 100,
        filter_extensions: Optional[list[str]] = None,
    ) -> dict[str, Any]:
        import tempfile
        if filter_extensions is None:
            filter_extensions = [".lock", ".log"]
        try:
            proj = self.gl.projects.get(project)
            mr = proj.mergerequests.get(mr_iid)
            # Get the diff as a list of dicts (one per file)
            diffs = mr.diffs.list(get_all=True)
            # Build unified diff string
            diff_content = ""
            for diff in diffs:
                old_path = diff.old_path
                new_path = diff.new_path
                # Filter by extension
                if any(
                    old_path.endswith(ext) or new_path.endswith(ext)
                    for ext in filter_extensions
                ):
                    continue
                diff_content += f"diff --git a/{old_path} b/{new_path}\n"
                diff_content += diff.diff + "\n"
            diff_size_kb = len(diff_content.encode("utf-8")) / 1024
            if diff_size_kb > max_size_kb:
                try:
                    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:
                    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
            }
        except Exception as e:
            return {"error": str(e)}

    def run_ci_pipeline(
        self,
        project: str,
        branch: str = None,
        variables: Optional[dict] = None,
        web_mode: bool = False,
        working_directory: str = None,
    ) -> dict[str, Any]:
        import subprocess
        try:
            proj = self.gl.projects.get(project)
            ref = branch
            if not ref:
                # Try to detect current branch
                try:
                    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():
                        ref = result.stdout.strip()
                except Exception:
                    ref = None
            if not ref:
                # If still no branch, let GitLab use default
                ref = proj.default_branch
            # Prepare variables
            pipeline_vars = variables.copy() if variables else {}
            if web_mode:
                pipeline_vars["CI_PIPELINE_SOURCE"] = "web"
            pipeline = proj.pipelines.create({
                "ref": ref,
                "variables": [
                    {"key": k, "value": v} for k, v in pipeline_vars.items()
                ] if pipeline_vars else None
            })
            info = {
                "success": True,
                "pipeline_id": pipeline.id,
                "pipeline_url": pipeline.web_url,
                "branch": ref,
                "web_mode": web_mode,
            }
            return info
        except Exception as e:
            return {"error": str(e)}

def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
    mcp = FastMCP("GitLab Python", host=host, port=port)

    @mcp.tool()
    def find_project(project_name: str, working_directory: str) -> list[dict[str, Any]]:
        """Find GitLab projects by name."""
        server = GitLabPythonServer(working_directory)
        return server.find_project(project_name)

    @mcp.tool()
    def search_issues(
        project: str,
        working_directory: str,
        author_id: Optional[int] = None,
        assignee_id: Optional[int] = None,
        state: Optional[str] = None,
        labels: Optional[list[str]] = None,
        milestone: Optional[str] = None,
        **kwargs
    ) -> dict[str, Any]:
        """Search for GitLab issues with various filters."""
        server = GitLabPythonServer(working_directory)
        filters = {k: v for k, v in locals().items() if v is not None and k not in ["project", "working_directory", "kwargs"]}
        if labels:
            filters["labels"] = ",".join(labels)
        filters.update(kwargs)
        return server.search_issues(project, **filters)

    @mcp.tool()
    def create_issue(
        project: str,
        title: str,
        description: str,
        working_directory: str,
        labels: Optional[list[str]] = None,
        assignee_ids: Optional[list[int]] = None,
        milestone_id: Optional[int] = None,
        **kwargs
    ) -> dict[str, Any]:
        """Create a new GitLab issue."""
        server = GitLabPythonServer(working_directory)
        data = {}
        if labels:
            data["labels"] = labels
        if assignee_ids:
            data["assignee_ids"] = assignee_ids
        if milestone_id:
            data["milestone_id"] = milestone_id
        data.update(kwargs)
        return server.create_issue(project, title, description, **data)

    @mcp.tool()
    def get_mr_diff(
        project: str,
        mr_iid: int,
        working_directory: str,
        max_size_kb: int = 100,
        filter_extensions: Optional[list[str]] = None,
    ) -> dict[str, Any]:
        """Get the diff for a merge request."""
        server = GitLabPythonServer(working_directory)
        return server.get_mr_diff(
            project=project,
            mr_iid=mr_iid,
            max_size_kb=max_size_kb,
            filter_extensions=filter_extensions,
        )

    @mcp.tool()
    def run_ci_pipeline(
        project: str,
        working_directory: str,
        branch: str = None,
        variables: Optional[dict] = None,
        web_mode: bool = False,
    ) -> dict[str, Any]:
        """Run a CI/CD pipeline on GitLab."""
        server = GitLabPythonServer(working_directory)
        return server.run_ci_pipeline(
            project=project,
            branch=branch,
            variables=variables,
            web_mode=web_mode,
            working_directory=working_directory,
        )

    return mcp

async def main(transport_type: str, host: str, port: int) -> None:
    logger.info("Starting MCP GitLab Python Server")
    mcp = create_server(host=host, port=port)
    if transport_type == "stdio":
        logger.info("Server running with stdio transport")
        await mcp.run_stdio_async()
    else:
        logger.info(f"Server running with remote transport on {host}:{port}")
        await mcp.run_sse_async()