Jira-GitLab MCP Server

import asyncio import logging from typing import Dict, List, Any, Optional import gitlab from gitlab.exceptions import GitlabError, GitlabAuthenticationError, GitlabGetError, GitlabCreateError from requests.exceptions import HTTPError, ConnectionError, Timeout from utils.error_handler import GitLabError, AuthenticationError, retry_on_failure logger = logging.getLogger(__name__) class GitLabClient: def __init__(self, config: Dict[str, str]): self.config = config self.gl = gitlab.Gitlab(config["base_url"], private_token=config["access_token"]) logger.info(f"GitLab client initialized for {config['base_url']}") @retry_on_failure(max_retries=3, delay=1.0) async def test_connection(self) -> bool: """Test the GitLab connection""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._test_connection_sync) logger.info("GitLab connection test successful") return result except Exception as e: logger.error(f"GitLab connection test failed: {e}") raise AuthenticationError(f"Failed to connect to GitLab: {str(e)}", "gitlab") def _test_connection_sync(self) -> bool: """Synchronous connection test""" try: # Test authentication by getting current user self.gl.auth() user = self.gl.user return user is not None except GitlabAuthenticationError: raise AuthenticationError("Invalid GitLab access token", "gitlab") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except ConnectionError: raise GitLabError("Failed to connect to GitLab server") except Exception as e: raise GitLabError(f"Unexpected error: {str(e)}") @retry_on_failure(max_retries=2, delay=0.5) async def create_branch(self, project_id: int, branch_name: str, ref: str = "main") -> str: """Create a new branch in GitLab project""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._create_branch_sync, project_id, branch_name, ref) logger.info(f"Created branch '{branch_name}' in project {project_id}") return result except Exception as e: logger.error(f"Failed to create branch '{branch_name}' in project {project_id}: {e}") raise GitLabError(f"Failed to create branch: {str(e)}") def _create_branch_sync(self, project_id: int, branch_name: str, ref: str) -> str: """Synchronous branch creation""" try: project = self.gl.projects.get(project_id) # Check if branch already exists try: existing_branch = project.branches.get(branch_name) if existing_branch: logger.warning(f"Branch '{branch_name}' already exists in project {project_id}") return f"{project.web_url}/-/tree/{branch_name}" except GitlabGetError: # Branch doesn't exist, which is what we want pass # Create the branch branch_data = { 'branch': branch_name, 'ref': ref } branch = project.branches.create(branch_data) branch_url = f"{project.web_url}/-/tree/{branch_name}" return branch_url except GitlabGetError as e: if "404" in str(e): raise GitLabError(f"Project {project_id} not found") else: raise GitLabError(f"Failed to access project: {str(e)}") except GitlabCreateError as e: if "already exists" in str(e).lower(): # Branch already exists, return its URL project = self.gl.projects.get(project_id) return f"{project.web_url}/-/tree/{branch_name}" else: raise GitLabError(f"Failed to create branch: {str(e)}") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error creating branch: {str(e)}") async def get_projects(self, owned: bool = True) -> List[Dict[str, Any]]: """Get GitLab projects""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._get_projects_sync, owned) logger.info(f"Retrieved {len(result)} projects from GitLab") return result except Exception as e: logger.error(f"Failed to get GitLab projects: {e}") raise GitLabError(f"Failed to fetch projects: {str(e)}") def _get_projects_sync(self, owned: bool) -> List[Dict[str, Any]]: """Synchronous project retrieval""" try: if owned: projects = self.gl.projects.list(owned=True, all=True) else: projects = self.gl.projects.list(membership=True, all=True) project_list = [] for project in projects: project_data = { "id": project.id, "name": project.name, "path": project.path, "web_url": project.web_url, "default_branch": getattr(project, 'default_branch', 'main'), "description": getattr(project, 'description', ''), "visibility": getattr(project, 'visibility', 'private') } project_list.append(project_data) return project_list except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error fetching projects: {str(e)}") async def get_project(self, project_id: int) -> Optional[Dict[str, Any]]: """Get a specific GitLab project""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._get_project_sync, project_id) logger.info(f"Retrieved project {project_id}") return result except Exception as e: logger.error(f"Failed to get project {project_id}: {e}") raise GitLabError(f"Failed to fetch project {project_id}: {str(e)}") def _get_project_sync(self, project_id: int) -> Optional[Dict[str, Any]]: """Synchronous single project retrieval""" try: project = self.gl.projects.get(project_id) if not project: return None return { "id": project.id, "name": project.name, "path": project.path, "web_url": project.web_url, "default_branch": getattr(project, 'default_branch', 'main'), "description": getattr(project, 'description', ''), "visibility": getattr(project, 'visibility', 'private'), "created_at": getattr(project, 'created_at', ''), "last_activity_at": getattr(project, 'last_activity_at', '') } except GitlabGetError as e: if "404" in str(e): raise GitLabError(f"Project {project_id} not found") else: raise GitLabError(f"Failed to access project: {str(e)}") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error fetching project: {str(e)}") async def get_branches(self, project_id: int) -> List[Dict[str, Any]]: """Get branches for a GitLab project""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._get_branches_sync, project_id) logger.info(f"Retrieved {len(result)} branches for project {project_id}") return result except Exception as e: logger.error(f"Failed to get branches for project {project_id}: {e}") raise GitLabError(f"Failed to fetch branches: {str(e)}") def _get_branches_sync(self, project_id: int) -> List[Dict[str, Any]]: """Synchronous branch retrieval""" try: project = self.gl.projects.get(project_id) branches = project.branches.list(all=True) branch_list = [] for branch in branches: branch_data = { "name": branch.name, "protected": getattr(branch, 'protected', False), "merged": getattr(branch, 'merged', False), "default": branch.name == getattr(project, 'default_branch', 'main'), "web_url": f"{project.web_url}/-/tree/{branch.name}" } branch_list.append(branch_data) return branch_list except GitlabGetError as e: if "404" in str(e): raise GitLabError(f"Project {project_id} not found") else: raise GitLabError(f"Failed to access project: {str(e)}") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error fetching branches: {str(e)}") @retry_on_failure(max_retries=2, delay=0.5) async def commit_files(self, project_id: int, branch_name: str, files: List[Dict[str, Any]], commit_message: str) -> str: """Commit files to a GitLab branch""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._commit_files_sync, project_id, branch_name, files, commit_message) logger.info(f"Committed {len(files)} files to branch '{branch_name}' in project {project_id}") return result except Exception as e: logger.error(f"Failed to commit files to branch '{branch_name}': {e}") raise GitLabError(f"Failed to commit files: {str(e)}") def _commit_files_sync(self, project_id: int, branch_name: str, files: List[Dict[str, Any]], commit_message: str) -> str: """Synchronous file commit""" try: project = self.gl.projects.get(project_id) # Prepare commit actions actions = [] for file_data in files: action = { 'action': file_data.get('action', 'create'), # create, update, delete 'file_path': file_data['path'], 'content': file_data.get('content', ''), } # Add encoding if specified if file_data.get('encoding'): action['encoding'] = file_data['encoding'] actions.append(action) # Create commit commit_data = { 'branch': branch_name, 'commit_message': commit_message, 'actions': actions } commit = project.commits.create(commit_data) commit_url = f"{project.web_url}/-/commit/{commit.id}" return commit_url except GitlabGetError as e: if "404" in str(e): raise GitLabError(f"Project {project_id} or branch '{branch_name}' not found") else: raise GitLabError(f"Failed to access project/branch: {str(e)}") except GitlabCreateError as e: raise GitLabError(f"Failed to create commit: {str(e)}") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error committing files: {str(e)}") @retry_on_failure(max_retries=2, delay=0.5) async def create_merge_request(self, project_id: int, source_branch: str, target_branch: str = "main", title: str = "", description: str = "", draft: bool = True) -> str: """Create a merge request in GitLab""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._create_merge_request_sync, project_id, source_branch, target_branch, title, description, draft) logger.info(f"Created merge request from '{source_branch}' to '{target_branch}' in project {project_id}") return result except Exception as e: logger.error(f"Failed to create merge request: {e}") raise GitLabError(f"Failed to create merge request: {str(e)}") def _create_merge_request_sync(self, project_id: int, source_branch: str, target_branch: str, title: str, description: str, draft: bool) -> str: """Synchronous merge request creation""" try: project = self.gl.projects.get(project_id) # Prepare MR data mr_data = { 'source_branch': source_branch, 'target_branch': target_branch, 'title': title or f"AI-generated fix from {source_branch}", 'description': description or "Automated fix generated by AI assistant", } # Mark as draft if requested if draft: mr_data['title'] = f"Draft: {mr_data['title']}" # Create merge request mr = project.mergerequests.create(mr_data) mr_url = mr.web_url return mr_url except GitlabGetError as e: if "404" in str(e): raise GitLabError(f"Project {project_id} not found") else: raise GitLabError(f"Failed to access project: {str(e)}") except GitlabCreateError as e: if "already exists" in str(e).lower(): # MR already exists, try to find and return it try: mrs = project.mergerequests.list(source_branch=source_branch, target_branch=target_branch) if mrs: return mrs[0].web_url except: pass raise GitLabError(f"Merge request already exists: {str(e)}") else: raise GitLabError(f"Failed to create merge request: {str(e)}") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error creating merge request: {str(e)}") @retry_on_failure(max_retries=2, delay=0.5) async def get_file_content(self, project_id: int, file_path: str, branch: str = "main") -> Optional[str]: """Get file content from GitLab repository""" try: loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._get_file_content_sync, project_id, file_path, branch) logger.info(f"Retrieved file '{file_path}' from branch '{branch}' in project {project_id}") return result except Exception as e: logger.error(f"Failed to get file content: {e}") raise GitLabError(f"Failed to get file content: {str(e)}") def _get_file_content_sync(self, project_id: int, file_path: str, branch: str) -> Optional[str]: """Synchronous file content retrieval""" try: project = self.gl.projects.get(project_id) try: file_info = project.files.get(file_path=file_path, ref=branch) # Decode base64 content import base64 content = base64.b64decode(file_info.content).decode('utf-8') return content except GitlabGetError as e: if "404" in str(e): return None # File doesn't exist else: raise GitLabError(f"Failed to get file: {str(e)}") except GitlabGetError as e: if "404" in str(e): raise GitLabError(f"Project {project_id} not found") else: raise GitLabError(f"Failed to access project: {str(e)}") except GitlabError as e: raise GitLabError(f"GitLab API error: {str(e)}") except Exception as e: raise GitLabError(f"Unexpected error getting file content: {str(e)}")