1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
import logging
import os
import sys
import yaml
import gitlab
import subprocess
from typing import Any, Optional
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger("mcp_gitlab_python_server")
def get_git_remote_url(working_directory: str) -> Optional[str]:
try:
result = subprocess.run([
"git", "remote", "get-url", "origin"
], capture_output=True, text=True, check=False, cwd=working_directory)
if result.returncode == 0:
return result.stdout.strip()
except Exception as e:
logger.warning(f"Could not get git remote url: {e}")
return None
def parse_gitlab_url_from_remote(remote_url: str) -> Optional[str]:
# Handles both SSH and HTTPS remotes
if remote_url.startswith("git@"):
# git@gitlab.com:namespace/project.git
host = remote_url.split('@')[1].split(':')[0]
return f"https://{host}"
elif remote_url.startswith("https://"):
# https://gitlab.com/namespace/project.git
parts = remote_url.split('/')
if len(parts) > 2:
return f"{parts[0]}//{parts[2]}"
return None
def get_token_from_glab_config() -> Optional[str]:
config_path = os.path.expanduser("~/.config/glab-cli/config.yml")
if not os.path.exists(config_path):
return None
try:
with open(config_path, "r") as f:
config = yaml.safe_load(f)
# Try to find a token in the config (glab stores tokens per host)
hosts = config.get('hosts', {})
for host, data in hosts.items():
if 'token' in data:
return data['token']
except Exception as e:
logger.warning(f"Could not parse glab-cli config: {e}")
return None
def get_gitlab_settings(working_directory: str) -> tuple[str, str]:
# URL
url = os.environ.get("GITLAB_URL")
if not url:
remote_url = get_git_remote_url(working_directory)
if remote_url:
url = parse_gitlab_url_from_remote(remote_url)
if not url:
url = "https://gitlab.com" # fallback default
# Token
token = os.environ.get("GITLAB_TOKEN")
if not token:
token = get_token_from_glab_config()
if not token:
raise RuntimeError("No GitLab token found in env or glab-cli config.")
return url, token
class GitLabPythonServer:
def __init__(self, working_directory: str):
url, token = get_gitlab_settings(working_directory)
self.gl = gitlab.Gitlab(url, private_token=token)
self.gl.auth()
def find_project(self, project_name: str) -> list[dict[str, Any]]:
projects = self.gl.projects.list(search=project_name, all=True)
return [
{
"id": p.id,
"name": p.name,
"path_with_namespace": p.path_with_namespace,
"web_url": p.web_url,
"description": p.description,
}
for p in projects
]
def search_issues(self, project: str, **filters) -> dict[str, Any]:
try:
proj = self.gl.projects.get(project)
issues = proj.issues.list(**filters, all=True)
return {
"issues": [
{
"id": i.id,
"iid": i.iid,
"title": i.title,
"web_url": i.web_url,
"state": i.state,
"created_at": i.created_at,
"updated_at": i.updated_at,
}
for i in issues
]
}
except Exception as e:
return {"error": str(e)}
def create_issue(self, project: str, title: str, description: str, **kwargs) -> dict[str, Any]:
try:
proj = self.gl.projects.get(project)
issue = proj.issues.create({"title": title, "description": description, **kwargs})
return {"url": issue.web_url}
except Exception as e:
return {"error": str(e)}
def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
mcp = FastMCP("GitLab Python", host=host, port=port)
@mcp.tool()
def find_project(project_name: str, working_directory: str) -> list[dict[str, Any]]:
"""Find GitLab projects by name."""
server = GitLabPythonServer(working_directory)
return server.find_project(project_name)
@mcp.tool()
def search_issues(
project: str,
working_directory: str,
author_id: Optional[int] = None,
assignee_id: Optional[int] = None,
state: Optional[str] = None,
labels: Optional[list[str]] = None,
milestone: Optional[str] = None,
**kwargs
) -> dict[str, Any]:
"""Search for GitLab issues with various filters."""
server = GitLabPythonServer(working_directory)
filters = {k: v for k, v in locals().items() if v is not None and k not in ["project", "working_directory", "kwargs"]}
if labels:
filters["labels"] = ",".join(labels)
filters.update(kwargs)
return server.search_issues(project, **filters)
@mcp.tool()
def create_issue(
project: str,
title: str,
description: str,
working_directory: str,
labels: Optional[list[str]] = None,
assignee_ids: Optional[list[int]] = None,
milestone_id: Optional[int] = None,
**kwargs
) -> dict[str, Any]:
"""Create a new GitLab issue."""
server = GitLabPythonServer(working_directory)
data = {}
if labels:
data["labels"] = labels
if assignee_ids:
data["assignee_ids"] = assignee_ids
if milestone_id:
data["milestone_id"] = milestone_id
data.update(kwargs)
return server.create_issue(project, title, description, **data)
return mcp
async def main(transport_type: str, host: str, port: int) -> None:
logger.info("Starting MCP GitLab Python Server")
mcp = create_server(host=host, port=port)
if transport_type == "stdio":
logger.info("Server running with stdio transport")
await mcp.run_stdio_async()
else:
logger.info(f"Server running with remote transport on {host}:{port}")
await mcp.run_sse_async()
|
