MCP Server Hero

#!/usr/bin/env python3 """GitLab MCP server example for code review operations.""" import sys import os import json import logging from pathlib import Path from typing import Optional, Dict, Any, List from dataclasses import dataclass from urllib.parse import quote import requests from dotenv import load_dotenv # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent.parent)) import asyncio import mcp.types as types from mcp.server.stdio import stdio_server from mcp_server_hero.core.server import MCPServerHero from mcp_server_hero.middleware.rate_limit import RateLimitMiddleware from mcp_server_hero.auth.base import BaseAuthProvider, AuthContext, Permission # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load environment variables load_dotenv() @dataclass class GitLabConfig: """GitLab configuration settings.""" host: str token: str api_version: str = "v4" class GitLabAuthProvider(BaseAuthProvider): """Authentication provider for GitLab integration.""" def __init__(self) -> None: super().__init__("gitlab") self.gitlab_token = os.getenv("GITLAB_TOKEN", "") async def authenticate(self, credentials: dict[str, any]) -> AuthContext: """Authenticate using GitLab token.""" token = credentials.get("token", self.gitlab_token) if not token: return AuthContext(authenticated=False) # Verify token by making a test API request try: config = GitLabConfig(host=os.getenv("GITLAB_HOST", "gitlab.com"), token=token) # Test API access url = f"https://{config.host}/api/{config.api_version}/user" headers = {"Accept": "application/json", "Private-Token": token} response = requests.get(url, headers=headers, timeout=10) if response.status_code == 200: user_data = response.json() return AuthContext( user_id=str(user_data.get("id", "unknown")), authenticated=True, permissions={ Permission.ADMIN, Permission.READ_TOOLS, Permission.CALL_TOOLS, Permission.READ_RESOURCES, }, metadata={"username": user_data.get("username", ""), "name": user_data.get("name", "")}, ) else: return AuthContext(authenticated=False) except Exception as e: logger.error(f"GitLab authentication failed: {e}") return AuthContext(authenticated=False) async def authorize(self, context: AuthContext, resource: str, action: str) -> bool: """Authorize access to GitLab resources.""" return context.authenticated def make_gitlab_api_request(endpoint: str, method: str = "GET", data: Optional[Dict[str, Any]] = None) -> Any: """Make a REST API request to GitLab and handle the response.""" config = GitLabConfig(host=os.getenv("GITLAB_HOST", "gitlab.com"), token=os.getenv("GITLAB_TOKEN", "")) if not config.token: logger.error("GitLab token not set") raise ValueError("GitLab token not set. Please set GITLAB_TOKEN in your environment.") url = f"https://{config.host}/api/{config.api_version}/{endpoint}" headers = {"Accept": "application/json", "User-Agent": "GitLabMCPCodeReview/1.0", "Private-Token": config.token} try: if method.upper() == "GET": response = requests.get(url, headers=headers, verify=True, timeout=30) elif method.upper() == "POST": response = requests.post(url, headers=headers, json=data, verify=True, timeout=30) else: raise ValueError(f"Unsupported HTTP method: {method}") if response.status_code == 401: logger.error("Authentication failed. Check your GitLab token.") raise Exception("Authentication failed. Please check your GitLab token.") response.raise_for_status() if not response.content: return {} try: return response.json() except json.JSONDecodeError as e: logger.error(f"Failed to parse JSON response: {str(e)}") raise Exception(f"Failed to parse GitLab response as JSON: {str(e)}") except requests.exceptions.RequestException as e: logger.error(f"REST request failed: {str(e)}") if hasattr(e, "response"): logger.error(f"Response status: {e.response.status_code}") raise Exception(f"Failed to make GitLab API request: {str(e)}") async def fetch_merge_request(project_id: str, merge_request_iid: str) -> str: """ Fetch a GitLab merge request and its contents. Args: project_id: The GitLab project ID or URL-encoded path merge_request_iid: The merge request IID (project-specific ID) Returns: JSON string containing the merge request information """ try: # Get merge request details mr_endpoint = f"projects/{quote(project_id, safe='')}/merge_requests/{merge_request_iid}" mr_info = make_gitlab_api_request(mr_endpoint) if not mr_info: raise ValueError(f"Merge request {merge_request_iid} not found in project {project_id}") # Get the changes (diffs) for this merge request changes_endpoint = f"{mr_endpoint}/changes" changes_info = make_gitlab_api_request(changes_endpoint) # Get the commit information commits_endpoint = f"{mr_endpoint}/commits" commits_info = make_gitlab_api_request(commits_endpoint) # Get the notes (comments) for this merge request notes_endpoint = f"{mr_endpoint}/notes" notes_info = make_gitlab_api_request(notes_endpoint) result = {"merge_request": mr_info, "changes": changes_info, "commits": commits_info, "notes": notes_info} return json.dumps(result, indent=2) except Exception as e: return f"Error fetching merge request: {str(e)}" async def fetch_merge_request_diff(project_id: str, merge_request_iid: str, file_path: Optional[str] = None) -> str: """ Fetch the diff for a specific file in a merge request, or all files if none specified. Args: project_id: The GitLab project ID or URL-encoded path merge_request_iid: The merge request IID (project-specific ID) file_path: Optional specific file path to get diff for Returns: JSON string containing the diff information """ try: # Get the changes for this merge request changes_endpoint = f"projects/{quote(project_id, safe='')}/merge_requests/{merge_request_iid}/changes" changes_info = make_gitlab_api_request(changes_endpoint) if not changes_info: raise ValueError(f"Changes not found for merge request {merge_request_iid}") # Extract all changes files = changes_info.get("changes", []) # Filter by file path if specified if file_path: files = [f for f in files if f.get("new_path") == file_path or f.get("old_path") == file_path] if not files: raise ValueError(f"File '{file_path}' not found in the merge request changes") result = {"merge_request_iid": merge_request_iid, "files": files} return json.dumps(result, indent=2) except Exception as e: return f"Error fetching merge request diff: {str(e)}" async def fetch_commit_diff(project_id: str, commit_sha: str, file_path: Optional[str] = None) -> str: """ Fetch the diff for a specific commit, or for a specific file in that commit. Args: project_id: The GitLab project ID or URL-encoded path commit_sha: The commit SHA file_path: Optional specific file path to get diff for Returns: JSON string containing the diff information """ try: # Get the diff for this commit diff_endpoint = f"projects/{quote(project_id, safe='')}/repository/commits/{commit_sha}/diff" diff_info = make_gitlab_api_request(diff_endpoint) if not diff_info: raise ValueError(f"Diff not found for commit {commit_sha}") # Filter by file path if specified if file_path: diff_info = [d for d in diff_info if d.get("new_path") == file_path or d.get("old_path") == file_path] if not diff_info: raise ValueError(f"File '{file_path}' not found in the commit diff") # Get the commit details commit_endpoint = f"projects/{quote(project_id, safe='')}/repository/commits/{commit_sha}" commit_info = make_gitlab_api_request(commit_endpoint) result = {"commit": commit_info, "diffs": diff_info} return json.dumps(result, indent=2) except Exception as e: return f"Error fetching commit diff: {str(e)}" async def compare_versions(project_id: str, from_sha: str, to_sha: str) -> str: """ Compare two commits/branches/tags to see the differences between them. Args: project_id: The GitLab project ID or URL-encoded path from_sha: The source commit/branch/tag to_sha: The target commit/branch/tag Returns: JSON string containing the comparison information """ try: # Compare the versions compare_endpoint = f"projects/{quote(project_id, safe='')}/repository/compare?from={quote(from_sha, safe='')}&to={quote(to_sha, safe='')}" compare_info = make_gitlab_api_request(compare_endpoint) if not compare_info: raise ValueError(f"Comparison failed between {from_sha} and {to_sha}") return json.dumps(compare_info, indent=2) except Exception as e: return f"Error comparing versions: {str(e)}" async def add_merge_request_comment( project_id: str, merge_request_iid: str, body: str, position: Optional[Dict[str, Any]] = None ) -> str: """ Add a comment to a merge request, optionally at a specific position in a file. Args: project_id: The GitLab project ID or URL-encoded path merge_request_iid: The merge request IID (project-specific ID) body: The comment text position: Optional position data for line comments (JSON string) Returns: JSON string containing the created comment information """ try: # Create the comment data data = {"body": body} # Parse position data if provided if position: if isinstance(position, str): data["position"] = json.loads(position) else: data["position"] = position # Add the comment comment_endpoint = f"projects/{quote(project_id, safe='')}/merge_requests/{merge_request_iid}/notes" comment_info = make_gitlab_api_request(comment_endpoint, method="POST", data=data) if not comment_info: raise ValueError("Failed to add comment to merge request") return json.dumps(comment_info, indent=2) except Exception as e: return f"Error adding merge request comment: {str(e)}" async def approve_merge_request( project_id: str, merge_request_iid: str, approvals_required: Optional[int] = None ) -> str: """ Approve a merge request. Args: project_id: The GitLab project ID or URL-encoded path merge_request_iid: The merge request IID (project-specific ID) approvals_required: Optional number of required approvals to set Returns: JSON string containing the approval information """ try: # Approve the merge request approve_endpoint = f"projects/{quote(project_id, safe='')}/merge_requests/{merge_request_iid}/approve" approve_info = make_gitlab_api_request(approve_endpoint, method="POST") # Set required approvals if specified if approvals_required is not None: approvals_endpoint = f"projects/{quote(project_id, safe='')}/merge_requests/{merge_request_iid}/approvals" data = {"approvals_required": approvals_required} make_gitlab_api_request(approvals_endpoint, method="POST", data=data) return json.dumps(approve_info, indent=2) except Exception as e: return f"Error approving merge request: {str(e)}" async def get_project_merge_requests(project_id: str, state: str = "all", limit: int = 20) -> str: """ Get all merge requests for a project. Args: project_id: The GitLab project ID or URL-encoded path state: Filter merge requests by state (all, opened, closed, merged, or locked) limit: Maximum number of merge requests to return Returns: JSON string containing list of merge request objects """ try: # Get the merge requests mrs_endpoint = f"projects/{quote(project_id, safe='')}/merge_requests?state={state}&per_page={limit}" mrs_info = make_gitlab_api_request(mrs_endpoint) return json.dumps(mrs_info, indent=2) except Exception as e: return f"Error fetching project merge requests: {str(e)}" async def get_gitlab_status() -> str: """Get GitLab connection status and configuration.""" config = GitLabConfig(host=os.getenv("GITLAB_HOST", "gitlab.com"), token=os.getenv("GITLAB_TOKEN", "")) status = {"host": config.host, "token_configured": bool(config.token), "api_version": config.api_version} if config.token: try: # Test connection user_endpoint = "user" user_info = make_gitlab_api_request(user_endpoint) status["connection"] = "success" status["user"] = { "username": user_info.get("username", ""), "name": user_info.get("name", ""), "id": user_info.get("id", ""), } except Exception as e: status["connection"] = "failed" status["error"] = str(e) return json.dumps(status, indent=2) async def main() -> None: """Run the GitLab MCP server.""" # Create server with GitLab configuration server = MCPServerHero(name="GitLab Code Review Server", debug=True) # Add rate limiting for API calls server.add_middleware( RateLimitMiddleware(tool_limit=30, resource_limit=50, prompt_limit=10, refill_rate=0.5, window_size=60) ) # Setup GitLab authentication auth_provider = GitLabAuthProvider() server.enable_auth_provider(auth_provider) # Register GitLab tools server.add_tool( name="fetch_merge_request", tool_func=fetch_merge_request, description="Fetch a GitLab merge request and its contents including changes, commits, and comments", schema={ "type": "object", "properties": { "project_id": { "type": "string", "description": "The GitLab project ID or URL-encoded path (e.g., 'group/project' or '12345')", }, "merge_request_iid": { "type": "string", "description": "The merge request IID (project-specific ID, not the global ID)", }, }, "required": ["project_id", "merge_request_iid"], }, ) server.add_tool( name="fetch_merge_request_diff", tool_func=fetch_merge_request_diff, description="Fetch the diff for files in a merge request, optionally filtered by specific file path", schema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "The GitLab project ID or URL-encoded path"}, "merge_request_iid": {"type": "string", "description": "The merge request IID (project-specific ID)"}, "file_path": {"type": "string", "description": "Optional specific file path to get diff for"}, }, "required": ["project_id", "merge_request_iid"], }, ) server.add_tool( name="fetch_commit_diff", tool_func=fetch_commit_diff, description="Fetch the diff for a specific commit, optionally filtered by file path", schema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "The GitLab project ID or URL-encoded path"}, "commit_sha": {"type": "string", "description": "The commit SHA"}, "file_path": {"type": "string", "description": "Optional specific file path to get diff for"}, }, "required": ["project_id", "commit_sha"], }, ) server.add_tool( name="compare_versions", tool_func=compare_versions, description="Compare two commits/branches/tags to see the differences between them", schema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "The GitLab project ID or URL-encoded path"}, "from_sha": {"type": "string", "description": "The source commit/branch/tag"}, "to_sha": {"type": "string", "description": "The target commit/branch/tag"}, }, "required": ["project_id", "from_sha", "to_sha"], }, ) server.add_tool( name="add_merge_request_comment", tool_func=add_merge_request_comment, description="Add a comment to a merge request, optionally at a specific position in a file", schema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "The GitLab project ID or URL-encoded path"}, "merge_request_iid": {"type": "string", "description": "The merge request IID (project-specific ID)"}, "body": {"type": "string", "description": "The comment text"}, "position": { "type": "string", "description": "Optional position data for line comments as JSON string", }, }, "required": ["project_id", "merge_request_iid", "body"], }, ) server.add_tool( name="approve_merge_request", tool_func=approve_merge_request, description="Approve a merge request and optionally set required approvals", schema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "The GitLab project ID or URL-encoded path"}, "merge_request_iid": {"type": "string", "description": "The merge request IID (project-specific ID)"}, "approvals_required": { "type": "integer", "description": "Optional number of required approvals to set", }, }, "required": ["project_id", "merge_request_iid"], }, ) server.add_tool( name="get_project_merge_requests", tool_func=get_project_merge_requests, description="Get all merge requests for a project with optional filtering by state", schema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "The GitLab project ID or URL-encoded path"}, "state": { "type": "string", "enum": ["all", "opened", "closed", "merged", "locked"], "description": "Filter merge requests by state", "default": "all", }, "limit": { "type": "integer", "description": "Maximum number of merge requests to return", "default": 20, "minimum": 1, "maximum": 100, }, }, "required": ["project_id"], }, ) # Register GitLab status resource server.add_resource( uri="gitlab://status", resource_func=get_gitlab_status, name="gitlab_status", description="GitLab connection status and configuration information", ) # Initialize server await server.initialize() print("🦊 GitLab Code Review MCP Server started with features:") print(" ✅ Middleware: Validation, Logging, Timing, Rate Limiting") print(" ✅ Authentication: GitLab token-based auth") print(" ✅ Tools: Merge Request operations, Diff analysis, Comments") print(" ✅ Resources: GitLab connection status") print(" 📋 Environment variables required:") print(" - GITLAB_TOKEN: Your GitLab personal access token") print(" - GITLAB_HOST: GitLab host (default: gitlab.com)") try: # Run the server async with stdio_server() as streams: await server.run(streams[0], streams[1], server.create_initialization_options()) finally: await server.shutdown() if __name__ == "__main__": asyncio.run(main())