feat: add bundled Python MCP server with Keychain credential storage
This commit is contained in:
@@ -0,0 +1,291 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
GitLab MCP Server for MPM CoWork
|
||||
Provides read-only access to MPM's GitLab repositories.
|
||||
|
||||
Credentials are stored in macOS Keychain via the `keyring` library.
|
||||
Run the `setup_credentials` tool once after installation to store your PAT.
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import httpx
|
||||
|
||||
try:
|
||||
import keyring
|
||||
KEYRING_AVAILABLE = True
|
||||
except ImportError:
|
||||
KEYRING_AVAILABLE = False
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
KEYRING_SERVICE = "mpm-gitlab"
|
||||
KEYRING_USERNAME = "personal_access_token"
|
||||
GITLAB_API_URL = os.environ.get("GITLAB_API_URL", "https://gitlab.com/api/v4")
|
||||
|
||||
mcp = FastMCP("gitlab-mpm")
|
||||
|
||||
|
||||
def get_token() -> str:
|
||||
"""Resolve PAT from Keychain first, fall back to environment variable."""
|
||||
if KEYRING_AVAILABLE:
|
||||
token = keyring.get_password(KEYRING_SERVICE, KEYRING_USERNAME)
|
||||
if token:
|
||||
return token
|
||||
return os.environ.get("GITLAB_PERSONAL_ACCESS_TOKEN", "")
|
||||
|
||||
|
||||
def gitlab_headers():
|
||||
return {"PRIVATE-TOKEN": get_token(), "Content-Type": "application/json"}
|
||||
|
||||
|
||||
def gitlab_get(path: str, params: dict = None) -> dict | list:
|
||||
token = get_token()
|
||||
if not token:
|
||||
raise ValueError(
|
||||
"No GitLab PAT found. Run the `setup_credentials` tool to store your token in Keychain."
|
||||
)
|
||||
url = f"{GITLAB_API_URL}/{path.lstrip('/')}"
|
||||
with httpx.Client(timeout=30) as client:
|
||||
response = client.get(url, headers=gitlab_headers(), params=params or {})
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
# ── Credential management ─────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
def setup_credentials(personal_access_token: str) -> str:
|
||||
"""
|
||||
Store a GitLab Personal Access Token securely in macOS Keychain.
|
||||
Only needs to be run once after installation, or when rotating the token.
|
||||
The token requires read_api scope on gitlab.com.
|
||||
"""
|
||||
if not KEYRING_AVAILABLE:
|
||||
return "Error: `keyring` package is not available. Cannot store credentials."
|
||||
keyring.set_password(KEYRING_SERVICE, KEYRING_USERNAME, personal_access_token)
|
||||
# Quick validation ping
|
||||
try:
|
||||
result = gitlab_get("user")
|
||||
username = result.get("username", "unknown")
|
||||
return f"Token saved to Keychain successfully. Authenticated as: {username}"
|
||||
except Exception as e:
|
||||
return f"Token saved to Keychain, but validation failed: {e}. Check that the token has read_api scope."
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def check_credentials() -> str:
|
||||
"""Check whether a GitLab PAT is stored and working."""
|
||||
token = get_token()
|
||||
if not token:
|
||||
return "No token found. Run `setup_credentials` to store your PAT."
|
||||
source = "Keychain" if (KEYRING_AVAILABLE and keyring.get_password(KEYRING_SERVICE, KEYRING_USERNAME)) else "environment variable"
|
||||
try:
|
||||
result = gitlab_get("user")
|
||||
username = result.get("username", "unknown")
|
||||
return f"Token found ({source}). Authenticated as: {username}"
|
||||
except Exception as e:
|
||||
return f"Token found ({source}) but API call failed: {e}"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def clear_credentials() -> str:
|
||||
"""Remove the stored GitLab PAT from macOS Keychain."""
|
||||
if not KEYRING_AVAILABLE:
|
||||
return "Error: `keyring` package is not available."
|
||||
existing = keyring.get_password(KEYRING_SERVICE, KEYRING_USERNAME)
|
||||
if not existing:
|
||||
return "No token found in Keychain — nothing to clear."
|
||||
keyring.delete_password(KEYRING_SERVICE, KEYRING_USERNAME)
|
||||
return "Token removed from Keychain."
|
||||
|
||||
|
||||
# ── Project discovery ─────────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
def list_projects(search: str = "", per_page: int = 20) -> str:
|
||||
"""List GitLab projects the current token has access to (membership only). Optionally filter by name."""
|
||||
params = {
|
||||
"per_page": per_page,
|
||||
"order_by": "last_activity_at",
|
||||
"sort": "desc",
|
||||
"membership": "true",
|
||||
}
|
||||
if search:
|
||||
params["search"] = search
|
||||
projects = gitlab_get("projects", params)
|
||||
results = [
|
||||
{
|
||||
"id": p["id"],
|
||||
"name": p["name"],
|
||||
"path": p["path_with_namespace"],
|
||||
"description": p.get("description", ""),
|
||||
"default_branch": p.get("default_branch", "main"),
|
||||
"last_activity": p.get("last_activity_at", ""),
|
||||
}
|
||||
for p in projects
|
||||
]
|
||||
return json.dumps(results, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_group_projects(group_path: str, per_page: int = 50) -> str:
|
||||
"""List all projects within a GitLab group. group_path is the group's URL slug (e.g. 'mpmedia-andriod')."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(group_path, safe="")
|
||||
params = {
|
||||
"per_page": per_page,
|
||||
"order_by": "last_activity_at",
|
||||
"sort": "desc",
|
||||
"include_subgroups": "true",
|
||||
}
|
||||
projects = gitlab_get(f"groups/{encoded}/projects", params)
|
||||
results = [
|
||||
{
|
||||
"id": p["id"],
|
||||
"name": p["name"],
|
||||
"path": p["path_with_namespace"],
|
||||
"description": p.get("description", ""),
|
||||
"default_branch": p.get("default_branch", "main"),
|
||||
"last_activity": p.get("last_activity_at", ""),
|
||||
}
|
||||
for p in projects
|
||||
]
|
||||
return json.dumps(results, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_project(project_id: str) -> str:
|
||||
"""Get details for a specific project by ID or path (e.g. 'group/project-name')."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(project_id, safe="")
|
||||
project = gitlab_get(f"projects/{encoded}")
|
||||
return json.dumps({
|
||||
"id": project["id"],
|
||||
"name": project["name"],
|
||||
"path": project["path_with_namespace"],
|
||||
"description": project.get("description", ""),
|
||||
"default_branch": project.get("default_branch", "main"),
|
||||
"web_url": project.get("web_url", ""),
|
||||
"visibility": project.get("visibility", ""),
|
||||
"last_activity": project.get("last_activity_at", ""),
|
||||
}, indent=2)
|
||||
|
||||
|
||||
# ── Repository browsing ───────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
def list_repository_tree(project_id: str, path: str = "", branch: str = "", recursive: bool = False) -> str:
|
||||
"""List files and directories in a repository. project_id can be numeric ID or 'group/project'."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(str(project_id), safe="")
|
||||
params = {"per_page": 100}
|
||||
if path:
|
||||
params["path"] = path
|
||||
if branch:
|
||||
params["ref"] = branch
|
||||
if recursive:
|
||||
params["recursive"] = "true"
|
||||
items = gitlab_get(f"projects/{encoded}/repository/tree", params)
|
||||
return json.dumps(items, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_file_contents(project_id: str, file_path: str, branch: str = "") -> str:
|
||||
"""Get the contents of a file from a repository."""
|
||||
import urllib.parse
|
||||
import base64
|
||||
encoded_project = urllib.parse.quote(str(project_id), safe="")
|
||||
encoded_file = urllib.parse.quote(file_path, safe="")
|
||||
params = {}
|
||||
if branch:
|
||||
params["ref"] = branch
|
||||
file_data = gitlab_get(f"projects/{encoded_project}/repository/files/{encoded_file}", params)
|
||||
content = base64.b64decode(file_data.get("content", "")).decode("utf-8", errors="replace")
|
||||
return json.dumps({
|
||||
"file_path": file_data.get("file_path", ""),
|
||||
"branch": file_data.get("ref", ""),
|
||||
"size": file_data.get("size", 0),
|
||||
"content": content,
|
||||
}, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def search_code(project_id: str, query: str, per_page: int = 20) -> str:
|
||||
"""Search for code within a specific project."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(str(project_id), safe="")
|
||||
params = {"scope": "blobs", "search": query, "per_page": per_page}
|
||||
results = gitlab_get(f"projects/{encoded}/search", params)
|
||||
simplified = [
|
||||
{
|
||||
"filename": r.get("filename", ""),
|
||||
"path": r.get("path", ""),
|
||||
"data": r.get("data", "")[:500],
|
||||
"ref": r.get("ref", ""),
|
||||
}
|
||||
for r in results
|
||||
]
|
||||
return json.dumps(simplified, indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_branches(project_id: str) -> str:
|
||||
"""List branches for a project."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(str(project_id), safe="")
|
||||
branches = gitlab_get(f"projects/{encoded}/repository/branches", {"per_page": 50})
|
||||
return json.dumps([
|
||||
{
|
||||
"name": b["name"],
|
||||
"default": b.get("default", False),
|
||||
"last_commit": b.get("commit", {}).get("title", ""),
|
||||
"committed_at": b.get("commit", {}).get("committed_date", ""),
|
||||
}
|
||||
for b in branches
|
||||
], indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_commits(project_id: str, branch: str = "", path: str = "", per_page: int = 20) -> str:
|
||||
"""List recent commits for a project, optionally filtered by branch or file path."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(str(project_id), safe="")
|
||||
params = {"per_page": per_page}
|
||||
if branch:
|
||||
params["ref_name"] = branch
|
||||
if path:
|
||||
params["path"] = path
|
||||
commits = gitlab_get(f"projects/{encoded}/repository/commits", params)
|
||||
return json.dumps([
|
||||
{
|
||||
"id": c["short_id"],
|
||||
"title": c["title"],
|
||||
"author": c.get("author_name", ""),
|
||||
"date": c.get("committed_date", ""),
|
||||
"message": c.get("message", "")[:200],
|
||||
}
|
||||
for c in commits
|
||||
], indent=2)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_commit(project_id: str, commit_sha: str) -> str:
|
||||
"""Get details for a specific commit including diff stats."""
|
||||
import urllib.parse
|
||||
encoded = urllib.parse.quote(str(project_id), safe="")
|
||||
commit = gitlab_get(f"projects/{encoded}/repository/commits/{commit_sha}")
|
||||
return json.dumps({
|
||||
"id": commit.get("id", ""),
|
||||
"short_id": commit.get("short_id", ""),
|
||||
"title": commit.get("title", ""),
|
||||
"message": commit.get("message", ""),
|
||||
"author_name": commit.get("author_name", ""),
|
||||
"authored_date": commit.get("authored_date", ""),
|
||||
"stats": commit.get("stats", {}),
|
||||
"web_url": commit.get("web_url", ""),
|
||||
}, indent=2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run(transport="stdio")
|
||||
Reference in New Issue
Block a user