diff options
Diffstat (limited to 'servers/gitlab_glab/src')
| -rw-r--r-- | servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py | 340 |
1 files changed, 340 insertions, 0 deletions
diff --git a/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py index d3fd074..f239246 100644 --- a/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py +++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py @@ -11,6 +11,7 @@ import os import re import subprocess import sys +import tempfile from typing import Any from mcp.server.fastmcp import FastMCP @@ -309,7 +310,273 @@ class GitLabServer: 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. @@ -455,6 +722,79 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP: 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 |