gitlab_mcp_server.py•26.8 kB
import os
from typing import List, Optional
from fastmcp import FastMCP
from pydantic import BaseModel
import gitlab
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Initialize FastMCP server
mcp = FastMCP("GitLab MCP Server")
# Initialize GitLab client
def get_gitlab_client():
url = os.getenv("GITLAB_URL", "https://gitlab.com")
token = os.getenv("GITLAB_TOKEN")
if not token:
raise ValueError("GITLAB_TOKEN environment variable is not set")
return gitlab.Gitlab(url=url, private_token=token)
# Models for request/response
class FileContent(BaseModel):
file_path: str
content: str
class FileAction(BaseModel):
file_path: str
content: str
action: str = "create" # create, update, delete
class ProjectCreate(BaseModel):
name: str
description: Optional[str] = None
visibility: str = "private"
initialize_with_readme: bool = True
class MergeRequestCreate(BaseModel):
title: str
source_branch: str
target_branch: str
description: Optional[str] = None
draft: bool = False
class LineComment(BaseModel):
line: int
content: str
path: str
position_type: str = "text"
# GitLab tools
@mcp.tool()
def create_repository(name: str, description: Optional[str] = None, visibility: str = "private", initialize_with_readme: bool = True) -> dict:
"""Create a new GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.create({
'name': name,
'description': description,
'visibility': visibility,
'initialize_with_readme': initialize_with_readme
})
return {
"status": "success",
"message": f"Create {name} project successfully",
"id": project.id,
"name": project.name,
"web_url": project.web_url
}
except Exception as e:
return {
"status": "error",
"message": f"Create {name} project failed: {str(e)}"
}
@mcp.tool()
def fork_repository(project_id: str, name: Optional[str] = None, path: Optional[str] = None, namespace_id: Optional[str] = None) -> dict:
"""Fork a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Fork {project_id} project failed: {str(e)}"
}
try:
# fork params
fork_params = {}
if name is not None:
fork_params['name'] = name
if path is not None:
fork_params['path'] = path
if namespace_id is not None:
fork_params['namespace_id'] = namespace_id
# Create fork
forked_project = project.forks.create(fork_params)
return {
"status": "success",
"message": f"Fork {project_id} project successfully",
"id": forked_project.id,
"name": forked_project.name,
"web_url": forked_project.web_url,
"forked_from_id": project_id
}
except Exception as e:
return {
"status": "error",
"message": f"Fork {project_id} project failed: {str(e)}"
}
@mcp.tool()
def delete_repository(project_id: str) -> dict:
"""Delete a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {project_id} project failed: {str(e)}"
}
try:
# Delete the project
project.delete()
return {
"status": "success",
"message": f"project {project_id} successfully deleted"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {project_id} project failed: {str(e)}"
}
@mcp.tool()
def search_repositories(search: str, page: int = 1, per_page: int = 20) -> dict:
"""Search for GitLab projects"""
try:
gl = get_gitlab_client()
projects = gl.projects.list(search=search, page=page, per_page=per_page)
return {
"status": "success",
"message": f"Search {search} project successfully",
"projects": [{"id": p.id, "name": p.name, "path": p.path_with_namespace} for p in projects]
}
except Exception as e:
return {
"status": "error",
"message": f"Search {search} project failed: {str(e)}"
}
@mcp.tool()
def create_or_update_file(project_id: str, file_path: str, content: str, commit_message: str, branch: str, ref_branch: str = "master") -> dict:
"""Create or update a file in a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create or update file in {project_id} failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
# check if branch exists, if not create it
try:
# get the branch, if it exists return existing branch info
project.branches.get(branch)
except gitlab.exceptions.GitlabGetError:
# branch not found, create new branch
# ref_branch is the base branch to create new branch from
try:
project.branches.create({
'branch': branch,
'ref': ref_branch
})
except Exception as e:
return {
"status": "error",
"message": f"Create {branch} branch base {ref_branch} branch in {project_id} project failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Create or update file in {branch} branch in {project_id} failed: {str(e)}"
}
try:
# Try to update existing file
f = project.files.get(file_path=file_path, ref=branch)
f.content = content
f.save(branch=branch, commit_message=commit_message)
return {
"status": "success",
"message": f"File {file_path} existing in {branch} branch in {project_id}",
"file_path": file_path
}
except gitlab.exceptions.GitlabGetError:
# Create new file
project.files.create({
'file_path': file_path,
'branch': branch,
'content': content,
'commit_message': commit_message
})
return {
"status": "success",
"message": f"File {file_path} created in {branch} branch in {project_id}",
"file_path": file_path
}
except Exception as e:
return {
"status": "error",
"message": f"Create or update file in {branch} branch in {project_id} failed: {str(e)}"
}
@mcp.tool()
def push_files(project_id: str, files: List[dict], commit_message: str, branch: str, ref_branch: str = "master") -> dict:
"""Push multiple files to a GitLab project in a single commit"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
# check if branch exists, if not create it
try:
# Try to get the branch, if it exists return existing branch info
project.branches.get(branch)
except gitlab.exceptions.GitlabGetError:
# If branch doesn't exist, create it base on ref_branch
try:
project.branches.create({
'branch': branch,
'ref': ref_branch
})
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
# Prepare actions for commit
actions = []
for file_data in files:
file_path = file_data.get('file_path')
content = file_data.get('content', '')
action = file_data.get('action', 'create')
if action == 'delete':
actions.append({
'action': 'delete',
'file_path': file_path
})
else:
# For create/update, first check if file exists to determine action
try:
project.files.get(file_path=file_path, ref=branch)
# File exists, so we update
actions.append({
'action': 'update',
'file_path': file_path,
'content': content
})
except gitlab.exceptions.GitlabGetError:
# File doesn't exist, so we create
actions.append({
'action': 'create',
'file_path': file_path,
'content': content
})
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
# Create commit with all actions
try:
commit = project.commits.create({
'branch': branch,
'commit_message': commit_message,
'actions': actions
})
return {
"status": "success",
"message": f"Files pushed to {branch} branch of {project_id} project successfully",
"commit_id": commit.id,
"commit_short_id": commit.short_id,
"files_count": len(actions)
}
except Exception as e:
return {
"status": "error",
"message": f"Push files to {project_id} failed: {str(e)}"
}
@mcp.tool()
def get_file_contents(project_id: str, file_path: str, ref: Optional[str] = None) -> dict:
"""Get the contents of a file from a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} file contents failed: {str(e)}"
}
try:
f = project.files.get(file_path=file_path, ref=ref or project.default_branch)
return {
"status": "success",
"message": f"Get {file_path} file contents successfully",
"content": f.decode().decode()
}
except gitlab.exceptions.GitlabGetError:
return {
"error": "error",
"message": f"File {file_path} not found in {project_id}"
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} file contents failed: {str(e)}"
}
@mcp.tool()
def create_issue(project_id: str, title: str, description: Optional[str] = None,
labels: Optional[List[str]] = None) -> dict:
"""Create a new issue in a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {title} issue in {project_id} project failed: {str(e)}"
}
try:
issue = project.issues.create({
'title': title,
'description': description,
'labels': labels or []
})
return {
"status": "success",
"message": f"Create {title} issue in {project_id} project successfully",
"id": issue.iid,
"web_url": issue.web_url,
"state": issue.state
}
except Exception as e:
return {
"status": "error",
"message": f"Create {title} issue in {project_id} project failed: {str(e)}"
}
@mcp.tool()
def get_issues(project_id: str, state: Optional[str] = None, labels: Optional[List[str]] = None, page: int = 1, per_page: int = 20) -> dict:
"""Get all issues from a GitLab project with optional filtering"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} issues failed: {str(e)}"
}
# Prepare query parameters
query_params = {
'page': page,
'per_page': per_page
}
# Add optional filtering conditions
if state is not None:
query_params['state'] = state # opened, closed, all
if labels is not None:
if isinstance(labels, list):
query_params['labels'] = ','.join(labels)
else:
query_params['labels'] = labels
try:
# Get issues list
issues = project.issues.list(**query_params)
# Format return result, referencing create_issue's return format
formatted_issues = []
for issue in issues:
formatted_issues.append({
"id": issue.iid,
"title": issue.title,
"description": getattr(issue, 'description', ''),
"state": issue.state,
"labels": getattr(issue, 'labels', []),
"web_url": issue.web_url,
"author": getattr(issue, 'author', {}).get('name', 'Unknown'),
"created_at": getattr(issue, 'created_at', ''),
"updated_at": getattr(issue, 'updated_at', '')
})
return {
"status": "success",
"message": f"Get issues in {project_id} project successfully",
"issues": formatted_issues,
"total_count": len(formatted_issues),
"page": page,
"per_page": per_page
}
except Exception as e:
return {
"status": "error",
"message": f"Get {project_id} issues failed: {str(e)}"
}
@mcp.tool()
def create_merge_request(project_id: str, title: str, source_branch: str, target_branch: str,
description: Optional[str] = None, draft: bool = False) -> dict:
"""Create a new merge request in a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {project_id} merge request failed: {str(e)}"
}
# Check if merge request already exists
try:
existing_mrs = project.mergerequests.list(
source_branch=source_branch,
target_branch=target_branch,
state='opened'
)
if len(existing_mrs) > 0:
# Return the first matching merge request
existing_mr = existing_mrs[0]
return {
"status": "success",
"message": f"Merge request already exists from {source_branch} to {target_branch} in {project_id} project",
"id": existing_mr.iid,
"web_url": existing_mr.web_url,
"state": existing_mr.state
}
except Exception as e:
return {
"status": "error",
"message": f"merge request in {project_id} project failed: {str(e)}"
}
payload = {
'title': title,
'source_branch': source_branch,
'target_branch': target_branch,
'draft': draft
}
if description is not None:
payload['description'] = description
try:
mr = project.mergerequests.create(payload)
return {
"status": "success",
"message": f"Create merge request in {project_id} project successfully",
"id": mr.iid,
"web_url": mr.web_url,
"state": mr.state
}
except Exception as e:
return {
"status": "error",
"message": f"Create {project_id} merge request failed: {str(e)}"
}
@mcp.tool()
def get_merge_request_diff(project_id: str, merge_request_iid: int) -> dict:
"""Get the diff of a merge request to find valid line positions for comments."""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"get {project_id} merge request diff failed failed: {str(e)}"
}
try:
# get the merge request by merge_request_iid
mr = project.mergerequests.get(merge_request_iid)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"merge_request {merge_request_iid} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"get {merge_request_iid} merge request diff failed failed: {str(e)}"
}
try:
# Get the diff refs
diff_refs = mr.diff_refs
# Fetch the changes
changes = mr.changes()
# Process the changes to include necessary information
diffs = []
for change in changes['changes']:
diff = {
'old_path': change['old_path'],
'new_path': change['new_path'],
'diff_refs': diff_refs,
'diff': change['diff']
}
diffs.append(diff)
return {
"status": "success",
"message": f"Get {merge_request_iid} merge request diff in {project_id} project successfully",
"diffs": diffs,
"merge_request_iid": merge_request_iid,
"project_id": project_id
}
except Exception as e:
return {
"status": "error",
"message": f"get {merge_request_iid} merge request diff failed failed: {str(e)}"
}
@mcp.tool()
def create_branches(project_id: str, branch_name: str, ref_branch: str = "master") -> dict:
"""Create a new branch in a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {branch_name} in {project_id} project failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
try:
# Try to get the branch, if it exists return existing branch info
existing_branch = project.branches.get(branch_name)
return {
"status": "success",
"message": f"Branch {branch_name} already exists in {project_id} project",
"branch_name": existing_branch.name,
"commit": existing_branch.commit
}
except gitlab.exceptions.GitlabGetError:
# Branch doesn't exist, create new branch
try:
new_branch = project.branches.create({
'branch': branch_name,
'ref': ref_branch
})
return {
"status": "success",
"message": f"Create {branch_name} in {project_id} project successfully",
"branch_name": new_branch.name,
"commit": new_branch.commit
}
except Exception as e:
return {
"status": "error",
"message": f"Create {branch_name} in {project_id} project failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {branch_name} in {project_id} project failed: {str(e)}"
}
@mcp.tool()
def delete_branches(project_id: str, branch_name: str) -> dict:
"""Delete a branch from a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {branch_name} in {project_id} project failed: {str(e)}"
}
# Check if it's the default branch or protected branch
if branch_name == project.default_branch:
return {
"status": "error",
"message": f"Cannot delete default branch: {branch_name} in {project_id} project"
}
# Delete the branch
try:
# Try to get the branch, if it exists then delete it
existing_branch = project.branches.get(branch_name)
existing_branch.delete()
return {
"status": "success",
"message": f"Branch {branch_name} in {project_id} project successfully deleted"
}
except gitlab.exceptions.GitlabGetError:
# Branch doesn't exist, which means deletion goal is already achieved
return {
"status": "success",
"message": f"Branch {branch_name} does not exist in {project_id} project (already deleted or never existed)"
}
except Exception as e:
return {
"status": "error",
"message": f"Failed to delete {branch_name} branch in {project_id} project: {str(e)}"
}
@mcp.tool()
def create_tags(project_id: str, tag_name: str, ref_branch: str = "master", message: Optional[str] = None) -> dict:
"""Create a new tag in a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {tag_name} tag in {project_id} project failed: {str(e)}"
}
if ref_branch is None:
ref_branch = project.default_branch
try:
# Try to get the tag, if it exists return existing tag info
existing_tag = project.tags.get(tag_name)
return {
"status": "success",
"message": f"Tag {tag_name} already exists in {project_id} project",
"tag_name": existing_tag.name,
"commit": existing_tag.commit,
"message": getattr(existing_tag, 'message', '')
}
except gitlab.exceptions.GitlabGetError:
# Tag doesn't exist, create new tag
try:
tag_params = {
'tag_name': tag_name,
'ref': ref_branch
}
if message is not None:
tag_params['message'] = message
new_tag = project.tags.create(tag_params)
return {
"status": "success",
"message": f"Create {tag_name} tag in {project_id} project successfully",
"tag_name": new_tag.name,
"commit": new_tag.commit,
"message": getattr(new_tag, 'message', '')
}
except Exception as e:
return {
"status": "error",
"message": f"Create {tag_name} tag in {project_id} project failed: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"Create {tag_name} tag in {project_id} project failed: {str(e)}"
}
@mcp.tool()
def delete_tags(project_id: str, tag_name: str) -> dict:
"""Delete a tag from a GitLab project"""
try:
gl = get_gitlab_client()
project = gl.projects.get(project_id)
except gitlab.exceptions.GitlabGetError:
return {
"status": "error",
"message": f"project {project_id} not found"
}
except Exception as e:
return {
"status": "error",
"message": f"Delete {tag_name} tag in {project_id} project failed: {str(e)}"
}
try:
# Try to get the tag, if it exists then delete it
existing_tag = project.tags.get(tag_name)
# Delete the tag
existing_tag.delete()
return {
"status": "success",
"message": f"delete {tag_name} tag successfully"
}
except gitlab.exceptions.GitlabGetError:
return {
"status": "success",
"message": f"Tag {tag_name} does not exist in {project_id} project (already deleted or never existed)"
}
except Exception as e:
# Tag doesn't exist
return {
"status": "error",
"message": f"Delete {tag_name} tag in {project_id} project failed: {str(e)}"
}
if __name__ == "__main__":
mcp.run()