# Directory Structure
```
├── .env.example
├── .gitignore
├── config.py
├── gitlab_api.py
├── logging_config.py
├── main.py
├── README.md
├── requirements.txt
├── run-mcp.sh
├── test_tools.py
├── tools
│ ├── __init__.py
│ ├── get_branch_merge_requests.py
│ ├── get_commit_discussions.py
│ ├── get_merge_request_details.py
│ ├── get_merge_request_reviews.py
│ ├── list_merge_requests.py
│ └── reply_to_review_comment.py
└── utils.py
```
# Files
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# GitLab MCP Server Configuration
# Copy this file to .env and fill in your actual values
# Required Environment Variables
GITLAB_PROJECT_ID=your-project-id
GITLAB_ACCESS_TOKEN=your-access-token
# Optional Environment Variables (defaults shown)
GITLAB_URL=https://gitlab.com
SERVER_NAME=gitlab-mcp-server
SERVER_VERSION=1.0.0
# Instructions:
# 1. Copy this file: cp .env.example .env
# 2. Edit .env with your actual GitLab project ID and access token
# 3. For GitLab.com, you can leave GITLAB_URL as is
# 4. For self-hosted GitLab, update GITLAB_URL to your instance URL
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
Pipfile.lock
# poetry
poetry.lock
# mypy
.mypy_cache/
.dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# VS Code
.vscode/
# JetBrains IDEs
.idea/
# Mac OS
.DS_Store
# Environment variables
.env
.env.*
!.env.example
# MCP/AI logs
*.log
# System
Thumbs.db
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# GitLab MCP Server
Connect your AI assistant to GitLab. Ask questions like _"List open merge requests"_, _"Show me reviews for MR #123"_, _"Get commit discussions for MR #456"_, or _"Find merge requests for the feature branch"_ directly in your chat.
## Table of Contents
- [Quick Setup](#quick-setup)
- [What You Can Do](#what-you-can-do)
- [Configuration Options](#configuration-options)
- [Troubleshooting](#troubleshooting)
- [Tool Reference](#tool-reference)
- [Development](#development)
- [Security Notes](#security-notes)
- [Support](#support)
## Quick Setup
1. **Install the server:**
```bash
git clone https://github.com/amirsina-mandegari/gitlab-mcp-server.git
cd gitlab-mcp-server
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
chmod +x run-mcp.sh
```
2. **Get your GitLab token:**
- Go to GitLab → Settings → Access Tokens
- Create token with **`read_api`** scope
- Copy the token
3. **Configure your project:**
In your project directory, create `gitlab-mcp.env`:
```env
GITLAB_PROJECT_ID=12345
GITLAB_ACCESS_TOKEN=glpat-xxxxxxxxxxxxxxxxxxxx
GITLAB_URL=https://gitlab.com
```
4. **Connect to Cursor:**
Create `.cursor/mcp.json` in your project:
```json
{
"mcpServers": {
"gitlab-mcp": {
"command": "/path/to/gitlab-mcp-server/run-mcp.sh",
"cwd": "/path/to/your-project"
}
}
}
```
5. **Restart Cursor** and start asking GitLab questions!
## What You Can Do
Once connected, try these commands in your chat:
- _"List open merge requests"_
- _"Show me details for merge request 456"_
- _"Get reviews and discussions for MR #123"_
- _"Show me commit discussions for MR #456"_
- _"Get all comments on commits in merge request #789"_
- _"Find merge requests for the feature/auth-improvements branch"_
- _"Show me closed merge requests targeting main"_
- _"Reply to discussion abc123 in MR #456 with 'Thanks for the feedback!'"_
- _"Create a new review comment in MR #789 asking about the error handling"_
- _"Resolve discussion def456 in MR #123"_
## Working with Review Comments
The enhanced review tools allow you to interact with merge request discussions:
1. **First, get the reviews** to see discussion IDs:
```
"Show me reviews for MR #123"
```
2. **Reply to specific discussions** using the discussion ID:
```
"Reply to discussion abc123 in MR #456 with 'I'll fix this in the next commit'"
```
3. **Create new discussion threads** to start conversations:
```
"Create a review comment in MR #789 asking 'Could you add error handling here?'"
```
4. **Resolve discussions** when issues are addressed:
```
"Resolve discussion def456 in MR #123"
```
**Note**: The `get_merge_request_reviews` tool now displays discussion IDs and note IDs in the output, making it easy to reference specific discussions when replying or resolving.
## Working with Commit Discussions
The `get_commit_discussions` tool provides comprehensive insights into discussions and comments on individual commits within a merge request:
1. **View all commit discussions** for a merge request:
```
"Show me commit discussions for MR #123"
```
2. **Get detailed commit conversation history**:
```
"Get all comments on commits in merge request #456"
```
This tool is particularly useful for:
- **Code Review Tracking**: See all feedback on specific commits
- **Discussion History**: Understand the evolution of code discussions
- **Commit-Level Context**: View comments tied to specific code changes
- **Review Progress**: Monitor which commits have been discussed
**Technical Implementation:**
- Uses `/projects/:project_id/merge_requests/:merge_request_iid/commits` to get all commits with proper pagination
- Fetches ALL merge request discussions using `/projects/:project_id/merge_requests/:merge_request_iid/discussions` with pagination support
- Filters discussions by commit SHA using position data to show commit-specific conversations
- Handles both individual comments and discussion threads correctly
The output includes:
- Summary of total commits and discussion counts
- Individual commit details (SHA, title, author, date)
- All discussions and comments for each commit with file positions
- Complete conversation threads with replies
- File positions for diff-related comments
- Thread conversations with replies
## Configuration Options
### Project-Level (Recommended)
Each project gets its own `gitlab-mcp.env` file with its own GitLab configuration. Keep tokens out of version control.
### Global Configuration
Set environment variables system-wide instead of per-project:
```bash
export GITLAB_PROJECT_ID=12345
export GITLAB_ACCESS_TOKEN=glpat-xxxxxxxxxxxxxxxxxxxx
export GITLAB_URL=https://gitlab.com
```
### Find Your Project ID
- Go to your GitLab project → Settings → General → Project ID
- Or check the URL: `https://gitlab.com/username/project` (use the numeric ID)
## Troubleshooting
**Authentication Error**: Verify your token has `read_api` permissions and is not expired.
**Project Not Found**: Double-check your project ID is correct (it's a number, not the project name).
**Connection Issues**: Make sure your GitLab URL is accessible and correct.
**Script Not Found**: Ensure the path in your MCP config points to the actual server location and the script is executable.
## Tool Reference
| Tool | Description | Parameters |
| --------------------------- | ---------------------------- | ------------------------------------------------ |
| `list_merge_requests` | List merge requests | `state`, `target_branch`, `limit` |
| `get_merge_request_details` | Get MR details | `merge_request_iid` |
| `get_merge_request_reviews` | Get reviews/discussions | `merge_request_iid` |
| `get_commit_discussions` | Get discussions on commits | `merge_request_iid` |
| `get_branch_merge_requests` | Find MRs for branch | `branch_name` |
| `reply_to_review_comment` | Reply to existing discussion | `merge_request_iid`, `discussion_id`, `body` |
| `create_review_comment` | Create new discussion thread | `merge_request_iid`, `body` |
| `resolve_review_discussion` | Resolve/unresolve discussion | `merge_request_iid`, `discussion_id`, `resolved` |
## Development
### Project Structure
```
gitlab-mcp-server/
├── main.py # MCP server entry point
├── config.py # Configuration management
├── gitlab_api.py # GitLab API client
├── utils.py # Utility functions
├── logging_config.py # Logging configuration
├── run-mcp.sh # Launch script
└── tools/ # Tool implementations package
├── __init__.py # Package initialization
├── list_merge_requests.py
├── get_merge_request_details.py
├── get_merge_request_reviews.py
├── get_commit_discussions.py
├── get_branch_merge_requests.py
└── reply_to_review_comment.py
```
### Adding Tools
1. Create new file in `tools/` directory
2. Add import and export to `tools/__init__.py`
3. Add to `list_tools()` in `main.py`
4. Add handler to `call_tool()` in `main.py`
### Testing
```bash
python test_tools.py
```
## Security Notes
- Add `gitlab-mcp.env` to your `.gitignore`
- Never commit access tokens
- Use project-specific tokens with minimal permissions
- Rotate tokens regularly
## Support
- Check [GitLab API documentation](https://docs.gitlab.com/ee/api/)
- Open issues at [github.com/amirsina-mandegari/gitlab-mcp-server](https://github.com/amirsina-mandegari/gitlab-mcp-server)
## License
MIT License - see LICENSE file for details.
```
--------------------------------------------------------------------------------
/logging_config.py:
--------------------------------------------------------------------------------
```python
import logging
def configure_logging():
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
aiohappyeyeballs==2.6.1
aiohttp==3.12.14
aiosignal==1.4.0
annotated-types==0.7.0
anyio==4.9.0
attrs==25.3.0
certifi==2025.7.14
click==8.2.1
frozenlist==1.7.0
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
httpx-sse==0.4.1
idna==3.10
jsonschema==4.24.0
jsonschema-specifications==2025.4.1
mcp==1.11.0
multidict==6.6.3
propcache==0.3.2
pydantic==2.11.7
pydantic-settings==2.10.1
pydantic_core==2.33.2
python-dotenv==1.1.1
python-multipart==0.0.20
python-decouple
referencing==0.36.2
rpds-py==0.26.0
setuptools==62.1.0
sniffio==1.3.1
sse-starlette==2.4.1
starlette==0.47.1
typing-inspection==0.4.1
typing_extensions==4.14.1
uvicorn==0.35.0
wheel==0.37.1
yarl==1.20.1
```
--------------------------------------------------------------------------------
/run-mcp.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Get the script directory (where the MCP server is located)
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Get the current working directory (where the script is being run from)
CURRENT_DIR="$(pwd)"
# Check if gitlab-mcp.env exists in the current working directory and load it
if [ -f "$CURRENT_DIR/gitlab-mcp.env" ]; then
echo "Loading environment variables from $CURRENT_DIR/gitlab-mcp.env"
set -a
source "$CURRENT_DIR/gitlab-mcp.env"
set +a
else
echo "No gitlab-mcp.env found in $CURRENT_DIR, using existing environment variables"
fi
# Activate the virtual environment and run the MCP server
source "$SCRIPT_DIR/.venv/bin/activate"
python "$SCRIPT_DIR/main.py" 2>&1 | tee /tmp/gitlab-mcp-server.log
```
--------------------------------------------------------------------------------
/tools/__init__.py:
--------------------------------------------------------------------------------
```python
"""
GitLab MCP Server Tools Package
This package contains all the tool implementations for the GitLab MCP server.
Each tool provides specific functionality for interacting with GitLab's API.
"""
from .list_merge_requests import list_merge_requests
from .get_merge_request_reviews import get_merge_request_reviews
from .get_merge_request_details import get_merge_request_details
from .get_branch_merge_requests import get_branch_merge_requests
from .reply_to_review_comment import (
reply_to_review_comment,
create_review_comment,
resolve_review_discussion
)
from .get_commit_discussions import get_commit_discussions
__all__ = [
"list_merge_requests",
"get_merge_request_reviews",
"get_merge_request_details",
"get_branch_merge_requests",
"reply_to_review_comment",
"create_review_comment",
"resolve_review_discussion",
"get_commit_discussions"
]
```
--------------------------------------------------------------------------------
/config.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Configuration management for GitLab MCP Server."""
from decouple import config
def get_gitlab_config():
"""Get GitLab configuration from environment variables."""
gitlab_url = config("GITLAB_URL", default="https://gitlab.com")
project_id = config("GITLAB_PROJECT_ID")
access_token = config("GITLAB_ACCESS_TOKEN")
if not project_id:
raise ValueError(
"GITLAB_PROJECT_ID environment variable is required"
)
if not access_token:
raise ValueError(
"GITLAB_ACCESS_TOKEN environment variable is required"
)
return {
'gitlab_url': gitlab_url,
'project_id': project_id,
'access_token': access_token,
'server_name': config("SERVER_NAME", default="gitlab-mcp-server"),
'server_version': config("SERVER_VERSION", default="1.0.0")
}
def get_headers(access_token):
"""Get HTTP headers for GitLab API requests."""
return {
"Private-Token": access_token,
"Content-Type": "application/json"
}
```
--------------------------------------------------------------------------------
/test_tools.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Test script to verify GitLab MCP tools return proper CallToolResult objects."""
import asyncio
import sys
from tools.list_merge_requests import list_merge_requests
from tools.get_merge_request_details import get_merge_request_details
from tools.get_merge_request_reviews import get_merge_request_reviews
from mcp.types import CallToolResult
from decouple import config
GITLAB_URL = config('GITLAB_URL', default='https://git.partnerz.io')
PROJECT_ID = config('GITLAB_PROJECT_ID', default='237')
ACCESS_TOKEN = config('GITLAB_ACCESS_TOKEN', default='')
async def test_list_merge_requests():
"""Test list_merge_requests tool."""
print("Testing list_merge_requests...")
args = {"state": "opened", "limit": 2}
result = await list_merge_requests(GITLAB_URL, PROJECT_ID, ACCESS_TOKEN, args)
print(f"Result type: {type(result)}")
print(f"Is CallToolResult: {isinstance(result, CallToolResult)}")
if isinstance(result, CallToolResult):
print(f"isError: {result.isError}")
print(f"Content type: {type(result.content)}")
if result.content:
print(f"First content item: {result.content[0].text[:100]}...")
print("=" * 50)
async def test_get_merge_request_details():
"""Test get_merge_request_details tool."""
print("Testing get_merge_request_details...")
args = {"merge_request_iid": 1047}
result = await get_merge_request_details(
GITLAB_URL, PROJECT_ID, ACCESS_TOKEN, args
)
print(f"Result type: {type(result)}")
print(f"Is CallToolResult: {isinstance(result, CallToolResult)}")
if isinstance(result, CallToolResult):
print(f"isError: {result.isError}")
print(f"Content type: {type(result.content)}")
if result.content:
print(f"First content item: {result.content[0].text[:100]}...")
print("=" * 50)
async def test_get_merge_request_reviews():
"""Test get_merge_request_reviews tool."""
print("Testing get_merge_request_reviews...")
args = {"merge_request_iid": 1047}
result = await get_merge_request_reviews(
GITLAB_URL, PROJECT_ID, ACCESS_TOKEN, args
)
print(f"Result type: {type(result)}")
print(f"Is CallToolResult: {isinstance(result, CallToolResult)}")
if isinstance(result, CallToolResult):
print(f"isError: {result.isError}")
print(f"Content type: {type(result.content)}")
if result.content:
print(f"Full content: {result.content[0].text}")
print("=" * 50)
async def main():
"""Run all tests."""
if not ACCESS_TOKEN:
print("Error: GITLAB_ACCESS_TOKEN not set")
sys.exit(1)
await test_list_merge_requests()
await test_get_merge_request_details()
await test_get_merge_request_reviews()
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/utils.py:
--------------------------------------------------------------------------------
```python
from datetime import datetime
def format_date(iso_date_string):
"""Convert ISO date to human-readable format"""
try:
dt = datetime.fromisoformat(iso_date_string.replace('Z', '+00:00'))
return dt.strftime('%Y-%m-%d %H:%M UTC')
except ValueError:
return iso_date_string
def get_state_explanation(state):
"""Get human-readable explanation of MR state"""
explanations = {
'opened': 'Ready for review',
'merged': 'Successfully merged',
'closed': 'Closed without merging',
'locked': 'Locked (no new discussions)',
'draft': 'Work in progress'
}
return explanations.get(state, state)
def get_pipeline_status_icon(status):
"""Get emoji for pipeline status"""
if not status:
return '⚪'
icons = {
'success': '✅',
'failed': '❌',
'running': '🔄',
'pending': '⏳',
'canceled': '⏹️',
'skipped': '⏭️',
'manual': '👤'
}
return icons.get(status, '❓')
def calculate_change_stats(changes):
"""Calculate lines added/removed from changes"""
if not changes or 'changes' not in changes:
return "No changes"
additions = 0
deletions = 0
for change in changes['changes']:
if 'diff' in change:
diff_lines = change['diff'].split('\n')
for line in diff_lines:
if line.startswith('+') and not line.startswith('+++'):
additions += 1
elif line.startswith('-') and not line.startswith('---'):
deletions += 1
return f"+{additions}/-{deletions}"
def analyze_mr_readiness(mr_data, pipeline_data=None, approvals=None):
"""Analyze if MR is ready to merge and what's blocking it"""
blockers = []
if mr_data.get('draft') or mr_data.get('work_in_progress'):
blockers.append("🚧 Draft/WIP status")
if mr_data.get('has_conflicts'):
blockers.append("⚠️ Merge conflicts")
if pipeline_data and pipeline_data.get('status') == 'failed':
blockers.append("❌ Pipeline failed")
elif pipeline_data and pipeline_data.get('status') == 'running':
blockers.append("🔄 Pipeline running")
if approvals and 'approvals_required' in approvals:
approved_count = len(approvals.get('approved_by', []))
required_count = approvals.get('approvals_required', 0)
if approved_count < required_count:
msg = f"👥 Needs approval ({approved_count}/{required_count})"
blockers.append(msg)
if mr_data.get('merge_status') == 'cannot_be_merged':
blockers.append("🚫 Cannot be merged")
if not blockers:
return "✅ Ready to merge"
else:
return f"🚫 Blocked by: {', '.join(blockers)}"
def get_mr_priority(mr_data):
"""Determine MR priority based on labels and other factors"""
labels = mr_data.get('labels', [])
for label in labels:
if 'critical' in label.lower() or 'urgent' in label.lower():
return '🔴 Critical'
elif 'high' in label.lower():
return '🟡 High'
elif 'low' in label.lower():
return '🟢 Low'
return '⚪ Normal'
```
--------------------------------------------------------------------------------
/tools/get_commit_discussions.py:
--------------------------------------------------------------------------------
```python
from mcp.types import TextContent
from gitlab_api import (
get_merge_request_commits,
get_merge_request_discussions_paginated
)
from utils import format_date
import logging
async def get_commit_discussions(
gitlab_url, project_id, access_token, args
):
"""Get discussions/comments on commits within a merge request"""
logging.info(f"get_commit_discussions called with args: {args}")
mr_iid = args["merge_request_iid"]
try:
commits_result = await get_merge_request_commits(
gitlab_url, project_id, access_token, mr_iid
)
commits_status, commits_data, commits_error = commits_result
if commits_status != 200:
logging.error(
f"Error fetching merge request commits: "
f"{commits_status} - {commits_error}"
)
raise Exception(
f"Error fetching merge request commits: {commits_error}"
)
if not commits_data:
return [TextContent(
type="text",
text="No commits found in this merge request."
)]
logging.info(f"Getting ALL MR discussions for MR #{mr_iid}...")
discussions_result = await get_merge_request_discussions_paginated(
gitlab_url, project_id, access_token, mr_iid
)
discussions_status, discussions_data, discussions_error = discussions_result
if discussions_status != 200:
logging.error(
f"Error fetching MR discussions: "
f"{discussions_status} - {discussions_error}"
)
discussions_data = []
commit_map = {commit['id']: commit for commit in commits_data}
commits_with_discussions = {}
total_discussions = 0
for discussion in discussions_data:
notes = discussion.get('notes', [])
for note in notes:
position = note.get('position')
if position and position.get('head_sha'):
commit_sha = position['head_sha']
if commit_sha in commit_map:
if commit_sha not in commits_with_discussions:
commits_with_discussions[commit_sha] = {
'commit': commit_map[commit_sha],
'discussions': []
}
commits_with_discussions[commit_sha]['discussions'].append({
'discussion_id': discussion.get('id'),
'note': note,
'position': position
})
total_discussions += 1
if not commits_with_discussions:
summary_text = (
f"## Commit Discussions for MR #{mr_iid}\n\n"
f"**Summary:**\n"
f"- Total commits: {len(commits_data)}\n"
f"- Commits with discussions: 0\n"
f"- Total discussions: 0\n\n"
f"No line-level discussions found on any commits in this "
f"merge request. Found {len(discussions_data)} total MR discussions."
)
return [TextContent(type="text", text=summary_text)]
response_text = (
f"## Commit Discussions for MR #{mr_iid}\n\n"
f"**Summary:**\n"
f"- Total commits: {len(commits_data)}\n"
f"- Commits with discussions: {len(commits_with_discussions)}\n"
f"- Total line-level discussions: {total_discussions}\n"
f"- Total MR discussions: {len(discussions_data)}\n\n"
)
for commit_sha, item in commits_with_discussions.items():
commit = item['commit']
discussions = item['discussions']
response_text += f"### 📝 Commit: {commit['short_id']}\n"
response_text += f"**Title:** {commit['title']}\n"
response_text += f"**Author:** {commit['author_name']}\n"
response_text += (
f"**Date:** {format_date(commit['committed_date'])}\n"
)
response_text += f"**SHA:** `{commit['id']}`\n\n"
for discussion_item in discussions:
discussion_id = discussion_item['discussion_id']
note = discussion_item['note']
position = discussion_item['position']
author_name = note['author']['name']
response_text += f"**💬 Comment by {author_name}:**\n"
response_text += f"{note['body']}\n"
if position.get('new_path'):
line_info = position.get('new_line', 'N/A')
response_text += (
f"*On file: {position['new_path']} "
f"(line {line_info})*\n"
)
created_at = format_date(note['created_at'])
response_text += f"*Posted: {created_at}*\n"
response_text += f"*Discussion ID: {discussion_id}*\n\n"
response_text += "---\n\n"
return [TextContent(type="text", text=response_text)]
except Exception as e:
logging.error(f"Error in get_commit_discussions: {str(e)}")
return [TextContent(
type="text",
text=f"Error retrieving commit discussions: {str(e)}"
)]
```
--------------------------------------------------------------------------------
/tools/reply_to_review_comment.py:
--------------------------------------------------------------------------------
```python
import logging
from mcp.types import TextContent
from gitlab_api import (
reply_to_merge_request_discussion,
create_merge_request_discussion,
resolve_merge_request_discussion
)
async def reply_to_review_comment(
gitlab_url, project_id, access_token, args
):
"""Reply to a specific discussion thread in a merge request review"""
logging.info(f"reply_to_review_comment called with args: {args}")
mr_iid = args["merge_request_iid"]
discussion_id = args["discussion_id"]
reply_body = args["body"]
try:
status, response_data, error_text = await reply_to_merge_request_discussion(
gitlab_url, project_id, access_token, mr_iid, discussion_id, reply_body
)
if status == 201:
author_name = response_data.get('author', {}).get('name', 'Unknown')
note_id = response_data.get('id', 'unknown')
result = "✅ **Reply posted successfully!**\n\n"
result += f"**Merge Request**: !{mr_iid}\n"
result += f"**Discussion ID**: `{discussion_id}`\n"
result += f"**Note ID**: `{note_id}`\n"
result += f"**Author**: {author_name}\n"
reply_preview = reply_body[:100] + ('...' if len(reply_body) > 100 else '')
result += f"**Reply**: {reply_preview}\n"
return [TextContent(type="text", text=result)]
else:
error_msg = "❌ **Error posting reply**\n\n"
error_msg += f"**Status**: {status}\n"
error_msg += f"**Error**: {error_text}\n"
error_msg += f"**MR**: !{mr_iid}\n"
error_msg += f"**Discussion**: {discussion_id}\n"
return [TextContent(type="text", text=error_msg)]
except Exception as e:
logging.error(f"Unexpected error in reply_to_review_comment: {e}")
error_result = f"❌ **Unexpected error**\n\n"
error_result += f"**Error**: {str(e)}\n"
error_result += f"**MR**: !{mr_iid}\n"
error_result += f"**Discussion**: {discussion_id}\n"
return [TextContent(type="text", text=error_result)]
async def create_review_comment(
gitlab_url, project_id, access_token, args
):
"""Create a new discussion thread in a merge request review"""
logging.info(f"create_review_comment called with args: {args}")
mr_iid = args["merge_request_iid"]
comment_body = args["body"]
try:
status, response_data, error_text = await create_merge_request_discussion(
gitlab_url, project_id, access_token, mr_iid, comment_body
)
if status == 201:
author_name = response_data.get('author', {}).get('name', 'Unknown')
discussion_id = response_data.get('id', 'unknown')
result = f"✅ **New discussion created!**\n\n"
result += f"**Merge Request**: !{mr_iid}\n"
result += f"**Discussion ID**: `{discussion_id}`\n"
result += f"**Author**: {author_name}\n"
result += f"**Comment**: {comment_body[:100]}{'...' if len(comment_body) > 100 else ''}\n"
return [TextContent(type="text", text=result)]
else:
error_msg = f"❌ **Error creating discussion**\n\n"
error_msg += f"**Status**: {status}\n"
error_msg += f"**Error**: {error_text}\n"
error_msg += f"**MR**: !{mr_iid}\n"
return [TextContent(type="text", text=error_msg)]
except Exception as e:
logging.error(f"Unexpected error in create_review_comment: {e}")
error_result = f"❌ **Unexpected error**\n\n"
error_result += f"**Error**: {str(e)}\n"
error_result += f"**MR**: !{mr_iid}\n"
return [TextContent(type="text", text=error_result)]
async def resolve_review_discussion(
gitlab_url, project_id, access_token, args
):
"""Resolve or unresolve a discussion thread in a merge request review"""
logging.info(f"resolve_review_discussion called with args: {args}")
mr_iid = args["merge_request_iid"]
discussion_id = args["discussion_id"]
resolved = args.get("resolved", True)
try:
status, response_data, error_text = await resolve_merge_request_discussion(
gitlab_url, project_id, access_token, mr_iid, discussion_id, resolved
)
if status == 200:
action = "resolved" if resolved else "reopened"
result = f"✅ **Discussion {action}!**\n\n"
result += f"**Merge Request**: !{mr_iid}\n"
result += f"**Discussion ID**: `{discussion_id}`\n"
result += f"**Status**: {'✅ Resolved' if resolved else '🔄 Reopened'}\n"
return [TextContent(type="text", text=result)]
else:
error_msg = f"❌ **Error {action} discussion**\n\n"
error_msg += f"**Status**: {status}\n"
error_msg += f"**Error**: {error_text}\n"
error_msg += f"**MR**: !{mr_iid}\n"
error_msg += f"**Discussion**: {discussion_id}\n"
return [TextContent(type="text", text=error_msg)]
except Exception as e:
logging.error(f"Unexpected error in resolve_review_discussion: {e}")
error_result = f"❌ **Unexpected error**\n\n"
error_result += f"**Error**: {str(e)}\n"
error_result += f"**MR**: !{mr_iid}\n"
error_result += f"**Discussion**: {discussion_id}\n"
return [TextContent(type="text", text=error_result)]
```
--------------------------------------------------------------------------------
/tools/get_branch_merge_requests.py:
--------------------------------------------------------------------------------
```python
import asyncio
from mcp.types import TextContent
from gitlab_api import (
get_branch_merge_requests as api_get_branch_merge_requests,
get_merge_request_pipeline,
get_merge_request_changes
)
from utils import (
format_date,
get_state_explanation,
get_pipeline_status_icon,
analyze_mr_readiness,
get_mr_priority,
calculate_change_stats
)
import logging
async def get_enhanced_mr_data(gitlab_url, project_id, access_token, mr_iid):
"""Get enhanced data for a single MR using parallel API calls"""
try:
pipeline_task = get_merge_request_pipeline(
gitlab_url, project_id, access_token, mr_iid
)
changes_task = get_merge_request_changes(
gitlab_url, project_id, access_token, mr_iid
)
pipeline_result, changes_result = await asyncio.gather(
pipeline_task, changes_task, return_exceptions=True
)
if isinstance(pipeline_result, Exception):
pipeline_data = None
logging.warning(
f"Pipeline fetch failed for MR {mr_iid}: {pipeline_result}"
)
else:
pipeline_status, pipeline_data, _ = pipeline_result
if pipeline_status != 200:
pipeline_data = None
if isinstance(changes_result, Exception):
changes_data = None
logging.warning(
f"Changes fetch failed for MR {mr_iid}: {changes_result}"
)
else:
changes_status, changes_data, _ = changes_result
if changes_status != 200:
changes_data = None
return pipeline_data, changes_data
except Exception as e:
logging.warning(
f"Error fetching enhanced data for MR {mr_iid}: {e}"
)
return None, None
async def get_branch_merge_requests(
gitlab_url, project_id, access_token, args
):
logging.info(f"get_branch_merge_requests called with args: {args}")
branch_name = args["branch_name"]
status, data, error = await api_get_branch_merge_requests(
gitlab_url, project_id, access_token, branch_name
)
if status != 200:
logging.error(
f"Error fetching branch merge requests: {status} - {error}"
)
raise Exception(
f"Error fetching branch merge requests: {status} - {error}"
)
result = f"# 🌿 Merge Requests for branch: **{branch_name}**\n"
result += (
f"*Found {len(data)} merge request"
f"{'s' if len(data) != 1 else ''}*\n\n"
)
if not data:
result += "📭 No merge requests found for this branch.\n"
result += (
"💡 **Tip**: Create a merge request to start the "
"review process.\n"
)
return [TextContent(type="text", text=result)]
enhanced_data_tasks = []
for mr in data:
task = get_enhanced_mr_data(
gitlab_url, project_id, access_token, mr['iid']
)
enhanced_data_tasks.append(task)
try:
enhanced_results = await asyncio.gather(*enhanced_data_tasks)
except Exception as e:
logging.warning(f"Error in parallel enhanced data fetch: {e}")
enhanced_results = [(None, None)] * len(data)
for i, mr in enumerate(data):
pipeline_data, changes_data = enhanced_results[i]
if mr['state'] == 'merged':
state_icon = "✅"
elif mr['state'] == 'opened':
state_icon = "🔄"
else:
state_icon = "❌"
result += f"## {state_icon} !{mr['iid']}: {mr['title']}\n"
author_name = mr['author']['name']
author_username = mr['author']['username']
result += f"**👤 Author**: {author_name} (@{author_username})\n"
result += f"**📊 Status**: {mr['state']} "
result += f"({get_state_explanation(mr['state'])})\n"
priority = get_mr_priority(mr)
readiness = analyze_mr_readiness(mr, pipeline_data)
result += f"**🏷️ Priority**: {priority}\n"
result += f"**🚦 Merge Status**: {readiness}\n"
result += f"**📅 Created**: {format_date(mr['created_at'])}\n"
result += f"**🔄 Updated**: {format_date(mr['updated_at'])}\n"
source_branch = mr['source_branch']
target_branch = mr['target_branch']
result += f"**🌿 Branches**: `{source_branch}` → `{target_branch}`\n"
if pipeline_data:
pipeline_status = pipeline_data.get('status')
pipeline_icon = get_pipeline_status_icon(pipeline_status)
result += f"**🔧 Pipeline**: {pipeline_icon} {pipeline_status}\n"
if pipeline_data.get('web_url'):
result += f" *[View Pipeline]({pipeline_data['web_url']})*\n"
elif mr.get('pipeline'):
pipeline_status = mr['pipeline'].get('status')
pipeline_icon = get_pipeline_status_icon(pipeline_status)
result += f"**🔧 Pipeline**: {pipeline_icon} {pipeline_status or 'unknown'}\n"
if changes_data:
change_stats = calculate_change_stats(changes_data)
result += f"**📈 Changes**: {change_stats}\n"
if mr.get('labels'):
labels_str = ', '.join(f"`{label}`" for label in mr['labels'])
result += f"**🏷️ Labels**: {labels_str}\n"
if mr.get('draft') or mr.get('work_in_progress'):
result += "**⚠️ Status**: 🚧 Draft/Work in Progress\n"
if mr.get('has_conflicts'):
result += "**⚠️ Warning**: 🔥 Has merge conflicts\n"
result += f"**🔗 Actions**: [View MR]({mr['web_url']})"
if mr['state'] == 'opened':
result += f" | [Review & Approve]({mr['web_url']})"
result += "\n"
result += "\n---\n\n"
result += "## 📊 Summary\n"
state_counts = {}
for mr in data:
state = mr['state']
state_counts[state] = state_counts.get(state, 0) + 1
result += "**State Breakdown**:\n"
for state, count in state_counts.items():
icon = "✅" if state == 'merged' else "🔄" if state == 'opened' else "❌"
result += f" • {icon} {state.title()}: {count}\n"
result += "\n**🎯 Action Items**:\n"
opened_mrs = [mr for mr in data if mr['state'] == 'opened']
if opened_mrs:
has_conflicts = sum(1 for mr in opened_mrs if mr.get('has_conflicts'))
drafts = sum(1 for mr in opened_mrs if mr.get('draft') or mr.get('work_in_progress'))
if has_conflicts:
result += f" • 🔥 {has_conflicts} MR{'s' if has_conflicts > 1 else ''} with merge conflicts\n"
if drafts:
result += f" • 🚧 {drafts} draft MR{'s' if drafts > 1 else ''} in progress\n"
ready_count = len(opened_mrs) - has_conflicts - drafts
if ready_count > 0:
result += f" • ✅ {ready_count} MR{'s' if ready_count > 1 else ''} ready for review\n"
else:
result += " • 🎉 No open merge requests - branch is clean!\n"
return [TextContent(type="text", text=result)]
```
--------------------------------------------------------------------------------
/tools/get_merge_request_details.py:
--------------------------------------------------------------------------------
```python
import asyncio
from mcp.types import TextContent
from gitlab_api import (
get_merge_request_details as api_get_merge_request_details,
get_merge_request_pipeline,
get_merge_request_changes,
get_merge_request_reviews
)
from utils import (
format_date,
get_state_explanation,
get_pipeline_status_icon,
calculate_change_stats,
analyze_mr_readiness,
get_mr_priority
)
import logging
async def get_merge_request_details(
gitlab_url, project_id, access_token, args
):
logging.info(f"get_merge_request_details called with args: {args}")
mr_iid = args["merge_request_iid"]
tasks = [
api_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),
get_merge_request_reviews(gitlab_url, project_id, access_token, mr_iid)
]
try:
details_result, pipeline_result, changes_result, reviews_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}")
mr_status, mr_data, mr_error = details_result
pipeline_status, pipeline_data, pipeline_error = pipeline_result
changes_status, changes_data, changes_error = changes_result
if mr_status != 200:
logging.error(f"Error fetching merge request details: {mr_status} - {mr_error}")
raise Exception(f"Error fetching merge request details: {mr_status} - {mr_error}")
state_icon = "✅" if mr_data['state'] == 'merged' else "🔄" if mr_data['state'] == 'opened' else "❌"
result = f"# {state_icon} Merge Request !{mr_data['iid']}: {mr_data['title']}\n\n"
result += "## 📋 Overview\n"
result += f"**👤 Author**: {mr_data['author']['name']} (@{mr_data['author']['username']})\n"
result += f"**📊 Status**: {mr_data['state']} ({get_state_explanation(mr_data['state'])})\n"
result += f"**🏷️ Priority**: {get_mr_priority(mr_data)}\n"
result += f"**📅 Created**: {format_date(mr_data['created_at'])}\n"
result += f"**🔄 Updated**: {format_date(mr_data['updated_at'])}\n"
result += f"**🌿 Branches**: `{mr_data['source_branch']}` → `{mr_data['target_branch']}`\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 pipeline_data.get('web_url'):
result += f" *[View Pipeline]({pipeline_data['web_url']})*\n"
elif mr_data.get('pipeline'):
pipeline_status = mr_data['pipeline'].get('status')
pipeline_icon = get_pipeline_status_icon(pipeline_status)
result += f"**🔧 Pipeline**: {pipeline_icon} {pipeline_status or 'unknown'}\n"
if changes_status == 200:
change_stats = calculate_change_stats(changes_data)
result += f"**📈 Changes**: {change_stats}\n"
readiness = analyze_mr_readiness(mr_data, pipeline_data)
result += f"**🚦 Merge Status**: {readiness}\n"
if mr_data.get('labels'):
labels_str = ', '.join(f"`{label}`" for label in mr_data['labels'])
result += f"**🏷️ Labels**: {labels_str}\n"
if mr_data.get('draft') or mr_data.get('work_in_progress'):
result += "**⚠️ Status**: 🚧 Draft/Work in Progress\n"
if mr_data.get('has_conflicts'):
result += "**⚠️ Warning**: 🔥 Has merge conflicts\n"
result += f"**🔗 URL**: {mr_data['web_url']}\n\n"
if mr_data.get('description'):
result += "## 📝 Description\n"
result += f"{mr_data['description']}\n\n"
result += "## 🔧 Technical Details\n"
if mr_data.get('merge_commit_sha'):
result += f"**📦 Merge Commit**: `{mr_data['merge_commit_sha'][:8]}`\n"
if mr_data.get('squash_commit_sha'):
result += f"**🔄 Squash Commit**: `{mr_data['squash_commit_sha'][:8]}`\n"
merge_options = []
if mr_data.get('squash'):
merge_options.append("🔄 Squash commits")
if mr_data.get('remove_source_branch'):
merge_options.append("🗑️ Remove source branch")
if mr_data.get('force_remove_source_branch'):
merge_options.append("🗑️ Force remove source branch")
if merge_options:
result += f"**⚙️ Merge Options**: {', '.join(merge_options)}\n"
if mr_data.get('assignees'):
assignees = ', '.join(f"@{user['username']}" for user in mr_data['assignees'])
result += f"**👥 Assignees**: {assignees}\n"
if mr_data.get('reviewers'):
reviewers = ', '.join(f"@{user['username']}" for user in mr_data['reviewers'])
result += f"**👀 Reviewers**: {reviewers}\n"
if mr_data.get('milestone'):
result += f"**🎯 Milestone**: {mr_data['milestone']['title']}\n"
result += "\n"
if reviews_result and 'discussions' in reviews_result:
discussions_status, discussions, _ = reviews_result['discussions']
approvals_status, approvals, _ = reviews_result['approvals']
result += "## 💬 Reviews Summary\n"
if discussions_status == 200 and discussions:
total_discussions = len(discussions)
resolved_count = sum(1 for d in discussions if d.get('resolved'))
unresolved_count = total_discussions - resolved_count
result += f"**Discussions**: {total_discussions} total, {resolved_count} resolved, {unresolved_count} unresolved\n"
if unresolved_count > 0:
result += f"⚠️ **{unresolved_count} unresolved discussion{'s' if unresolved_count > 1 else ''}**\n"
if approvals_status == 200 and approvals:
approved_by = approvals.get('approved_by', [])
approvals_left = approvals.get('approvals_left', 0)
if approved_by:
result += f"**Approvals**: ✅ {len(approved_by)} approval{'s' if len(approved_by) > 1 else ''}\n"
if approvals_left > 0:
result += f"**Needed**: ⏳ {approvals_left} more approval{'s' if approvals_left > 1 else ''}\n"
result += "\n"
result += "## 📊 Action Items\n"
action_items = []
if mr_data.get('draft') or mr_data.get('work_in_progress'):
action_items.append("🚧 Remove draft/WIP status")
if mr_data.get('has_conflicts'):
action_items.append("⚠️ Resolve merge conflicts")
if pipeline_status == 200 and pipeline_data and pipeline_data.get('status') == 'failed':
action_items.append("❌ Fix failing pipeline")
elif pipeline_status == 200 and pipeline_data and pipeline_data.get('status') == 'running':
action_items.append("🔄 Wait for pipeline completion")
if reviews_result and 'discussions' in reviews_result:
discussions_status, discussions, _ = reviews_result['discussions']
approvals_status, approvals, _ = reviews_result['approvals']
if discussions_status == 200 and 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_status == 200 and 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 mr_data['state'] == 'opened' and not action_items:
action_items.append("✅ Ready to merge!")
if action_items:
for item in action_items:
result += f"• {item}\n"
else:
result += "✅ No action items identified\n"
result += "\n## 🚀 Quick Actions\n"
if mr_data['state'] == 'opened':
result += f"• [📝 Edit MR]({mr_data['web_url']}/edit)\n"
result += f"• [💬 Add Comment]({mr_data['web_url']}#note_form)\n"
result += f"• [🔄 View Changes]({mr_data['web_url']}/diffs)\n"
if pipeline_data and pipeline_data.get('web_url'):
result += f"• [🔧 View Pipeline]({pipeline_data['web_url']})\n"
return [TextContent(type="text", text=result)]
```
--------------------------------------------------------------------------------
/tools/get_merge_request_reviews.py:
--------------------------------------------------------------------------------
```python
import asyncio
from gitlab_api import (
get_merge_request_reviews as api_get_merge_request_reviews,
get_merge_request_details,
get_merge_request_pipeline,
get_merge_request_changes
)
from utils import (
format_date,
get_state_explanation,
get_pipeline_status_icon,
analyze_mr_readiness,
get_mr_priority,
calculate_change_stats
)
from mcp.types import TextContent
import logging
def get_review_type_icon(note):
"""Get appropriate icon for review type"""
if note.get('resolvable'):
return '💬'
elif note.get('position'):
return '📝'
elif 'approved' in note.get('body', '').lower():
return '✅'
elif any(word in note.get('body', '').lower()
for word in ['reject', 'needs work', 'changes requested']):
return '❌'
else:
return '💭'
def get_approval_summary(approvals):
"""Generate enhanced approval summary"""
if not approvals:
return "## 👥 Approvals\n❓ No approval information available\n\n"
result = "## 👥 Approvals\n"
approved_by = approvals.get('approved_by', [])
approvals_required = approvals.get('approvals_required', 0)
approvals_left = approvals.get('approvals_left', 0)
if approved_by:
result += f"**✅ Approved by ({len(approved_by)} reviewer"
result += f"{'s' if len(approved_by) > 1 else ''}):**\n"
for approval in approved_by:
user = approval['user']
result += f" • **{user['name']}** (@{user['username']})\n"
result += "\n"
if approvals_required > 0:
if approvals_left == 0:
status = "✅ Approval requirements met"
else:
plural = 's' if approvals_left > 1 else ''
status = f"⏳ {approvals_left} approval{plural} needed"
result += f"**Status**: {status}\n"
received_count = len(approved_by)
result += f"**Required**: {approvals_required} | **Received**: {received_count}\n\n"
elif not approved_by:
result += "📝 No approvals yet\n\n"
return result
def get_discussion_summary(discussions):
"""Generate enhanced discussion summary with counts and status"""
if not discussions:
return "## 💬 Discussions\n❓ No discussion information available\n\n"
total_discussions = len(discussions)
resolved_count = sum(1 for d in discussions if d.get('resolved'))
unresolved_count = total_discussions - resolved_count
result = "## 💬 Discussions & Reviews\n"
result += f"**Total**: {total_discussions} | **Resolved**: {resolved_count} | **Unresolved**: {unresolved_count}\n\n"
if unresolved_count > 0:
result += f"⚠️ **{unresolved_count} unresolved discussion{'s' if unresolved_count > 1 else ''}** - action needed\n\n"
elif total_discussions > 0:
result += "✅ All discussions resolved\n\n"
return result
def format_discussion_thread(discussion):
"""Format a single discussion thread with enhanced formatting"""
if not discussion.get('notes'):
return ""
result = ""
thread_resolved = discussion.get('resolved', False)
thread_icon = "✅" if thread_resolved else "🟡"
discussion_id = discussion.get('id', 'unknown')
result += f"### {thread_icon} Discussion Thread\n"
result += f"**Discussion ID**: `{discussion_id}`\n"
if thread_resolved:
result += "*Resolved*\n"
else:
result += "*Unresolved*\n"
for note in discussion['notes']:
if note.get('system'):
continue
author_name = note['author']['name']
author_username = note['author']['username']
note_icon = get_review_type_icon(note)
note_id = note.get('id', 'unknown')
result += f"\n{note_icon} **{author_name}** (@{author_username})\n"
timestamp = format_date(note['created_at'])
result += f"*{timestamp}* | Note ID: `{note_id}`\n"
if note.get('position'):
pos = note['position']
if pos.get('new_path'):
result += f"📁 **File**: `{pos['new_path']}`\n"
if pos.get('new_line'):
result += f"📍 **Line**: {pos['new_line']}\n"
body = note.get('body', '').strip()
if body:
result += f"\n{body}\n"
result += "\n---\n"
return result + "\n"
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 += f"## 📋 Merge Request Overview\n"
result += f"**Title**: {mr_details.get('title', 'N/A')}\n"
result += f"**Status**: {mr_details.get('state', 'N/A')} ({get_state_explanation(mr_details.get('state', 'N/A'))})\n"
result += f"**Author**: {mr_details.get('author', {}).get('name', 'N/A')} (@{mr_details.get('author', {}).get('username', 'N/A')})\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)]
```
--------------------------------------------------------------------------------
/tools/list_merge_requests.py:
--------------------------------------------------------------------------------
```python
import asyncio
from mcp.types import TextContent
from gitlab_api import (
get_merge_requests,
get_merge_request_pipeline,
get_merge_request_changes
)
from utils import (
format_date,
get_state_explanation,
get_pipeline_status_icon,
analyze_mr_readiness,
get_mr_priority,
calculate_change_stats
)
import logging
async def get_enhanced_mr_data(gitlab_url, project_id, access_token, mr_iid):
"""Get enhanced data for a single MR using parallel API calls"""
try:
pipeline_task = get_merge_request_pipeline(
gitlab_url, project_id, access_token, mr_iid
)
changes_task = get_merge_request_changes(
gitlab_url, project_id, access_token, mr_iid
)
pipeline_result, changes_result = await asyncio.gather(
pipeline_task, changes_task, return_exceptions=True
)
if isinstance(pipeline_result, Exception):
pipeline_data = None
logging.warning(f"Pipeline fetch failed for MR {mr_iid}: {pipeline_result}")
else:
pipeline_status, pipeline_data, _ = pipeline_result
if pipeline_status != 200:
pipeline_data = None
if isinstance(changes_result, Exception):
changes_data = None
logging.warning(f"Changes fetch failed for MR {mr_iid}: {changes_result}")
else:
changes_status, changes_data, _ = changes_result
if changes_status != 200:
changes_data = None
return pipeline_data, changes_data
except Exception as e:
logging.warning(f"Error fetching enhanced data for MR {mr_iid}: {e}")
return None, None
async def list_merge_requests(gitlab_url, project_id, access_token, args):
logging.info(f"list_merge_requests called with args: {args}")
state = args.get("state", "opened")
target_branch = args.get("target_branch")
limit = args.get("limit", 10)
params = {
"state": state,
"per_page": limit,
"order_by": "updated_at",
"sort": "desc"
}
if target_branch:
params["target_branch"] = target_branch
status, data, error = await get_merge_requests(
gitlab_url, project_id, access_token, params
)
if status != 200:
logging.error(f"Error listing merge requests: {status} - {error}")
raise Exception(f"Error listing merge requests: {status} - {error}")
state_filter = f" ({state})" if state != "all" else ""
result = f"# 📋 Merge Requests{state_filter}\n"
result += f"*Found {len(data)} merge request{'s' if len(data) != 1 else ''}*\n\n"
if not data:
result += "📭 No merge requests found.\n"
if state == "opened":
result += "💡 **Tip**: Create a merge request to start the development workflow.\n"
return [TextContent(type="text", text=result)]
enhanced_data_tasks = []
for mr in data[:5]:
task = get_enhanced_mr_data(
gitlab_url, project_id, access_token, mr['iid']
)
enhanced_data_tasks.append(task)
try:
enhanced_results = await asyncio.gather(*enhanced_data_tasks)
except Exception as e:
logging.warning(f"Error in parallel enhanced data fetch: {e}")
enhanced_results = [(None, None)] * len(data[:5])
for i, mr in enumerate(data):
if i < len(enhanced_results):
pipeline_data, changes_data = enhanced_results[i]
else:
pipeline_data, changes_data = None, None
if mr['state'] == 'merged':
state_icon = "✅"
elif mr['state'] == 'opened':
state_icon = "🔄"
else:
state_icon = "❌"
result += f"## {state_icon} !{mr['iid']}: {mr['title']}\n"
author_name = mr['author']['name']
author_username = mr['author']['username']
result += f"**👤 Author**: {author_name} (@{author_username})\n"
result += f"**📊 Status**: {mr['state']} ({get_state_explanation(mr['state'])})\n"
priority = get_mr_priority(mr)
readiness = analyze_mr_readiness(mr, pipeline_data)
result += f"**🏷️ Priority**: {priority}\n"
result += f"**🚦 Merge Status**: {readiness}\n"
result += f"**📅 Created**: {format_date(mr['created_at'])}\n"
result += f"**🔄 Updated**: {format_date(mr['updated_at'])}\n"
source_branch = mr['source_branch']
target_branch = mr['target_branch']
result += f"**🌿 Branches**: `{source_branch}` → `{target_branch}`\n"
if pipeline_data:
pipeline_status = pipeline_data.get('status')
pipeline_icon = get_pipeline_status_icon(pipeline_status)
result += f"**🔧 Pipeline**: {pipeline_icon} {pipeline_status}\n"
if pipeline_data.get('web_url'):
result += f" *[View Pipeline]({pipeline_data['web_url']})*\n"
elif mr.get('pipeline'):
pipeline_status = mr['pipeline'].get('status')
pipeline_icon = get_pipeline_status_icon(pipeline_status)
result += f"**🔧 Pipeline**: {pipeline_icon} {pipeline_status or 'unknown'}\n"
if changes_data:
change_stats = calculate_change_stats(changes_data)
result += f"**📈 Changes**: {change_stats}\n"
if mr.get('labels'):
labels_str = ', '.join(f"`{label}`" for label in mr['labels'])
result += f"**🏷️ Labels**: {labels_str}\n"
if mr.get('draft') or mr.get('work_in_progress'):
result += "**⚠️ Status**: 🚧 Draft/Work in Progress\n"
if mr.get('has_conflicts'):
result += "**⚠️ Warning**: 🔥 Has merge conflicts\n"
result += f"**🔗 Actions**: [View MR]({mr['web_url']})"
if mr['state'] == 'opened':
result += f" | [Review]({mr['web_url']})"
result += "\n\n"
result += "## 📊 Summary\n"
state_counts = {}
for mr in data:
state = mr['state']
state_counts[state] = state_counts.get(state, 0) + 1
result += "**State Breakdown**:\n"
for state, count in state_counts.items():
if state == 'merged':
icon = "✅"
elif state == 'opened':
icon = "🔄"
else:
icon = "❌"
result += f" • {icon} {state.title()}: {count}\n"
priority_counts = {}
for mr in data:
priority = get_mr_priority(mr)
priority_counts[priority] = priority_counts.get(priority, 0) + 1
if len(priority_counts) > 1:
result += "\n**Priority Breakdown**:\n"
for priority, count in priority_counts.items():
result += f" • {priority}: {count}\n"
opened_mrs = [mr for mr in data if mr['state'] == 'opened']
if opened_mrs:
result += "\n**🎯 Action Items**:\n"
has_conflicts = sum(1 for mr in opened_mrs if mr.get('has_conflicts'))
drafts = sum(1 for mr in opened_mrs if mr.get('draft') or mr.get('work_in_progress'))
failed_pipelines = 0
for i, mr in enumerate(opened_mrs):
if i < len(enhanced_results):
pipeline_data, _ = enhanced_results[i]
if pipeline_data and pipeline_data.get('status') == 'failed':
failed_pipelines += 1
if has_conflicts:
result += f" • 🔥 {has_conflicts} MR{'s' if has_conflicts > 1 else ''} with merge conflicts\n"
if drafts:
result += f" • 🚧 {drafts} draft MR{'s' if drafts > 1 else ''} in progress\n"
if failed_pipelines:
result += f" • ❌ {failed_pipelines} MR{'s' if failed_pipelines > 1 else ''} with failed pipelines\n"
ready_count = len(opened_mrs) - has_conflicts - drafts - failed_pipelines
if ready_count > 0:
result += f" • ✅ {ready_count} MR{'s' if ready_count > 1 else ''} ready for review\n"
result += "\n**📋 Next Steps**:\n"
if has_conflicts:
result += " • 🔧 Resolve merge conflicts to unblock development\n"
if failed_pipelines:
result += " • 🔧 Fix failing pipelines to ensure quality\n"
if ready_count > 0:
result += " • 👀 Review and approve ready merge requests\n"
else:
result += "\n**🎯 Action Items**:\n"
if state == "opened":
result += " • 🎉 No open merge requests - ready for new features!\n"
else:
result += " • 📊 Consider filtering by 'opened' state to see active work\n"
return [TextContent(type="text", text=result)]
```
--------------------------------------------------------------------------------
/gitlab_api.py:
--------------------------------------------------------------------------------
```python
import aiohttp
def _headers(access_token):
return {"Private-Token": access_token, "Content-Type": "application/json"}
async def get_merge_requests(gitlab_url, project_id, access_token, params):
url = f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests"
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(
url, headers=headers, params=params
) as response:
return (
response.status,
await response.json(),
await response.text()
)
async def get_merge_request_pipeline(
gitlab_url, project_id, access_token, mr_iid
):
"""Get the latest pipeline for a merge request"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/"
f"merge_requests/{mr_iid}/pipelines"
)
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
params = {"per_page": 1}
async with session.get(url, headers=headers, params=params) as response:
data = await response.json()
return (
response.status,
data[0] if data else None,
await response.text()
)
async def get_merge_request_changes(
gitlab_url, project_id, access_token, mr_iid
):
"""Get changes/diff stats for a merge request"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/"
f"merge_requests/{mr_iid}/changes"
)
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
return (
response.status,
await response.json(),
await response.text()
)
async def get_project_info(gitlab_url, project_id, access_token):
"""Get project information to check for merge conflicts"""
url = f"{gitlab_url}/api/v4/projects/{project_id}"
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
return (
response.status,
await response.json(),
await response.text()
)
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
),
}
async def get_merge_request_details(
gitlab_url, project_id, access_token, mr_iid
):
url = f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests/{mr_iid}"
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
return (
response.status,
await response.json(),
await response.text()
)
async def create_merge_request_discussion(
gitlab_url, project_id, access_token, mr_iid, body
):
"""Create a new discussion/comment on a merge request"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests/"
f"{mr_iid}/discussions"
)
headers = _headers(access_token)
data = {"body": body}
async with aiohttp.ClientSession() as session:
async with session.post(
url, headers=headers, json=data
) as response:
json_data = (
await response.json()
if response.content_type == 'application/json' else {}
)
return (
response.status,
json_data,
await response.text()
)
async def reply_to_merge_request_discussion(
gitlab_url, project_id, access_token, mr_iid, discussion_id, body
):
"""Reply to an existing discussion on a merge request"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests/"
f"{mr_iid}/discussions/{discussion_id}/notes"
)
headers = _headers(access_token)
data = {"body": body}
async with aiohttp.ClientSession() as session:
async with session.post(
url, headers=headers, json=data
) as response:
json_data = (
await response.json()
if response.content_type == 'application/json' else {}
)
return (
response.status,
json_data,
await response.text()
)
async def resolve_merge_request_discussion(
gitlab_url, project_id, access_token, mr_iid, discussion_id, resolved
):
"""Resolve or unresolve a discussion on a merge request"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests/"
f"{mr_iid}/discussions/{discussion_id}"
)
headers = _headers(access_token)
data = {"resolved": resolved}
async with aiohttp.ClientSession() as session:
async with session.put(
url, headers=headers, json=data
) as response:
json_data = (
await response.json()
if response.content_type == 'application/json' else {}
)
return (
response.status,
json_data,
await response.text()
)
async def get_branch_merge_requests(
gitlab_url, project_id, access_token, branch_name
):
"""Get merge requests for a specific branch"""
params = {
"source_branch": branch_name,
"state": "all",
"per_page": 100
}
url = f"{gitlab_url}/api/v4/projects/{project_id}/merge_requests"
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(
url, headers=headers, params=params
) as response:
return (
response.status,
await response.json(),
await response.text()
)
async def get_merge_request_commits(
gitlab_url, project_id, access_token, mr_iid
):
"""Get all commits in a merge request (handles pagination)"""
base_url = (
f"{gitlab_url}/api/v4/projects/{project_id}/"
f"merge_requests/{mr_iid}/commits"
)
headers = _headers(access_token)
all_commits = []
page = 1
per_page = 100 # Maximum allowed per page
async with aiohttp.ClientSession() as session:
while True:
params = {"page": page, "per_page": per_page}
async with session.get(
base_url, headers=headers, params=params
) as response:
if response.status != 200:
return (
response.status,
await response.json(),
await response.text()
)
page_data = await response.json()
if not page_data: # No more results
break
all_commits.extend(page_data)
# If we got fewer results than per_page, we're done
if len(page_data) < per_page:
break
page += 1
return (200, all_commits, "Success")
async def get_commit_comments(
gitlab_url, project_id, access_token, commit_sha
):
"""Get simple comments for a specific commit"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/"
f"repository/commits/{commit_sha}/comments"
)
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
return (
response.status,
await response.json(),
await response.text()
)
async def get_commit_discussions(
gitlab_url, project_id, access_token, commit_sha
):
"""Get discussions/comments for a specific commit"""
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/"
f"repository/commits/{commit_sha}/discussions"
)
headers = _headers(access_token)
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
return (
response.status,
await response.json(),
await response.text()
)
async def get_commit_all_comments_and_discussions(
gitlab_url, project_id, access_token, commit_sha
):
"""Get both comments and discussions for a commit, combining them"""
discussions_result = await get_commit_discussions(
gitlab_url, project_id, access_token, commit_sha
)
discussions_status, discussions_data, discussions_error = discussions_result
comments_result = await get_commit_comments(
gitlab_url, project_id, access_token, commit_sha
)
comments_status, comments_data, comments_error = (
comments_result
)
combined_data = []
if discussions_status == 200 and discussions_data:
combined_data.extend(discussions_data)
if comments_status == 200 and comments_data:
for comment in comments_data:
discussion_format = {
"id": f"comment_{comment.get('id', 'unknown')}",
"individual_note": True,
"notes": [{
"id": comment.get("id"),
"body": comment.get("note", ""),
"author": comment.get("author", {}),
"created_at": comment.get("created_at"),
"updated_at": comment.get("created_at"),
"system": False,
"noteable_type": "Commit",
"noteable_id": commit_sha,
"resolvable": False,
"position": {
"new_path": comment.get("path"),
"new_line": comment.get("line"),
"line_type": comment.get("line_type")
} if comment.get("path") else None
}]
}
combined_data.append(discussion_format)
if combined_data:
return (200, combined_data, "Success")
elif discussions_status == 200 or comments_status == 200:
return (200, [], "No comments or discussions found")
else:
return discussions_result
async def get_merge_request_discussions_paginated(
gitlab_url, project_id, access_token, mr_iid
):
"""Get all discussions from a merge request with pagination"""
all_discussions = []
page = 1
per_page = 100 # Maximum allowed per page
async with aiohttp.ClientSession() as session:
headers = _headers(access_token)
while True:
url = (
f"{gitlab_url}/api/v4/projects/{project_id}/"
f"merge_requests/{mr_iid}/discussions"
)
params = {"page": page, "per_page": per_page}
async with session.get(url, headers=headers, params=params) as response:
if response.status != 200:
return (
response.status,
await response.json(),
await response.text()
)
discussions = await response.json()
if not discussions: # No more results
break
all_discussions.extend(discussions)
link_header = response.headers.get('Link', '')
if 'rel="next"' not in link_header:
break
page += 1
return (200, all_discussions, "Success")
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import asyncio
import logging
from typing import Any, Dict, List
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
INTERNAL_ERROR,
INVALID_PARAMS,
METHOD_NOT_FOUND,
JSONRPCError
)
from logging_config import configure_logging
from config import get_gitlab_config
from tools import (
list_merge_requests,
get_merge_request_reviews,
get_merge_request_details,
get_branch_merge_requests,
reply_to_review_comment,
create_review_comment,
resolve_review_discussion,
get_commit_discussions
)
class GitLabMCPServer:
def __init__(self):
configure_logging()
logging.info("Initializing GitLabMCPServer")
self.config = get_gitlab_config()
self.server = Server(self.config['server_name'])
self.setup_handlers()
def setup_handlers(self):
@self.server.list_tools()
async def list_tools() -> List[Tool]:
logging.info("list_tools called")
tools = [
Tool(
name="list_merge_requests",
description="List merge requests for the GitLab project",
inputSchema={
"type": "object",
"properties": {
"state": {
"type": "string",
"enum": ["opened", "closed", "merged", "all"],
"default": "opened",
"description": "Filter by merge request state"
},
"target_branch": {
"type": "string",
"description": (
"Filter by target branch (optional)"
)
},
"limit": {
"type": "integer",
"default": 10,
"minimum": 1,
"maximum": 100,
"description": "Maximum number of results"
}
},
"additionalProperties": False
}
),
Tool(
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
}
),
Tool(
name="get_merge_request_details",
description=(
"Get detailed information about 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
}
),
Tool(
name="get_branch_merge_requests",
description=(
"Get all merge requests for a specific branch"
),
inputSchema={
"type": "object",
"properties": {
"branch_name": {
"type": "string",
"description": "Name of the branch"
}
},
"required": ["branch_name"],
"additionalProperties": False
}
),
Tool(
name="reply_to_review_comment",
description=(
"Reply to a specific discussion thread in a "
"merge request review"
),
inputSchema={
"type": "object",
"properties": {
"merge_request_iid": {
"type": "integer",
"minimum": 1,
"description": (
"Internal ID of the merge request"
)
},
"discussion_id": {
"type": "string",
"description": (
"ID of the discussion thread to reply to"
)
},
"body": {
"type": "string",
"description": "Content of the reply comment"
}
},
"required": [
"merge_request_iid", "discussion_id", "body"
],
"additionalProperties": False
}
),
Tool(
name="create_review_comment",
description=(
"Create a new discussion thread in a "
"merge request review"
),
inputSchema={
"type": "object",
"properties": {
"merge_request_iid": {
"type": "integer",
"minimum": 1,
"description": (
"Internal ID of the merge request"
)
},
"body": {
"type": "string",
"description": (
"Content of the new discussion comment"
)
}
},
"required": ["merge_request_iid", "body"],
"additionalProperties": False
}
),
Tool(
name="resolve_review_discussion",
description=(
"Resolve or unresolve a discussion thread in a "
"merge request review"
),
inputSchema={
"type": "object",
"properties": {
"merge_request_iid": {
"type": "integer",
"minimum": 1,
"description": (
"Internal ID of the merge request"
)
},
"discussion_id": {
"type": "string",
"description": (
"ID of the discussion thread to "
"resolve/unresolve"
)
},
"resolved": {
"type": "boolean",
"default": True,
"description": (
"Whether to resolve (true) or unresolve "
"(false) the discussion"
)
}
},
"required": ["merge_request_iid", "discussion_id"],
"additionalProperties": False
}
),
Tool(
name="get_commit_discussions",
description=(
"Get discussions and comments on commits within 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
}
)
]
tool_names = [t.name for t in tools]
logging.info(f"Returning {len(tools)} tools: {tool_names}")
return tools
@self.server.call_tool()
async def call_tool(
name: str, arguments: Dict[str, Any]
) -> List[TextContent]:
logging.info(
f"call_tool called: {name} with arguments: {arguments}"
)
try:
if name not in [
"list_merge_requests",
"get_merge_request_reviews",
"get_merge_request_details",
"get_branch_merge_requests",
"reply_to_review_comment",
"create_review_comment",
"resolve_review_discussion",
"get_commit_discussions"
]:
logging.warning(f"Unknown tool called: {name}")
raise JSONRPCError(
METHOD_NOT_FOUND,
f"Unknown tool: {name}"
)
if name == "list_merge_requests":
return await list_merge_requests(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
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
)
elif name == "get_merge_request_details":
return await get_merge_request_details(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
elif name == "get_branch_merge_requests":
return await get_branch_merge_requests(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
elif name == "reply_to_review_comment":
return await reply_to_review_comment(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
elif name == "create_review_comment":
return await create_review_comment(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
elif name == "resolve_review_discussion":
return await resolve_review_discussion(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
elif name == "get_commit_discussions":
return await get_commit_discussions(
self.config['gitlab_url'],
self.config['project_id'],
self.config['access_token'],
arguments
)
except JSONRPCError:
raise
except ValueError as e:
logging.error(f"Validation error in {name}: {e}")
raise JSONRPCError(
INVALID_PARAMS,
f"Invalid parameters: {str(e)}"
)
except Exception as e:
logging.error(
f"Unexpected error in call_tool for {name}: {e}",
exc_info=True
)
raise JSONRPCError(
INTERNAL_ERROR,
f"Internal server error: {str(e)}"
)
async def run(self):
logging.info("Starting MCP stdio server")
try:
async with stdio_server() as (read_stream, write_stream):
logging.info("stdio_server context entered successfully")
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name=self.config['server_name'],
server_version=self.config['server_version'],
capabilities={
"tools": {},
"logging": {}
}
)
)
except Exception as e:
logging.error(f"Error in stdio_server: {e}", exc_info=True)
raise
async def main():
try:
logging.info("Starting main function")
server = GitLabMCPServer()
logging.info("GitLabMCPServer created successfully")
await server.run()
except Exception as e:
logging.error(f"Error starting server: {e}", exc_info=True)
print(f"Error starting server: {e}")
return 1
if __name__ == "__main__":
asyncio.run(main())
```