server.py•10.8 kB
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)")