|
|
@@ -0,0 +1,334 @@
|
|
|
+import logging
|
|
|
+from pathlib import Path
|
|
|
+from typing import Optional
|
|
|
+from enum import Enum
|
|
|
+
|
|
|
+import git
|
|
|
+from mcp.types import TextContent
|
|
|
+from mcp.server.fastmcp import FastMCP
|
|
|
+
|
|
|
+
|
|
|
+# --- Pydantic Models are REMOVED ---
|
|
|
+
|
|
|
+# --- GitTools Enum (remains the same) ---
|
|
|
+class GitTools(str, Enum):
|
|
|
+ STATUS = "git_status"
|
|
|
+ DIFF_UNSTAGED = "git_diff_unstaged"
|
|
|
+ DIFF_STAGED = "git_diff_staged"
|
|
|
+ DIFF = "git_diff"
|
|
|
+ COMMIT = "git_commit"
|
|
|
+ ADD = "git_add"
|
|
|
+ RESET = "git_reset"
|
|
|
+ LOG = "git_log"
|
|
|
+ CREATE_BRANCH = "git_create_branch"
|
|
|
+ CHECKOUT = "git_checkout"
|
|
|
+ SHOW = "git_show"
|
|
|
+ INIT = "git_init"
|
|
|
+
|
|
|
+# --- Low-level Git Functions (remain the same, using gitpython.Repo) ---
|
|
|
+# Note: Type hint updated to gitpython.Repo
|
|
|
+def _get_repo(repo_path: str | Path) -> git.Repo:
|
|
|
+ """Helper to get Repo object and handle errors."""
|
|
|
+ try:
|
|
|
+ return git.Repo(str(repo_path))
|
|
|
+ except git.InvalidGitRepositoryError:
|
|
|
+ raise ValueError(f"'{repo_path}' is not a valid Git repository.")
|
|
|
+ except git.NoSuchPathError:
|
|
|
+ raise ValueError(f"Repository path '{repo_path}' does not exist.")
|
|
|
+ except Exception as e:
|
|
|
+ raise ValueError(f"Error accessing repository at '{repo_path}': {e}")
|
|
|
+
|
|
|
+def git_status(repo: git.Repo) -> str:
|
|
|
+ return repo.git.status()
|
|
|
+
|
|
|
+def git_diff_unstaged(repo: git.Repo) -> str:
|
|
|
+ return repo.git.diff()
|
|
|
+
|
|
|
+def git_diff_staged(repo: git.Repo) -> str:
|
|
|
+ return repo.git.diff("--cached")
|
|
|
+
|
|
|
+def git_diff(repo: git.Repo, target: str) -> str:
|
|
|
+ return repo.git.diff(target)
|
|
|
+
|
|
|
+def git_commit(repo: git.Repo, message: str) -> str:
|
|
|
+ try:
|
|
|
+ # Check if there's anything staged to commit *before* attempting
|
|
|
+ # This prevents errors if the index matches HEAD but there are unstaged changes
|
|
|
+ if not repo.index.diff("HEAD"):
|
|
|
+ return "No changes added to commit (working tree clean or changes not staged)."
|
|
|
+ commit = repo.index.commit(message)
|
|
|
+ return f"Changes committed successfully with hash {commit.hexsha}"
|
|
|
+ except Exception as e:
|
|
|
+ # Catch potential errors during commit check or commit itself
|
|
|
+ return f"Error committing changes: {str(e)}"
|
|
|
+
|
|
|
+def git_add(repo: git.Repo, files: list[str]) -> str:
|
|
|
+ try:
|
|
|
+ repo.index.add(files)
|
|
|
+ return f"Files staged successfully: {', '.join(files)}"
|
|
|
+ except FileNotFoundError as e:
|
|
|
+ return f"Error staging files: File not found - {e.filename}"
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error staging files: {str(e)}"
|
|
|
+
|
|
|
+
|
|
|
+def git_reset(repo: git.Repo) -> str:
|
|
|
+ try:
|
|
|
+ # Resetting the index to HEAD (unstaging all)
|
|
|
+ repo.index.reset()
|
|
|
+ return "All staged changes reset (unstaged)"
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error resetting staged changes: {str(e)}"
|
|
|
+
|
|
|
+def git_log(repo: git.Repo, max_count: int = 10) -> list[str]:
|
|
|
+ commits = list(repo.iter_commits(max_count=max_count))
|
|
|
+ log = []
|
|
|
+ for commit in commits:
|
|
|
+ log.append(
|
|
|
+ f"Commit: {commit.hexsha}\n"
|
|
|
+ f"Author: {commit.author}\n"
|
|
|
+ f"Date: {commit.authored_datetime}\n"
|
|
|
+ f"Message: {commit.message.strip()}\n" # Use strip() for cleaner output
|
|
|
+ )
|
|
|
+ return log
|
|
|
+
|
|
|
+def git_create_branch(repo: git.Repo, branch_name: str, base_branch: Optional[str] = None) -> str:
|
|
|
+ try:
|
|
|
+ if base_branch:
|
|
|
+ try:
|
|
|
+ base = repo.refs[base_branch]
|
|
|
+ except IndexError:
|
|
|
+ try:
|
|
|
+ base = repo.commit(base_branch)
|
|
|
+ except git.BadName:
|
|
|
+ return f"Error: Base reference '{base_branch}' not found (neither branch nor commit)."
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error resolving base reference '{base_branch}': {str(e)}"
|
|
|
+ else:
|
|
|
+ base = repo.head.commit
|
|
|
+ if branch_name in repo.heads:
|
|
|
+ return f"Error: Branch '{branch_name}' already exists."
|
|
|
+
|
|
|
+ new_branch = repo.create_head(branch_name, base)
|
|
|
+ base_ref_name = getattr(base, 'name', base.hexsha) # Get branch name if possible, else hash
|
|
|
+ return f"Created branch '{new_branch.name}' based on '{base_ref_name}'"
|
|
|
+ except git.GitCommandError as e:
|
|
|
+ # Catch specific git errors if possible
|
|
|
+ return f"Error creating branch '{branch_name}': {e.stderr or e.stdout}"
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error creating branch '{branch_name}': {str(e)}"
|
|
|
+
|
|
|
+
|
|
|
+def git_checkout(repo: git.Repo, branch_name: str) -> str:
|
|
|
+ try:
|
|
|
+ repo.git.checkout(branch_name)
|
|
|
+ return f"Switched to branch '{branch_name}'"
|
|
|
+ except git.GitCommandError as e:
|
|
|
+ # Provide more specific feedback
|
|
|
+ if "did not match any file(s) known to git" in (e.stderr or ""):
|
|
|
+ return f"Error: Branch or pathspec '{branch_name}' not found."
|
|
|
+ elif "Please commit your changes or stash them before you switch branches" in (e.stderr or ""):
|
|
|
+ return f"Error: Cannot checkout branch '{branch_name}'. You have unstaged changes. Please commit or stash them first."
|
|
|
+ else:
|
|
|
+ return f"Error checking out branch '{branch_name}': {e.stderr or e.stdout}"
|
|
|
+ except Exception as e:
|
|
|
+ return f"An unexpected error occurred during checkout: {str(e)}"
|
|
|
+
|
|
|
+
|
|
|
+def git_init(repo_path: str) -> str:
|
|
|
+ try:
|
|
|
+ # Check if it already exists and is a repo
|
|
|
+ target_path = Path(repo_path)
|
|
|
+ if target_path.exists() and target_path.joinpath(".git").is_dir():
|
|
|
+ # Check if it's actually a valid repo
|
|
|
+ try:
|
|
|
+ existing_repo = git.Repo(repo_path)
|
|
|
+ return f"Repository already exists at {existing_repo.git_dir}"
|
|
|
+ except git.InvalidGitRepositoryError:
|
|
|
+ # Path exists, .git dir exists, but it's invalid. Allow re-init? Or error?
|
|
|
+ # Let's error for safety. User can delete .git if they want re-init.
|
|
|
+ return f"Error: An invalid Git repository structure already exists at '{repo_path}'. Remove '.git' folder to reinitialize."
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error checking existing repository at '{repo_path}': {e}"
|
|
|
+
|
|
|
+ # Initialize (mkdir=True handles non-existent parent dirs)
|
|
|
+ repo = git.Repo.init(path=repo_path, mkdir=True)
|
|
|
+ return f"Initialized empty Git repository in {repo.git_dir}"
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error initializing repository at '{repo_path}': {str(e)}"
|
|
|
+
|
|
|
+def git_show(repo: git.Repo, revision: str) -> str:
|
|
|
+ try:
|
|
|
+ commit = repo.commit(revision)
|
|
|
+ except git.BadName:
|
|
|
+ return f"Error: Revision '{revision}' not found."
|
|
|
+ except Exception as e:
|
|
|
+ return f"Error finding revision '{revision}': {str(e)}"
|
|
|
+
|
|
|
+ output = [
|
|
|
+ f"Commit: {commit.hexsha}\n"
|
|
|
+ f"Author: {commit.author}\n"
|
|
|
+ f"Date: {commit.authored_datetime}\n"
|
|
|
+ f"Message:\n{commit.message.strip()}\n"
|
|
|
+ ]
|
|
|
+ try:
|
|
|
+ # Show diff against first parent, or initial commit diff
|
|
|
+ diffs = commit.diff(commit.parents[0] if commit.parents else git.NULL_TREE, create_patch=True)
|
|
|
+ if diffs:
|
|
|
+ output.append("\nChanges:\n")
|
|
|
+ for d in diffs:
|
|
|
+ # Use a safer way to decode, ignoring errors
|
|
|
+ diff_text = d.diff.decode('utf-8', errors='ignore') if d.diff else ""
|
|
|
+ a_path = d.a_path or (d.a_blob.path if d.a_blob else 'unknown')
|
|
|
+ b_path = d.b_path or (d.b_blob.path if d.b_blob else 'unknown')
|
|
|
+ output.append(f"--- a/{a_path}\n+++ b/{b_path}\n")
|
|
|
+ output.append(diff_text)
|
|
|
+ else:
|
|
|
+ # Check if it's the initial commit
|
|
|
+ if not commit.parents:
|
|
|
+ output.append("\nChanges: (Initial commit)\n")
|
|
|
+ # Show the initial tree content or a message indicating it's the first commit
|
|
|
+ # For brevity, just stating it's initial might be enough.
|
|
|
+ # Or iterate through tree: for item in commit.tree.traverse(): output.append(f"+ {item.path}\n")
|
|
|
+ else:
|
|
|
+ output.append("\nNo changes in this commit compared to its parent.")
|
|
|
+
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ output.append(f"\nWarning: Could not generate diff for commit {commit.hexsha}: {str(e)}")
|
|
|
+
|
|
|
+ return "".join(output)
|
|
|
+
|
|
|
+
|
|
|
+# --- MCPFast Server Setup ---
|
|
|
+logger = logging.getLogger(__name__)
|
|
|
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
|
+
|
|
|
+# Initialize MCPFast
|
|
|
+mcp = FastMCP("git")
|
|
|
+
|
|
|
+# --- Tool Definitions using @mcp_.tool() ---
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def list_repos_tool(uuid: str) -> list[TextContent]:
|
|
|
+ """
|
|
|
+ 列出指定用户在 GitNexus 平台上的所有仓库名称（即 repo 文件夹名）。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ uuid: 用户的唯一标识，用于定位其仓库根路径。
|
|
|
+ """
|
|
|
+ user_path = Path('/www/gitnexus_repos') / uuid
|
|
|
+ logger.info(f"Executing tool: list_repos_tool for user {uuid}")
|
|
|
+
|
|
|
+ if not user_path.exists() or not user_path.is_dir():
|
|
|
+ return [TextContent(type="text", text=f"用户还没有可用的Git仓库")]
|
|
|
+
|
|
|
+ try:
|
|
|
+ repo_names = [
|
|
|
+ f.name for f in user_path.iterdir()
|
|
|
+ if f.is_dir() and (f / ".git").exists()
|
|
|
+ ]
|
|
|
+ if not repo_names:
|
|
|
+ return [TextContent(type="text", text="当前没有可用的仓库。")]
|
|
|
+
|
|
|
+ result = "您当前拥有以下仓库：\n" + "\n".join(f"- {name}" for name in repo_names)
|
|
|
+ return [TextContent(type="text", text=result)]
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception("Error while listing repositories")
|
|
|
+ return [TextContent(type="text", text=f"获取仓库列表时出错：{e}")]
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def status_tool(repo_path: str) -> list[TextContent]:
|
|
|
+ """Gets the status of a Git repository (shows staged, unstaged, and untracked files).
|
|
|
+
|
|
|
+ Args:
|
|
|
+ repo_path: The file system path to the Git repository.
|
|
|
+ """
|
|
|
+ logger.info(f"Executing tool: {GitTools.STATUS} for repo {repo_path}")
|
|
|
+ try:
|
|
|
+ repo = _get_repo(repo_path)
|
|
|
+ status = git_status(repo)
|
|
|
+ result = f"Repository status for '{repo_path}':\n{status}"
|
|
|
+ except ValueError as e:
|
|
|
+ result = str(e)
|
|
|
+ return [TextContent(type="text", text=result)]
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def diff_tool(repo_path: str, target: str) -> list[TextContent]:
|
|
|
+ """Shows differences between the current HEAD and a specified target (e.g., a branch, tag, or commit hash).
|
|
|
+
|
|
|
+ Args:
|
|
|
+ repo_path: The file system path to the Git repository.
|
|
|
+ target: The branch, tag, commit hash, or other refspec to compare HEAD against.
|
|
|
+ """
|
|
|
+ logger.info(f"Executing tool: {GitTools.DIFF} for repo {repo_path} against {target}")
|
|
|
+ try:
|
|
|
+ repo = _get_repo(repo_path)
|
|
|
+ diff = git_diff(repo, target)
|
|
|
+ result = f"Diff between HEAD and '{target}' in '{repo_path}':\n{diff or 'No differences found.'}"
|
|
|
+ except ValueError as e:
|
|
|
+ result = str(e)
|
|
|
+ except git.GitCommandError as e:
|
|
|
+ result = f"Error running git diff against '{target}': {e.stderr or e.stdout}"
|
|
|
+ return [TextContent(type="text", text=result)]
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def log_tool(repo_path: str, max_count: int = 10) -> list[TextContent]:
|
|
|
+ """Shows the commit history log for the current branch.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ repo_path: The file system path to the Git repository.
|
|
|
+ max_count: The maximum number of commits to display (default: 10).
|
|
|
+ """
|
|
|
+ logger.info(f"Executing tool: {GitTools.LOG} for repo {repo_path}, max_count={max_count}")
|
|
|
+ try:
|
|
|
+ repo = _get_repo(repo_path)
|
|
|
+ logs = git_log(repo, max_count)
|
|
|
+ if not logs:
|
|
|
+ result = f"No commit history found for '{repo_path}' (possibly an empty repository)."
|
|
|
+ else:
|
|
|
+ result = f"Commit history for '{repo_path}' (last {len(logs)} commits):\n\n" + "\n\n".join(logs) # Add double newline for readability
|
|
|
+ except ValueError as e:
|
|
|
+ result = str(e)
|
|
|
+ return [TextContent(type="text", text=result)]
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def checkout_tool(repo_path: str, branch_name: str) -> list[TextContent]:
|
|
|
+ """Switches the working directory to a different branch.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ repo_path: The file system path to the Git repository.
|
|
|
+ branch_name: The name of the branch to switch to.
|
|
|
+ """
|
|
|
+ logger.info(f"Executing tool: {GitTools.CHECKOUT} for repo {repo_path}, branch: {branch_name}")
|
|
|
+ try:
|
|
|
+ repo = _get_repo(repo_path)
|
|
|
+ result = git_checkout(repo, branch_name)
|
|
|
+ except ValueError as e:
|
|
|
+ result = str(e)
|
|
|
+ return [TextContent(type="text", text=result)]
|
|
|
+
|
|
|
+@mcp.tool()
|
|
|
+async def show_tool(repo_path: str, revision: str) -> list[TextContent]:
|
|
|
+ """Shows details (metadata and content changes) for a specific commit or object.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ repo_path: The file system path to the Git repository.
|
|
|
+ revision: The commit hash, tag, or branch name to show details for (e.g., 'HEAD', 'main', 'v1.0', 'abcdef123').
|
|
|
+ """
|
|
|
+ logger.info(f"Executing tool: {GitTools.SHOW} for repo {repo_path}, revision: {revision}")
|
|
|
+ try:
|
|
|
+ repo = _get_repo(repo_path)
|
|
|
+ result = git_show(repo, revision)
|
|
|
+ except ValueError as e:
|
|
|
+ result = str(e)
|
|
|
+ return [TextContent(type="text", text=result)]
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ logger.info("Starting Git Tool Server using MCPFast stdio transport...")
|
|
|
+ # Run using stdio transport. Ensure the environment calling this script
|
|
|
+ # is set up to communicate via stdin/stdout as expected by MCPFast stdio.
|
|
|
+ mcp.run(transport='stdio')
|
|
|
+ logger.info("MCPFast stdio transport finished.")
|
