Git MCP

  • git_mcp
from dataclasses import dataclass from typing import AsyncIterator, Dict, List, Optional, Tuple from datetime import datetime import subprocess import os import re from mcp.server.fastmcp import FastMCP, Context from contextlib import asynccontextmanager @dataclass class GitContext: git_repos_path: str @asynccontextmanager async def git_lifespan(server: FastMCP) -> AsyncIterator[GitContext]: """Manage Git repositories lifecycle""" # read .env import dotenv dotenv.load_dotenv() git_repos_path = os.environ.get("GIT_REPOS_PATH", os.getcwd()) try: yield GitContext(git_repos_path=git_repos_path) finally: pass # No cleanup needed mcp = FastMCP("git-mcp", lifespan=git_lifespan) def _run_git_command(repo_path: str, command: List[str]) -> str: """Run a git command in the specified repository path""" if not os.path.exists(repo_path): raise ValueError(f"Repository path does not exist: {repo_path}") full_command = ["git"] + command try: result = subprocess.run( full_command, cwd=repo_path, check=True, capture_output=True, text=True ) return result.stdout.strip() except subprocess.CalledProcessError as e: error_message = e.stderr.strip() if e.stderr else str(e) raise ValueError(f"Git command failed: {error_message}") @mcp.tool() def get_last_git_tag(ctx: Context, repo_name: str) -> Dict[str, str]: """Find the last git tag in the repository Args: repo_name: Name of the git repository Returns: Dictionary containing tag version and date """ git_repos_path = ctx.request_context.lifespan_context.git_repos_path repo_path = os.path.join(git_repos_path, repo_name) # Get the most recent tag try: # Get the most recent tag name tag_name = _run_git_command(repo_path, ["describe", "--tags", "--abbrev=0"]) # Get the tag date tag_date_str = _run_git_command( repo_path, ["log", "-1", "--format=%ai", tag_name] ) # Parse the date string into a datetime object tag_date = datetime.strptime(tag_date_str, "%Y-%m-%d %H:%M:%S %z") formatted_date = tag_date.strftime("%Y-%m-%d %H:%M:%S") return {"version": tag_name, "date": formatted_date} except ValueError as e: if "No names found" in str(e): return {"version": "No tags found", "date": ""} raise e @mcp.tool() def list_commits_since_last_tag( ctx: Context, repo_name: str, max_count: Optional[int] = None ) -> List[Dict[str, str]]: """List commit messages since main HEAD and the last git tag Args: repo_name: Name of the git repository max_count: Maximum number of commits to return Returns: List of dictionaries containing commit hash, author, date, and message """ git_repos_path = ctx.request_context.lifespan_context.git_repos_path repo_path = os.path.join(git_repos_path, repo_name) try: # Try to get the most recent tag last_tag = _run_git_command(repo_path, ["describe", "--tags", "--abbrev=0"]) except ValueError as e: if "No names found" in str(e): # If no tags found, return all commits last_tag = "" else: raise e # Build the git log command log_command = ["log", "--pretty=format:%H|%an|%ad|%s", "--date=iso"] if last_tag: log_command.append(f"{last_tag}..HEAD") if max_count is not None: log_command.extend(["-n", str(max_count)]) # Get the commit logs commit_logs = _run_git_command(repo_path, log_command) # Parse the commit logs commits = [] if commit_logs: for line in commit_logs.split("\n"): if not line.strip(): continue parts = line.split("|", 3) if len(parts) == 4: commit_hash, author, date, message = parts commits.append( { "hash": commit_hash, "author": author, "date": date, "message": message, } ) return commits @mcp.tool() def list_repositories(ctx: Context) -> List[str]: """List all git repositories in the configured path Returns: List of repository names """ git_repos_path = ctx.request_context.lifespan_context.git_repos_path if not os.path.exists(git_repos_path): raise ValueError(f"Git repositories path does not exist: {git_repos_path}") repos = [] for item in os.listdir(git_repos_path): item_path = os.path.join(git_repos_path, item) if os.path.isdir(item_path) and os.path.exists(os.path.join(item_path, ".git")): repos.append(item) return repos @mcp.tool() def create_git_tag( ctx: Context, repo_name: str, tag_name: str, message: Optional[str] = None ) -> Dict[str, str]: """Create a new git tag in the repository Args: repo_name: Name of the git repository tag_name: Name of the tag to create message: Optional message for annotated tag Returns: Dictionary containing status and tag information """ git_repos_path = ctx.request_context.lifespan_context.git_repos_path repo_path = os.path.join(git_repos_path, repo_name) # Validate repository exists if not os.path.exists(repo_path) or not os.path.exists( os.path.join(repo_path, ".git") ): raise ValueError(f"Repository not found: {repo_name}") # Create the tag command if message: # Create annotated tag with message tag_command = ["tag", "-a", tag_name, "-m", message] else: # Create lightweight tag tag_command = ["tag", tag_name] # Execute the tag command try: _run_git_command(repo_path, tag_command) # Get tag date tag_date_str = _run_git_command( repo_path, ["log", "-1", "--format=%ai", tag_name] ) # Parse the date string into a datetime object tag_date = datetime.strptime(tag_date_str, "%Y-%m-%d %H:%M:%S %z") formatted_date = tag_date.strftime("%Y-%m-%d %H:%M:%S") return { "status": "success", "version": tag_name, "date": formatted_date, "type": "annotated" if message else "lightweight", } except ValueError as e: return {"status": "error", "error": str(e)} @mcp.tool() def push_git_tag(ctx: Context, repo_name: str, tag_name: str) -> Dict[str, str]: """Push a git tag to the default remote Args: repo_name: Name of the git repository tag_name: Name of the tag to push Returns: Dictionary containing status and information about the operation """ git_repos_path = ctx.request_context.lifespan_context.git_repos_path repo_path = os.path.join(git_repos_path, repo_name) # Validate repository exists if not os.path.exists(repo_path) or not os.path.exists( os.path.join(repo_path, ".git") ): raise ValueError(f"Repository not found: {repo_name}") # Validate tag exists try: _run_git_command(repo_path, ["tag", "-l", tag_name]) except ValueError: return { "status": "error", "error": f"Tag {tag_name} not found in repository {repo_name}", } # Get the default remote (usually 'origin') try: remote = _run_git_command(repo_path, ["remote"]) if not remote: return { "status": "error", "error": f"No remote configured for repository {repo_name}", } # Use the first remote if multiple are available default_remote = remote.split("\n")[0].strip() # Push the tag to the remote push_result = _run_git_command(repo_path, ["push", default_remote, tag_name]) return { "status": "success", "remote": default_remote, "tag": tag_name, "message": f"Successfully pushed tag {tag_name} to remote {default_remote}", } except ValueError as e: return {"status": "error", "error": str(e)} @mcp.tool() async def refresh_repository(ctx: Context, repo_name: str) -> Dict: """Refresh repository by checking out main branch and pulling all remotes Args: repo_name: Name of the git repository Returns: Dictionary containing status and information about the operation """ git_ctx = ctx.get_context(GitContext) repo_path = os.path.join(git_ctx.git_repos_path, repo_name) # Validate repository exists if not os.path.exists(repo_path) or not os.path.exists( os.path.join(repo_path, ".git") ): raise ValueError(f"Repository not found: {repo_name}") try: # Get all remotes remotes = _run_git_command(repo_path, ["remote"]).strip().split("\n") if not remotes or remotes[0] == "": return { "status": "error", "error": f"No remotes configured for repository {repo_name}", } # Checkout main branch try: _run_git_command(repo_path, ["checkout", "main"]) except ValueError as e: # Try master if main doesn't exist try: _run_git_command(repo_path, ["checkout", "master"]) except ValueError: return { "status": "error", "error": f"Failed to checkout main or master branch: {str(e)}", } # Pull from all remotes pull_results = {} for remote in remotes: if remote: # Skip empty remote names try: result = _run_git_command(repo_path, ["pull", remote, "main"]) pull_results[remote] = "success" except ValueError as e: # Try master if main doesn't exist try: result = _run_git_command(repo_path, ["pull", remote, "master"]) pull_results[remote] = "success" except ValueError as e: pull_results[remote] = f"error: {str(e)}" return { "status": "success", "repository": repo_name, "branch": "main", "pull_results": pull_results, } except Exception as e: return {"status": "error", "error": f"Failed to refresh repository: {str(e)}"} from importlib.metadata import version try: __version__ = version("git-mcp") print(f"Starting MCP server version {__version__}") except Exception: print("Starting MCP server (version unknown)")