GitLab MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• tools/get_merge_request_reviews.py:135-229 (handler)

  The primary handler function for the 'get_merge_request_reviews' tool. Orchestrates parallel API calls to fetch discussions, approvals, details, pipeline, and changes, then generates a formatted Markdown report with overviews, summaries, detailed threads, and action items.
  async def get_merge_request_reviews(gitlab_url, project_id, access_token, args): logging.info(f"get_merge_request_reviews called with args: {args}") mr_iid = args["merge_request_iid"] tasks = [ api_get_merge_request_reviews(gitlab_url, project_id, access_token, mr_iid), get_merge_request_details(gitlab_url, project_id, access_token, mr_iid), get_merge_request_pipeline(gitlab_url, project_id, access_token, mr_iid), get_merge_request_changes(gitlab_url, project_id, access_token, mr_iid), ] try: reviews_result, details_result, pipeline_result, changes_result = await asyncio.gather(*tasks) except Exception as e: logging.error(f"Error in parallel API calls: {e}") raise Exception(f"Error fetching merge request data: {e}") discussions_status, discussions, discussions_text = reviews_result["discussions"] approvals_status, approvals, approvals_text = reviews_result["approvals"] details_status, mr_details, details_text = details_result pipeline_status, pipeline_data, pipeline_text = pipeline_result changes_status, changes_data, changes_text = changes_result if discussions_status != 200: logging.error(f"Error fetching discussions {discussions_status}: {discussions_text}") raise Exception(f"Error fetching discussions: {discussions_status} - {discussions_text}") result = f"# 🔍 Reviews & Discussions for MR !{mr_iid}\n\n" if details_status == 200: result += "## 📋 Merge Request Overview\n" result += f"**Title**: {mr_details.get('title', 'N/A')}\n" state = mr_details.get("state", "N/A") result += f"**Status**: {state} ({get_state_explanation(state)})\n" author = mr_details.get("author", {}) author_name = author.get("name", "N/A") author_username = author.get("username", "N/A") result += f"**Author**: {author_name} (@{author_username})\n" result += f"**Priority**: {get_mr_priority(mr_details)}\n" if pipeline_status == 200 and pipeline_data: pipeline_icon = get_pipeline_status_icon(pipeline_data.get("status")) result += f"**Pipeline**: {pipeline_icon} {pipeline_data.get('status', 'unknown')}\n" if changes_status == 200: change_stats = calculate_change_stats(changes_data) result += f"**Changes**: {change_stats}\n" readiness = analyze_mr_readiness(mr_details, pipeline_data, approvals) result += f"**Merge Status**: {readiness}\n" result += f"**Updated**: {format_date(mr_details.get('updated_at', 'N/A'))}\n\n" result += get_approval_summary(approvals) result += get_discussion_summary(discussions) if discussions: result += "## 📝 Detailed Discussions\n\n" for discussion in discussions: thread_content = format_discussion_thread(discussion) if thread_content: result += thread_content else: result += "💬 No discussions found\n\n" result += "## 📊 Action Items\n" action_items = [] if discussions: unresolved_count = sum(1 for d in discussions if not d.get("resolved")) if unresolved_count > 0: action_items.append( f"🟡 Resolve {unresolved_count} pending discussion{'s' if unresolved_count > 1 else ''}" ) if approvals and approvals.get("approvals_left", 0) > 0: action_items.append( f"👥 Obtain {approvals['approvals_left']} more approval{'s' if approvals['approvals_left'] > 1 else ''}" ) if pipeline_status == 200 and pipeline_data and pipeline_data.get("status") == "failed": action_items.append("❌ Fix failing pipeline") if details_status == 200 and mr_details.get("has_conflicts"): action_items.append("⚠️ Resolve merge conflicts") if action_items: for item in action_items: result += f"• {item}\n" else: result += "✅ No action items - ready for next steps\n" return [TextContent(type="text", text=result)]
• main.py:69-82 (schema)

  Input schema definition for the tool, specifying that 'merge_request_iid' (integer, min 1) is required and no additional properties allowed.
  name="get_merge_request_reviews", description=("Get reviews and discussions for a specific " "merge request"), inputSchema={ "type": "object", "properties": { "merge_request_iid": { "type": "integer", "minimum": 1, "description": ("Internal ID of the merge request"), } }, "required": ["merge_request_iid"], "additionalProperties": False, },
• main.py:303-306 (registration)

  Registration and dispatch logic in the MCP server's call_tool method, mapping the tool name to its handler invocation.
  elif name == "get_merge_request_reviews": return await get_merge_request_reviews( self.config["gitlab_url"], self.config["project_id"], self.config["access_token"], arguments )
• gitlab_api.py:82-100 (helper)

  Low-level helper function that fetches paginated discussions and approvals data from GitLab API endpoints, used by the main handler.
  async def get_merge_request_reviews(gitlab_url, project_id, access_token, mr_iid): discussions_result = await get_merge_request_discussions_paginated(gitlab_url, project_id, access_token, mr_iid) discussions_status, discussions, discussions_text = discussions_result approvals_url = f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests/{mr_iid}/approvals" headers = _headers(access_token) async with aiohttp.ClientSession() as session: async with session.get(approvals_url, headers=headers) as approvals_response: if approvals_response.status == 200: approvals = await approvals_response.json() else: approvals = None approvals_status = approvals_response.status approvals_text = await approvals_response.text() return { "discussions": (discussions_status, discussions, discussions_text), "approvals": (approvals_status, approvals, approvals_text), }