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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
|
#!/usr/bin/env python3
"""GitLab CLI MCP Server implementation.
This module provides an MCP server that integrates with GitLab through the GitLab CLI
(glab).
"""
import json
import logging
import os
import subprocess
import sys
from typing import Any
from mcp.server.fastmcp import FastMCP
# reconfigure UnicodeEncodeError prone default (i.e. windows-1252) to utf-8
if sys.platform == "win32" and os.environ.get("PYTHONIOENCODING") is None:
sys.stdin.reconfigure(encoding="utf-8")
sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")
logger = logging.getLogger("mcp_gitlab_glab_server")
class GitLabServer:
"""GitLab server implementation using the GitLab CLI (glab)."""
def __init__(self) -> None:
"""Initialize the GitLab server."""
self.auth_message = (
"Authentication required. Please run 'glab auth login' to authenticate."
)
def execute_glab_command(
self, args: list[str], working_directory: str
) -> tuple[bool, Any]:
"""Execute a glab command and return the result.
Args:
args: List of command arguments to pass to glab.
working_directory: The directory to execute the command in.
Returns:
A tuple containing:
- A boolean indicating success (True) or failure (False)
- The command output (if successful) or an error message (if failed)
"""
try:
result = subprocess.run(
["glab"] + args,
capture_output=True,
text=True,
check=False,
cwd=working_directory,
)
if result.returncode != 0:
error_msg = result.stderr.strip()
logger.error(f"glab command failed: {error_msg}")
# Check for authentication errors
if "authentication required" in error_msg.lower():
return False, {"error": self.auth_message}
return False, {"error": error_msg}
# For API commands, parse JSON output
if args and args[0] == "api":
try:
return True, json.loads(result.stdout)
except json.JSONDecodeError:
logger.error("Failed to parse JSON response")
return False, {"error": "Failed to parse JSON response"}
return True, result.stdout.strip()
except FileNotFoundError:
logger.error("glab command not found")
return False, {
"error": "glab command not found. Please install GitLab CLI."
}
except Exception as e:
logger.error(f"Command execution failed: {str(e)}")
return False, {"error": f"Command execution failed: {str(e)}"}
def check_availability(self, working_directory: str) -> dict[str, Any]:
"""Check if the glab CLI tool is available and accessible.
Args:
working_directory: The directory to execute the command in.
Returns:
A dictionary containing availability status and version information.
"""
success, result = self.execute_glab_command(["--version"], working_directory)
if success:
return {
"available": True,
"version": result,
}
else:
return {
"available": False,
"error": result.get("error", "Unknown error"),
}
def find_project(
self, project_name: str, working_directory: str
) -> dict[str, Any] | list[dict[str, Any]]:
"""Find GitLab projects by name.
Args:
project_name: The name of the project to search for.
working_directory: The directory to execute the command in.
Returns:
A list of dictionaries containing project information if found,
or an error message.
"""
success, result = self.execute_glab_command(
["api", f"/projects?search_namespaces=true&search={project_name}"], working_directory
)
if not success:
return result
# Check if any projects were found
if isinstance(result, list) and len(result) > 0:
# Return all matching projects
projects = []
for project in result:
projects.append(
{
"id": project.get("id"),
"name": project.get("name"),
"path_with_namespace": project.get("path_with_namespace"),
"web_url": project.get("web_url"),
"description": project.get("description"),
}
)
return projects
else:
return {"error": f"Project '{project_name}' not found"}
def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
"""Create and configure the FastMCP server.
Args:
host: The host to bind to for remote transport.
port: The port to bind to for remote transport.
Returns:
A configured FastMCP server instance.
"""
# Create a FastMCP server with host and port settings
mcp = FastMCP("GitLab CLI", host=host, port=port)
# Create a GitLabServer instance
gitlab = GitLabServer()
# Add check_glab_availability tool
@mcp.tool()
def check_glab_availability(working_directory: str) -> dict[str, Any]:
"""Check if the glab CLI tool is available and accessible.
Args:
working_directory: The directory to execute the command in.
Returns:
A dictionary containing availability status and version information.
"""
return gitlab.check_availability(working_directory)
# Add find_project tool
@mcp.tool()
def find_project(
project_name: str, working_directory: str
) -> dict[str, Any] | list[dict[str, Any]]:
"""Find GitLab projects by name.
Args:
project_name: The name of the project to search for.
working_directory: The directory to execute the command in.
Returns:
A list of dictionaries containing project information if found,
or an error message.
"""
return gitlab.find_project(project_name, working_directory)
return mcp
async def main(transport_type: str, host: str, port: int) -> None:
"""Start the server with the specified transport.
Args:
transport_type: The transport type to use ("stdio" or "remote").
host: The host to bind to for remote transport.
port: The port to bind to for remote transport.
"""
logger.info("Starting MCP GitLab CLI Server")
logger.info(f"Starting GitLab CLI MCP Server with {transport_type} transport")
# Create the server with host and port
mcp = create_server(host=host, port=port)
# Run the server with the appropriate transport
if transport_type == "stdio":
logger.info("Server running with stdio transport")
await mcp.run_stdio_async()
else: # remote transport
logger.info(f"Server running with remote transport on {host}:{port}")
await mcp.run_sse_async()
|
