summaryrefslogtreecommitdiff
path: root/servers/gitlab_python/src
diff options
context:
space:
mode:
Diffstat (limited to 'servers/gitlab_python/src')
-rw-r--r--servers/gitlab_python/src/mcp_server_gitlab_python/cli.py2
-rw-r--r--servers/gitlab_python/src/mcp_server_gitlab_python/server.py245
2 files changed, 197 insertions, 50 deletions
diff --git a/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py b/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py
index 1b4db35..323c46a 100644
--- a/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py
+++ b/servers/gitlab_python/src/mcp_server_gitlab_python/cli.py
@@ -6,8 +6,8 @@ Common CLI functionality for the GitLab Python MCP server.
import argparse
import asyncio
import logging
-import sys
import os
+import sys
from .server import main
diff --git a/servers/gitlab_python/src/mcp_server_gitlab_python/server.py b/servers/gitlab_python/src/mcp_server_gitlab_python/server.py
index 1b6ba76..05e84c5 100644
--- a/servers/gitlab_python/src/mcp_server_gitlab_python/server.py
+++ b/servers/gitlab_python/src/mcp_server_gitlab_python/server.py
@@ -1,26 +1,23 @@
import logging
import os
-import sys
-import yaml
+from typing import Any
+
import gitlab
-import subprocess
-from typing import Any, Optional
+import yaml
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger("mcp_gitlab_python_server")
-def get_git_remote_url(working_directory: str) -> Optional[str]:
+def get_git_remote_url(working_directory: str) -> str | None:
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()
+ 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) -> Optional[str]:
+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
@@ -33,28 +30,22 @@ def parse_gitlab_url_from_remote(remote_url: str) -> Optional[str]:
return f"{parts[0]}//{parts[2]}"
return None
-def get_token_from_glab_config(host: str) -> Optional[str]:
+def get_token_from_glab_config(host: str) -> str | None:
"""
Retrieve the GitLab token for a specific host from the glab-cli config file.
-
- Args:
- host (str): The GitLab host (e.g., 'gitlab.com').
-
- Returns:
- Optional[str]: The token if found, otherwise None.
"""
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:
+ 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():
+ 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:
@@ -86,18 +77,25 @@ def get_gitlab_settings(working_directory: str) -> tuple[str, str]:
# Extract host from URL
from urllib.parse import urlparse
parsed = urlparse(url)
- host = parsed.hostname or url.replace("https://", "").replace("http://", "").split("/")[0]
+ 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.")
+ 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):
+ 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()
@@ -115,7 +113,7 @@ class GitLabPythonServer:
for p in projects
]
- def search_issues(self, project: str, **filters) -> dict[str, Any]:
+ 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)
@@ -136,10 +134,20 @@ class GitLabPythonServer:
except Exception as e:
return {"error": str(e)}
- def create_issue(self, project: str, title: str, description: str, **kwargs) -> dict[str, Any]:
+ 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})
+ issue = proj.issues.create({
+ "title": title,
+ "description": description,
+ **kwargs,
+ })
return {"url": issue.web_url}
except Exception as e:
return {"error": str(e)}
@@ -149,7 +157,7 @@ class GitLabPythonServer:
project: str,
mr_iid: int,
max_size_kb: int = 100,
- filter_extensions: Optional[list[str]] = None,
+ filter_extensions: list[str] | None = None,
) -> dict[str, Any]:
import tempfile
if filter_extensions is None:
@@ -214,22 +222,19 @@ class GitLabPythonServer:
self,
project: str,
branch: str = None,
- variables: Optional[dict] = None,
+ variables: dict | None = None,
web_mode: bool = False,
working_directory: str = None,
) -> dict[str, Any]:
- import subprocess
try:
proj = self.gl.projects.get(project)
ref = branch
if not ref:
- # Try to detect current branch
+ # Try to detect current branch using GitPython
try:
- 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():
- ref = result.stdout.strip()
+ from git import Repo
+ repo = Repo(working_directory)
+ ref = repo.active_branch.name
except Exception:
ref = None
if not ref:
@@ -256,6 +261,80 @@ class GitLabPythonServer:
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 create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
mcp = FastMCP("GitLab Python", host=host, port=port)
@@ -269,16 +348,20 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
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
+ 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"]}
+ 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)
@@ -290,10 +373,10 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
title: str,
description: str,
working_directory: str,
- labels: Optional[list[str]] = None,
- assignee_ids: Optional[list[int]] = None,
- milestone_id: Optional[int] = None,
- **kwargs
+ 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)
@@ -313,7 +396,7 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
mr_iid: int,
working_directory: str,
max_size_kb: int = 100,
- filter_extensions: Optional[list[str]] = None,
+ filter_extensions: list[str] | None = None,
) -> dict[str, Any]:
"""Get the diff for a merge request."""
server = GitLabPythonServer(working_directory)
@@ -329,7 +412,7 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
project: str,
working_directory: str,
branch: str = None,
- variables: Optional[dict] = None,
+ variables: dict | None = None,
web_mode: bool = False,
) -> dict[str, Any]:
"""Run a CI/CD pipeline on GitLab."""
@@ -342,6 +425,70 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
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)
+
return mcp
async def main(transport_type: str, host: str, port: int) -> None: