gitlab-mpm/server/gitlab_mcp.py

#!/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")