diff options
Diffstat (limited to 'servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py')
| -rw-r--r-- | servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py | 194 |
1 files changed, 194 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 new file mode 100644 index 0000000..38a9eb6 --- /dev/null +++ b/servers/gitlab_glab/src/mcp_server_gitlab_glab/server.py @@ -0,0 +1,194 @@ +#!/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 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]) -> tuple[bool, Any]: + """Execute a glab command and return the result. + + Args: + args: List of command arguments to pass to glab. + + 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, + ) + + 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) -> dict[str, Any]: + """Check if the glab CLI tool is available and accessible. + + Returns: + A dictionary containing availability status and version information. + """ + success, result = self.execute_glab_command(["--version"]) + + if success: + return { + "available": True, + "version": result, + } + else: + return { + "available": False, + "error": result.get("error", "Unknown error"), + } + + def find_project(self, project_name: str) -> dict[str, Any]: + """Find a GitLab project by name. + + Args: + project_name: The name of the project to search for. + + Returns: + A dictionary containing project information if found, or an error message. + """ + success, result = self.execute_glab_command( + ["api", f"/projects?search={project_name}"] + ) + + if not success: + return result + + # Check if any projects were found + if isinstance(result, list) and len(result) > 0: + # Return the first matching project + project = result[0] + return { + "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"), + } + else: + return {"error": f"Project '{project_name}' not found"} + + +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() -> dict[str, Any]: + """Check if the glab CLI tool is available and accessible. + + Returns: + A dictionary containing availability status and version information. + """ + return gitlab.check_availability() + + # Add find_project tool + @mcp.tool() + def find_project(project_name: str) -> dict[str, Any]: + """Find a GitLab project by name and return its ID. + + Args: + project_name: The name of the project to search for. + + Returns: + A dictionary containing project information if found, or an error message. + """ + return gitlab.find_project(project_name) + + 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() |