feat: initial gitlab_python mcp

1 files changed, 178 insertions, 0 deletions

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