From 4e89e51901d9b3726a56d15980a0845b8e2a36b0 Mon Sep 17 00:00:00 2001 From: Dawid Rycerz Date: Tue, 15 Jul 2025 16:54:54 +0300 Subject: feat: add epics functionality --- .../src/mcp_server_gitlab_python/server.py | 245 ++++++++++++++++----- 1 file changed, 196 insertions(+), 49 deletions(-) (limited to 'servers/gitlab_python/src/mcp_server_gitlab_python/server.py') 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: -- cgit v1.2.3