feat: initial gitlab_python mcp

4 files changed, 272 insertions, 0 deletions

diff --git a/servers/gitlab_python/src/mcp_server_gitlab_python/__init__.py b/servers/gitlab_python/src/mcp_server_gitlab_python/__init__.py
new file mode 100644
index 0000000..5ee1366
--- /dev/null
+++ b/servers/gitlab_python/src/mcp_server_gitlab_python/__init__.py
@@ -0,0 +1,6 @@
+"""GitLab Python MCP Server package.
+
+This package provides an MCP server that integrates with GitLab using python-gitlab.
+"""
+
+__version__ = "0.1.0" 
\ No newline at end of file
diff --git a/servers/gitlab_python/src/mcp_server_gitlab_python/__main__.py b/servers/gitlab_python/src/mcp_server_gitlab_python/__main__.py
new file mode 100644
index 0000000..8f98d0b
--- /dev/null
+++ b/servers/gitlab_python/src/mcp_server_gitlab_python/__main__.py
@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+"""Command-line interface for the GitLab Python MCP server."""
+
+from .cli import run_server
+
+if __name__ == "__main__":
+    run_server() 
\ No newline at end of file
diff --git a/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py b/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py
new file mode 100644
index 0000000..1b4db35
--- /dev/null
+++ b/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""
+Common CLI functionality for the GitLab Python MCP server.
+"""
+
+import argparse
+import asyncio
+import logging
+import sys
+import os
+
+from .server import main
+
+logger = logging.getLogger("mcp_gitlab_python_server")
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="GitLab Python MCP Server")
+    parser.add_argument(
+        "--transport",
+        choices=["stdio", "remote"],
+        default="stdio",
+        help="Transport type (stdio or remote)",
+    )
+    parser.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="Host to bind to for remote transport (default: 127.0.0.1)",
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=8080,
+        help="Port to bind to for remote transport (default: 8080)",
+    )
+    parser.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+        help="Set the logging level (default: INFO)",
+    )
+    return parser.parse_args()
+
+def validate_args(args: argparse.Namespace) -> argparse.Namespace:
+    if (args.transport == "remote" and args.port < 1024 
+            and not sys.platform.startswith("win")):
+        logger.warning(
+            "Using a port below 1024 may require root privileges on Unix-like systems."
+        )
+    return args
+
+def setup_logging(level: str, transport: str) -> None:
+    os.makedirs("logs", exist_ok=True)
+    file_handler = logging.FileHandler("logs/mcp_server.log")
+    file_handler.setLevel(getattr(logging, level))
+    file_handler.setFormatter(logging.Formatter(
+        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    ))
+    console_handler = logging.StreamHandler()
+    if transport == "stdio":
+        console_handler.setLevel(logging.WARNING)
+    else:
+        console_handler.setLevel(getattr(logging, level))
+    console_handler.setFormatter(logging.Formatter(
+        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    ))
+    root_logger = logging.getLogger()
+    root_logger.setLevel(getattr(logging, level))
+    root_logger.handlers = []
+    root_logger.addHandler(file_handler)
+    root_logger.addHandler(console_handler)
+
+def run_server() -> None:
+    args = validate_args(parse_args())
+    setup_logging(args.log_level, args.transport)
+    try:
+        asyncio.run(main(args.transport, args.host, args.port))
+    except KeyboardInterrupt:
+        logger.info("Server stopped by user")
+    except Exception as e:
+        logger.error(f"Error running server: {e}")
+        sys.exit(1) 
\ No newline at end of file
diff --git a/servers/gitlab_python/src/mcp_server_gitlab_python/server.py b/servers/gitlab_python/src/mcp_server_gitlab_python/server.py
new file mode 100644
index 0000000..3005d30
--- /dev/null
+++ b/servers/gitlab_python/src/mcp_server_gitlab_python/server.py
@@ -0,0 +1,178 @@
+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() -> Optional[str]:
+    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)
+        # Try to find a token in the config (glab stores tokens per host)
+        hosts = config.get('hosts', {})
+        for host, data in hosts.items():
+            if '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]:
+    # URL
+    url = os.environ.get("GITLAB_URL")
+    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
+    # Token
+    token = os.environ.get("GITLAB_TOKEN")
+    if not token:
+        token = get_token_from_glab_config()
+    if not token:
+        raise RuntimeError("No GitLab token found 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 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)
+
+    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() 
\ No newline at end of file