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

import logging
import os
from typing import Any

import gitlab
import yaml
from mcp.server.fastmcp import FastMCP

logger = logging.getLogger("mcp_gitlab_python_server")

def get_git_remote_url(working_directory: str) -> str | None:
    try:
        from git import Repo
        repo = Repo(working_directory)
        return repo.remotes.origin.url
    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) -> str | None:
    # 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) -> str | None:
    """
    Retrieve the GitLab token for a specific host from the glab-cli config file.
    """
    config_path = os.path.expanduser("~/.config/glab-cli/config.yml")
    if not os.path.exists(config_path):
        return None
    try:
        with open(config_path) 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():  # Renamed h to _h for Ruff B007
            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) -> None:
        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: object) -> 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: object,
    ) -> 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 update_issue(
        self,
        project: str,
        issue_iid: int,
        **kwargs: object,
    ) -> dict[str, Any]:
        """Update an existing GitLab issue.

        Args:
            project (str): The project full path or ID.
            issue_iid (int): The internal ID of the issue.
            **kwargs: Fields to update (e.g., title, description, labels, etc.).

        Returns:
            dict[str, Any]: Dictionary with the issue URL or error message.
        """
        try:
            proj = self.gl.projects.get(project)
            issue = proj.issues.get(issue_iid)
            for k, v in kwargs.items():
                setattr(issue, k, v)
            issue.save()
            return {"url": issue.web_url}
        except Exception as e:
            logger.error(f"Failed to update issue {project}#{issue_iid}: {e}")
            return {"error": str(e)}

    def get_mr_diff(
        self,
        project: str,
        mr_iid: int,
        max_size_kb: int = 100,
        filter_extensions: list[str] | None = 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: dict | None = None,
        web_mode: bool = False,
        working_directory: str = None,
    ) -> dict[str, Any]:
        try:
            proj = self.gl.projects.get(project)
            ref = branch
            if not ref:
                # Try to detect current branch using GitPython
                try:
                    from git import Repo
                    repo = Repo(working_directory)
                    ref = repo.active_branch.name
                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 find_group(self, group_name: str) -> list[dict[str, Any]]:
        """Find GitLab groups by name or path."""
        groups = self.gl.groups.list(search=group_name, all=True)
        return [
            {
                "id": g.id,
                "name": g.name,
                "full_path": g.full_path,
                "web_url": g.web_url,
                "description": g.description,
            }
            for g in groups
        ]

    def search_epics(self, group: str, **filters: object) -> dict[str, Any]:
        """Search for GitLab epics in a group with various filters.

        Args:
            group (str): The group full path or ID.
            **filters: Additional filters for the search.

        Returns:
            dict[str, Any]: Dictionary with a list of epics or error message.
        """
        try:
            grp = self.gl.groups.get(group)
            epics = grp.epics.list(**filters, all=True)
            return {
                "epics": [
                    {
                        "id": e.id,
                        "iid": e.iid,
                        "title": e.title,
                        "web_url": e.web_url,
                        "state": e.state,
                        "created_at": e.created_at,
                        "updated_at": e.updated_at,
                        "author": getattr(e, "author", None),
                        "labels": getattr(e, "labels", []),
                        "description": getattr(e, "description", None),
                    }
                    for e in epics
                ]
            }
        except Exception as e:
            return {"error": str(e)}

    def create_epic(
        self,
        group: str,
        title: str,
        description: str = "",
        **kwargs: object,
    ) -> dict[str, Any]:
        """Create a new GitLab epic in a group.

        Args:
            group (str): The group full path or ID.
            title (str): The title of the epic.
            description (str): The description of the epic.
            **kwargs: Additional fields for the epic.

        Returns:
            dict[str, Any]: Dictionary with the epic URL or error message.
        """
        try:
            grp = self.gl.groups.get(group)
            data = {"title": title, "description": description}
            data.update(kwargs)
            epic = grp.epics.create(data)
            return {"url": epic.web_url}
        except Exception as e:
            return {"error": str(e)}

    def update_epic(
        self,
        group: str,
        epic_iid: int,
        **kwargs: object,
    ) -> dict[str, Any]:
        """Update an existing GitLab epic.

        Args:
            group (str): The group full path or ID.
            epic_iid (int): The internal ID of the epic.
            **kwargs: Fields to update (e.g., title, description, labels, etc.).

        Returns:
            dict[str, Any]: Dictionary with the epic URL or error message.
        """
        try:
            grp = self.gl.groups.get(group)
            epic = grp.epics.get(epic_iid)
            for k, v in kwargs.items():
                setattr(epic, k, v)
            epic.save()
            return {"url": epic.web_url}
        except Exception as e:
            logger.error(f"Failed to update epic {group}#{epic_iid}: {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: int | None = None,
        assignee_id: int | None = None,
        state: str | None = None,
        labels: list[str] | None = None,
        milestone: str | None = None,
        **kwargs: object
    ) -> 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: list[str] | None = None,
        assignee_ids: list[int] | None = None,
        milestone_id: int | None = None,
        **kwargs: object
    ) -> 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 update_issue(
        project: str,
        issue_iid: int,
        working_directory: str,
        title: str | None = None,
        description: str | None = None,
        state_event: str | None = None,
        labels: list[str] | None = None,
        assignee_ids: list[int] | None = None,
        milestone_id: int | None = None,
        **kwargs: object
    ) -> dict[str, Any]:
        """Update an existing GitLab issue."""
        server = GitLabPythonServer(working_directory)
        data = {}
        if title is not None:
            data["title"] = title
        if description is not None:
            data["description"] = description
        if state_event is not None:
            data["state_event"] = state_event
        if labels is not None:
            data["labels"] = labels
        if assignee_ids is not None:
            data["assignee_ids"] = assignee_ids
        if milestone_id is not None:
            data["milestone_id"] = milestone_id
        data.update(kwargs)
        return server.update_issue(project, issue_iid, **data)

    @mcp.tool()
    def get_mr_diff(
        project: str,
        mr_iid: int,
        working_directory: str,
        max_size_kb: int = 100,
        filter_extensions: list[str] | None = 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: dict | None = 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,
        )

    @mcp.tool()
    def find_group(group_name: str, working_directory: str) -> list[dict[str, Any]]:
        """Find GitLab groups by name or path."""
        server = GitLabPythonServer(working_directory)
        return server.find_group(group_name)

    @mcp.tool()
    def search_epics(
        group: str,
        working_directory: str,
        author_id: int | None = None,
        state: str | None = None,
        labels: list[str] | None = None,
        order_by: str | None = None,
        sort: str | None = None,
        search: str | None = None,
        **kwargs: object
    ) -> dict[str, Any]:
        """Search for GitLab epics in a group 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 ["group", "working_directory", "kwargs"]
        }
        if labels:
            filters["labels"] = ",".join(labels)
        filters.update(kwargs)
        return server.search_epics(group, **filters)

    @mcp.tool()
    def create_epic(
        group: str,
        title: str,
        description: str,
        working_directory: str,
        labels: list[str] | None = None,
        parent_id: int | None = None,
        color: str | None = None,
        confidential: bool | None = None,
        start_date_fixed: str | None = None,
        due_date_fixed: str | None = None,
        **kwargs: object
    ) -> dict[str, Any]:
        """Create a new GitLab epic in a group."""
        server = GitLabPythonServer(working_directory)
        data = {}
        if labels:
            data["labels"] = labels
        if parent_id is not None:
            data["parent_id"] = parent_id
        if color is not None:
            data["color"] = color
        if confidential is not None:
            data["confidential"] = confidential
        if start_date_fixed is not None:
            data["start_date_fixed"] = start_date_fixed
            data["start_date_is_fixed"] = True
        if due_date_fixed is not None:
            data["due_date_fixed"] = due_date_fixed
            data["due_date_is_fixed"] = True
        data.update(kwargs)
        return server.create_epic(group, title, description, **data)

    @mcp.tool()
    def update_epic(
        group: str,
        epic_iid: int,
        working_directory: str,
        title: str | None = None,
        description: str | None = None,
        state_event: str | None = None,
        labels: list[str] | None = None,
        parent_id: int | None = None,
        color: str | None = None,
        confidential: bool | None = None,
        start_date_fixed: str | None = None,
        due_date_fixed: str | None = None,
        **kwargs: object
    ) -> dict[str, Any]:
        """Update an existing GitLab epic."""
        server = GitLabPythonServer(working_directory)
        data = {}
        if title is not None:
            data["title"] = title
        if description is not None:
            data["description"] = description
        if state_event is not None:
            data["state_event"] = state_event
        if labels is not None:
            data["labels"] = labels
        if parent_id is not None:
            data["parent_id"] = parent_id
        if color is not None:
            data["color"] = color
        if confidential is not None:
            data["confidential"] = confidential
        if start_date_fixed is not None:
            data["start_date_fixed"] = start_date_fixed
            data["start_date_is_fixed"] = True
        if due_date_fixed is not None:
            data["due_date_fixed"] = due_date_fixed
            data["due_date_is_fixed"] = True
        data.update(kwargs)
        return server.update_epic(group, epic_iid, **data)

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