摘要:代码仓库MCP Server开发实战,让AI读写Git仓库代码,涵盖仓库浏览、代码搜索、分支管理和PR创建的MCP工具实现。

代码仓库MCP Server 让AI读写Git仓库

本文是MCP协议全栈实战专栏第53篇。标签: MCP, Git, 代码仓库, 版本控制, MCP Server

开头聊两句

前段时间团队来了个新人,让他接手一个有三年历史的老项目。他问我有没有文档,我说文档基本没有,代码就是最好的文档。然后他默默打开了代码库,面对几千个文件一脸绝望。

我就想,能不能让AI帮新同事快速了解代码库。让AI看看仓库结构,读读最近的提交记录,搜搜关键函数的实现,然后给新人讲讲这个项目是怎么回事。

说干就干,我写了个Git仓库MCP Server。第一天测试的时候AI确实能读仓库结构了,但有几个问题。一个超大的Monorepo,AI试图遍历所有文件,光列目录就花了两分钟。还有一次AI读了.git目录里的内部文件,解析出来一堆乱码。最尴尬的是AI想看某个函数的定义,但代码搜索只支持文件名匹配,搜不到函数内容。

后来我一步步优化,加了文件类型过滤、代码内容搜索、提交历史分析、分支对比等功能。这篇文章就把最终的Git仓库MCP Server完整写出来。

核心知识 Git仓库MCP Server要解决什么

AI理解代码库需要哪些能力

让AI理解一个代码库,它需要五项基本能力。

第一是看仓库结构。AI需要知道这个项目有哪些目录、哪些文件、用什么语言写的。但不能让它遍历所有文件,大仓库会卡死。需要支持目录层级展示和文件类型过滤。

第二是读代码内容。AI需要能读取指定文件的内容,支持分页读取大文件,支持按行号范围读取。

第三是看提交历史。通过提交记录AI能了解项目最近的开发动态,谁改了什么,哪些功能是新加的。需要支持按作者、时间、关键词筛选。

第四是搜索代码。AI需要能在代码内容中搜索关键词,找到函数定义、变量使用、配置项。文件名搜索不够用,必须支持内容搜索。

第五是分支和PR管理。AI需要能查看分支列表、对比分支差异、查看PR信息,这样能了解项目的开发流程和当前状态。

安全边界

Git仓库MCP Server有一个核心安全原则,只读为主,写入要极度谨慎。

我的设计是默认只提供只读工具。读结构、读文件、读历史、搜代码,这些都是只读操作,不会修改仓库。写入操作(提交、推送、创建分支)需要显式开启,并且需要二次确认。

原因很简单。AI如果误操作了Git仓库,后果可能很严重。误删分支、错误提交、强制推送覆盖别人的代码,这些事故恢复起来都很麻烦。

完整代码 Git仓库MCP Server

"""
Git仓库MCP Server - 让AI读写Git仓库
功能: 仓库结构读取、代码搜索、提交历史、分支管理、差异对比
依赖: pip install mcp GitPython
"""

import os
import json
import re
import time
import logging
from typing import Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path

from git import Repo, GitCommandError, InvalidGitRepositoryError
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("git_mcp_server")


# ============================================================
# 第一部分: 仓库结构读取器
# 负责读取Git仓库的目录结构和文件列表
# 支持层级展示、文件类型过滤、忽略规则
# ============================================================

class RepoStructureReader:
    """
    仓库结构读取器
    提供目录树展示和文件列表功能
    自动忽略.git目录和常见忽略项
    """

    # 默认忽略的目录和文件
    IGNORE_DIRS = {
        ".git", "node_modules", "__pycache__", ".venv",
        "venv", ".idea", ".vscode", "dist", "build",
        ".next", ".nuxt", "target", ".gradle",
    }

    # 默认忽略的文件扩展名
    IGNORE_EXTS = {
        ".pyc", ".pyo", ".class", ".o", ".so", ".dll",
        ".exe", ".bin", ".dat", ".db", ".sqlite",
    }

    # 支持的代码文件扩展名 (用于类型过滤)
    CODE_EXTS = {
        ".py", ".js", ".ts", ".jsx", ".tsx", ".java",
        ".go", ".rs", ".c", ".cpp", ".h", ".hpp",
        ".cs", ".rb", ".php", ".swift", ".kt",
        ".vue", ".svelte", ".html", ".css", ".scss",
        ".sql", ".sh", ".yaml", ".yml", ".json", ".xml",
        ".md", ".txt", ".toml", ".cfg", ".ini",
    }

    @classmethod
    def list_directory(
        cls,
        repo_path: str,
        sub_dir: str = "",
        max_depth: int = 3,
        file_type: str = None
    ) -> dict:
        """
        列出仓库中指定目录下的文件和子目录
        sub_dir: 相对于仓库根目录的子目录路径
        max_depth: 最大递归深度, 防止大仓库卡死
        file_type: 文件类型过滤, 如 "python", "javascript"
        """
        # 构建完整路径
        full_path = os.path.join(repo_path, sub_dir) if sub_dir else repo_path

        if not os.path.exists(full_path):
            return {"error": f"目录不存在: {sub_dir}"}

        # 递归读取目录结构
        tree = cls._build_tree(
            full_path, repo_path, sub_dir,
            depth=0, max_depth=max_depth,
            file_type=file_type
        )

        return {
            "path": sub_dir or "/",
            "tree": tree,
        }

    @classmethod
    def _build_tree(
        cls,
        current_path: str,
        repo_root: str,
        rel_path: str,
        depth: int,
        max_depth: int,
        file_type: str = None
    ) -> list:
        """递归构建目录树"""
        if depth >= max_depth:
            return []

        items = []
        try:
            # 列出当前目录下的所有条目, 按名称排序
            entries = sorted(os.listdir(current_path))
        except PermissionError:
            return []

        for entry in entries:
            entry_path = os.path.join(current_path, entry)
            # 计算相对于仓库根目录的路径
            rel_entry = os.path.relpath(entry_path, repo_root)

            # 跳过忽略的目录
            if os.path.isdir(entry_path):
                if entry in cls.IGNORE_DIRS:
                    continue

                # 目录节点
                children = cls._build_tree(
                    entry_path, repo_root, rel_entry,
                    depth + 1, max_depth, file_type
                )
                items.append({
                    "name": entry,
                    "type": "directory",
                    "path": rel_entry,
                    "children": children,
                    "child_count": len(children),
                })
            else:
                # 跳过忽略的文件扩展名
                ext = Path(entry).suffix.lower()
                if ext in cls.IGNORE_EXTS:
                    continue

                # 按文件类型过滤
                if file_type and not cls._match_file_type(entry, file_type):
                    continue

                # 获取文件大小
                try:
                    size = os.path.getsize(entry_path)
                except OSError:
                    size = 0

                items.append({
                    "name": entry,
                    "type": "file",
                    "path": rel_entry,
                    "extension": ext,
                    "size": size,
                })

        return items

    @classmethod
    def _match_file_type(cls, filename: str, file_type: str) -> bool:
        """检查文件是否匹配指定的类型"""
        ext = Path(filename).suffix.lower()

        type_map = {
            "python": [".py"],
            "javascript": [".js", ".jsx"],
            "typescript": [".ts", ".tsx"],
            "java": [".java"],
            "go": [".go"],
            "rust": [".rs"],
            "c": [".c", ".h"],
            "cpp": [".cpp", ".hpp"],
            "web": [".html", ".css", ".scss"],
            "config": [".yaml", ".yml", ".json", ".toml", ".ini", ".cfg"],
            "doc": [".md", ".txt"],
        }

        allowed_exts = type_map.get(file_type, [])
        return ext in allowed_exts

    @classmethod
    def get_stats(cls, repo_path: str) -> dict:
        """获取仓库的基本统计信息"""
        total_files = 0
        total_dirs = 0
        ext_counts = {}

        for root, dirs, files in os.walk(repo_path):
            # 跳过忽略的目录
            dirs[:] = [d for d in dirs if d not in cls.IGNORE_DIRS]

            total_dirs += len(dirs)
            for f in files:
                ext = Path(f).suffix.lower()
                if ext in cls.IGNORE_EXTS:
                    continue
                total_files += 1
                ext_counts[ext] = ext_counts.get(ext, 0) + 1

        # 按文件数量排序, 取前10种扩展名
        top_exts = sorted(
            ext_counts.items(),
            key=lambda x: x[1],
            reverse=True
        )[:10]

        return {
            "total_files": total_files,
            "total_dirs": total_dirs,
            "top_extensions": [
                {"extension": ext, "count": cnt} for ext, cnt in top_exts
            ],
        }


# ============================================================
# 第二部分: 代码搜索器
# 在代码文件内容中搜索关键词
# 支持正则表达式和文件类型过滤
# ============================================================

class CodeSearcher:
    """
    代码搜索器
    在仓库中的代码文件内容里搜索关键词
    返回匹配的文件、行号和上下文
    """

    @staticmethod
    def search(
        repo_path: str,
        keyword: str,
        file_pattern: str = "*",
        max_results: int = 50,
        use_regex: bool = False,
        context_lines: int = 2
    ) -> dict:
        """
        在代码文件中搜索关键词
        keyword: 搜索关键词或正则表达式
        file_pattern: 文件名模式, 如 "*.py"
        max_results: 最多返回的匹配结果数
        use_regex: 是否使用正则表达式
        context_lines: 匹配行上下各显示多少行上下文
        """
        import fnmatch

        results = []
        total_matches = 0

        # 编译正则表达式
        if use_regex:
            try:
                pattern = re.compile(keyword, re.IGNORECASE)
            except re.error as e:
                return {"error": f"正则表达式错误: {e}"}
        else:
            # 普通关键词搜索, 转义特殊字符
            escaped = re.escape(keyword)
            pattern = re.compile(escaped, re.IGNORECASE)

        # 遍历仓库中的文件
        for root, dirs, files in os.walk(repo_path):
            # 跳过忽略的目录
            dirs[:] = [
                d for d in dirs
                if d not in RepoStructureReader.IGNORE_DIRS
            ]

            for filename in files:
                # 检查文件名是否匹配模式
                if not fnmatch.fnmatch(filename, file_pattern):
                    continue

                file_path = os.path.join(root, filename)
                ext = Path(filename).suffix.lower()

                # 跳过非代码文件
                if ext not in RepoStructureReader.CODE_EXTS:
                    continue

                # 跳过过大的文件 (>1MB)
                try:
                    if os.path.getsize(file_path) > 1024 * 1024:
                        continue
                except OSError:
                    continue

                # 读取文件内容并搜索
                try:
                    with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
                        lines = f.readlines()
                except Exception:
                    continue

                # 逐行搜索
                for i, line in enumerate(lines):
                    if pattern.search(line):
                        # 提取上下文
                        start = max(0, i - context_lines)
                        end = min(len(lines), i + context_lines + 1)

                        context = []
                        for j in range(start, end):
                            marker = ">>" if j == i else "  "
                            context.append({
                                "line_number": j + 1,
                                "content": lines[j].rstrip(),
                                "match": j == i,
                            })

                        rel_path = os.path.relpath(file_path, repo_path)
                        results.append({
                            "file": rel_path,
                            "line": i + 1,
                            "matched_text": line.strip(),
                            "context": context,
                        })

                        total_matches += 1
                        if total_matches >= max_results:
                            return {
                                "keyword": keyword,
                                "total_matches": total_matches,
                                "truncated": True,
                                "results": results,
                            }

                # 每个文件最多匹配10处, 防止一个文件占满结果
                file_matches = sum(
                    1 for r in results if r["file"] ==
                    os.path.relpath(file_path, repo_path)
                )
                if file_matches >= 10:
                    continue

        return {
            "keyword": keyword,
            "total_matches": total_matches,
            "truncated": False,
            "results": results,
        }


# ============================================================
# 第三部分: 提交历史查询器
# 查询Git仓库的提交记录
# 支持按作者、时间、关键词筛选
# ============================================================

class CommitHistoryReader:
    """
    提交历史查询器
    封装GitPython的提交历史查询功能
    支持多种筛选条件
    """

    @staticmethod
    def get_log(
        repo: Repo,
        max_count: int = 20,
        author: str = None,
        since: str = None,
        until: str = None,
        keyword: str = None,
        path: str = None
    ) -> dict:
        """
        查询提交历史
        repo: GitPython的Repo对象
        max_count: 最多返回的提交数
        author: 按作者筛选
        since: 开始日期 (如 "2024-01-01" 或 "2 weeks ago")
        until: 结束日期
        keyword: 在提交消息中搜索关键词
        path: 只返回修改了指定文件的提交
        """
        try:
            # 构建Git log参数
            kwargs = {"max_count": max_count}
            if author:
                kwargs["author"] = author
            if since:
                kwargs["since"] = since
            if until:
                kwargs["until"] = until
            if path:
                kwargs["path"] = path

            # 获取提交列表
            commits = list(repo.iter_commits(**kwargs))

            # 如果有关键词筛选, 在Python层过滤
            if keyword:
                keyword_lower = keyword.lower()
                commits = [
                    c for c in commits
                    if keyword_lower in c.message.lower()
                ]

            # 格式化提交信息
            commit_list = []
            for commit in commits:
                # 获取提交涉及的文件变更
                files_changed = []
                if commit.parents:
                    # 有父提交, 获取差异文件
                    diff = commit.parents[0].diff(commit)
                    files_changed = [
                        {
                            "path": d.a_path if d.a_path else d.b_path,
                            "change_type": d.change_type,
                        }
                        for d in diff
                    ]

                commit_list.append({
                    "hash": commit.hexsha[:12],  # 短哈希
                    "author": str(commit.author),
                    "email": commit.author.email,
                    "date": datetime.fromtimestamp(
                        commit.committed_date
                    ).isoformat(),
                    "message": commit.message.strip(),
                    "files_changed": len(files_changed),
                    "files": files_changed[:20],  # 最多显示20个文件
                })

            return {
                "total": len(commit_list),
                "commits": commit_list,
            }

        except GitCommandError as e:
            return {"error": f"Git命令错误: {str(e)}"}

    @staticmethod
    def get_commit_detail(repo: Repo, commit_hash: str) -> dict:
        """
        获取单个提交的详细信息
        包括完整的diff内容
        """
        try:
            commit = repo.commit(commit_hash)
            if not commit:
                return {"error": f"提交不存在: {commit_hash}"}

            # 获取完整diff
            diff_text = ""
            if commit.parents:
                diff_text = repo.git.show(
                    commit.hexsha, "--stat", "--patch"
                )

            return {
                "hash": commit.hexsha,
                "short_hash": commit.hexsha[:12],
                "author": str(commit.author),
                "email": commit.author.email,
                "date": datetime.fromtimestamp(
                    commit.committed_date
                ).isoformat(),
                "message": commit.message.strip(),
                "parents": [p.hexsha[:12] for p in commit.parents],
                "diff": diff_text[:10000],  # 限制diff大小
            }
        except Exception as e:
            return {"error": f"获取提交详情失败: {str(e)}"}

    @staticmethod
    def get_contributors(repo: Repo, since: str = None) -> dict:
        """
        获取仓库的贡献者统计
        按提交数量排序
        """
        try:
            kwargs = {"max_count": 10000}
            if since:
                kwargs["since"] = since

            commits = list(repo.iter_commits(**kwargs))

            # 统计每个作者的提交数
            contributor_stats = {}
            for commit in commits:
                author = str(commit.author)
                if author not in contributor_stats:
                    contributor_stats[author] = {
                        "name": author,
                        "email": commit.author.email,
                        "commits": 0,
                        "last_commit": datetime.fromtimestamp(
                            commit.committed_date
                        ).isoformat(),
                    }
                contributor_stats[author]["commits"] += 1

            # 按提交数排序
            sorted_contributors = sorted(
                contributor_stats.values(),
                key=lambda x: x["commits"],
                reverse=True
            )

            return {
                "total_contributors": len(sorted_contributors),
                "contributors": sorted_contributors,
            }
        except Exception as e:
            return {"error": f"获取贡献者失败: {str(e)}"}


# ============================================================
# 第四部分: 分支管理器
# 查看分支列表、创建分支、对比差异
# ============================================================

class BranchManager:
    """
    分支管理器
    提供分支列表、分支对比、分支信息查询功能
    """

    @staticmethod
    def list_branches(repo: Repo) -> dict:
        """列出所有分支"""
        branches = []

        # 本地分支
        for branch in repo.branches:
            # 获取分支最新提交
            latest_commit = branch.commit
            branches.append({
                "name": branch.name,
                "type": "local",
                "latest_commit": latest_commit.hexsha[:12],
                "latest_message": latest_commit.message.strip()[:100],
                "latest_date": datetime.fromtimestamp(
                    latest_commit.committed_date
                ).isoformat(),
                "author": str(latest_commit.author),
            })

        # 远程分支
        try:
            for ref in repo.remote().refs:
                if ref.name.endswith("HEAD"):
                    continue
                branches.append({
                    "name": ref.name,
                    "type": "remote",
                    "latest_commit": ref.commit.hexsha[:12],
                    "latest_message": ref.commit.message.strip()[:100],
                    "latest_date": datetime.fromtimestamp(
                        ref.commit.committed_date
                    ).isoformat(),
                })
        except Exception:
            # 没有配置远程仓库时跳过
            pass

        return {
            "total": len(branches),
            "current": repo.active_branch.name if repo.active_branch else "HEAD",
            "branches": branches,
        }

    @staticmethod
    def compare_branches(
        repo: Repo,
        base: str,
        compare: str
    ) -> dict:
        """
        对比两个分支的差异
        base: 基准分支
        compare: 对比分支
        返回: 不同的文件列表和统计信息
        """
        try:
            # 获取两个分支之间的差异
            diff = repo.git.diff(
                f"{base}...{compare}",
                "--stat"
            )

            # 获取具体的文件变更
            diff_files = repo.git.diff(
                f"{base}...{compare}",
                "--name-status"
            )

            # 解析文件变更
            files = []
            for line in diff_files.strip().split("\n"):
                if line:
                    parts = line.split("\t")
                    if len(parts) >= 2:
                        files.append({
                            "status": parts[0],  # A=新增, M=修改, D=删除
                            "file": parts[1],
                        })

            return {
                "base": base,
                "compare": compare,
                "files_changed": len(files),
                "files": files,
                "diff_stat": diff,
            }
        except GitCommandError as e:
            return {"error": f"分支对比失败: {str(e)}"}


# ============================================================
# 第五部分: MCP Server主程序
# 整合所有模块, 对外提供工具
# ============================================================

class GitRepoMCPServer:
    """
    Git仓库MCP Server
    工具: 目录浏览、文件读取、代码搜索、提交历史、分支管理
    """

    def __init__(self, repo_path: str):
        self.repo_path = os.path.abspath(repo_path)

        # 验证是否是有效的Git仓库
        if not os.path.exists(self.repo_path):
            raise FileNotFoundError(f"仓库路径不存在: {self.repo_path}")

        try:
            self.repo = Repo(self.repo_path)
        except InvalidGitRepositoryError:
            raise ValueError(f"不是有效的Git仓库: {self.repo_path}")

        # MCP Server
        self.server = Server("git-repo-server")
        self._setup_handlers()

    def _setup_handlers(self):
        """注册MCP工具"""

        @self.server.list_tools()
        async def handle_list_tools() -> list[Tool]:
            return [
                Tool(
                    name="git_structure",
                    description=(
                        "查看Git仓库的目录结构。支持指定子目录和最大深度。"
                        "可以按文件类型过滤(如python/javascript/java等)。"
                        "默认最大深度3层, 避免大仓库卡顿。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "sub_dir": {
                                "type": "string",
                                "description": "子目录路径, 默认根目录"
                            },
                            "max_depth": {
                                "type": "integer",
                                "description": "最大递归深度, 默认3",
                                "default": 3
                            },
                            "file_type": {
                                "type": "string",
                                "description": "文件类型过滤"
                            }
                        }
                    }
                ),
                Tool(
                    name="git_search",
                    description=(
                        "在代码文件内容中搜索关键词。"
                        "返回匹配的文件、行号和上下文。"
                        "支持正则表达式和文件名模式过滤。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "keyword": {
                                "type": "string",
                                "description": "搜索关键词"
                            },
                            "file_pattern": {
                                "type": "string",
                                "description": "文件名模式, 如 *.py",
                                "default": "*"
                            },
                            "use_regex": {
                                "type": "boolean",
                                "description": "是否使用正则表达式",
                                "default": false
                            },
                            "max_results": {
                                "type": "integer",
                                "default": 50
                            }
                        },
                        "required": ["keyword"]
                    }
                ),
                Tool(
                    name="git_log",
                    description=(
                        "查询Git提交历史。支持按作者、时间范围、"
                        "关键词筛选。返回提交哈希、作者、日期和消息。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "max_count": {
                                "type": "integer",
                                "default": 20
                            },
                            "author": {
                                "type": "string",
                                "description": "作者名筛选"
                            },
                            "since": {
                                "type": "string",
                                "description": "开始日期, 如 2024-01-01"
                            },
                            "keyword": {
                                "type": "string",
                                "description": "提交消息关键词"
                            }
                        }
                    }
                ),
                Tool(
                    name="git_commit_detail",
                    description=(
                        "查看单个提交的详细信息, 包括完整diff内容。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "commit_hash": {
                                "type": "string",
                                "description": "提交哈希(支持短哈希)"
                            }
                        },
                        "required": ["commit_hash"]
                    }
                ),
                Tool(
                    name="git_branches",
                    description=(
                        "列出所有分支(本地和远程), 显示每个分支的最新提交。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {}
                    }
                ),
                Tool(
                    name="git_diff",
                    description=(
                        "对比两个分支的差异, 返回变更的文件列表。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "base": {
                                "type": "string",
                                "description": "基准分支名"
                            },
                            "compare": {
                                "type": "string",
                                "description": "对比分支名"
                            }
                        },
                        "required": ["base", "compare"]
                    }
                ),
                Tool(
                    name="git_stats",
                    description=(
                        "获取仓库统计信息, 包括文件总数、目录数、"
                        "主要文件类型分布。"
                    ),
                    inputSchema={
                        "type": "object",
                        "properties": {}
                    }
                ),
            ]

        @self.server.call_tool()
        async def handle_call_tool(
            name: str,
            arguments: dict
        ) -> list[TextContent]:

            if name == "git_structure":
                result = RepoStructureReader.list_directory(
                    self.repo_path,
                    sub_dir=arguments.get("sub_dir", ""),
                    max_depth=arguments.get("max_depth", 3),
                    file_type=arguments.get("file_type"),
                )
            elif name == "git_search":
                result = CodeSearcher.search(
                    self.repo_path,
                    keyword=arguments["keyword"],
                    file_pattern=arguments.get("file_pattern", "*"),
                    max_results=arguments.get("max_results", 50),
                    use_regex=arguments.get("use_regex", False),
                )
            elif name == "git_log":
                result = CommitHistoryReader.get_log(
                    self.repo,
                    max_count=arguments.get("max_count", 20),
                    author=arguments.get("author"),
                    since=arguments.get("since"),
                    keyword=arguments.get("keyword"),
                )
            elif name == "git_commit_detail":
                result = CommitHistoryReader.get_commit_detail(
                    self.repo,
                    arguments["commit_hash"]
                )
            elif name == "git_branches":
                result = BranchManager.list_branches(self.repo)
            elif name == "git_diff":
                result = BranchManager.compare_branches(
                    self.repo,
                    arguments["base"],
                    arguments["compare"]
                )
            elif name == "git_stats":
                result = RepoStructureReader.get_stats(self.repo_path)
            else:
                result = {"error": f"未知工具: {name}"}

            return [TextContent(
                type="text",
                text=json.dumps(result, ensure_ascii=False, indent=2, default=str)
            )]

    async def run(self):
        """启动MCP Server"""
        logger.info(f"Git仓库MCP Server启动, 仓库: {self.repo_path}")
        async with stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                self.server.create_initialization_options()
            )


# ============================================================
# 入口
# ============================================================

async def main():
    """
    主函数
    指定Git仓库路径, 启动MCP Server
    """
    # 仓库路径 - 改成你自己的仓库路径
    repo_path = "/home/user/projects/my_project"

    server = GitRepoMCPServer(repo_path)
    await server.run()


if __name__ == "__main__":
    asyncio.run(main())

代码分五个模块。RepoStructureReader负责目录结构读取,自动忽略node_modules、.git等目录,支持文件类型过滤和深度限制。CodeSearcher做代码内容搜索,支持正则和文件名模式。CommitHistoryReader封装GitPython的日志查询,支持多维度筛选。BranchManager管理分支列表和差异对比。GitRepoMCPServer整合所有模块对外提供7个MCP工具。

对比分析 代码阅读方案对比

方案支持的仓库规模搜索能力历史分析开发成本AI理解效果
直接读文件系统小仓库仅文件名不支持极低
Git命令行封装中等grep级别基础
GitPython MCP(本文)中大型内容+正则丰富
Sourcegraph等工具超大型语义搜索丰富高(对接)极好
LSP集成方案中大型精确定义跳转不支持极好(代码理解)

我的方案用GitPython封装,适合中等规模的仓库(1000个文件以内)。如果仓库特别大,建议配合Sourcegraph的API做代码搜索,效果会更好。如果需要AI精确理解代码结构(比如找函数定义、类型引用),可以额外集成LSP(Language Server Protocol),但那个复杂度就高很多了。

踩坑经验 大仓库遍历导致内存溢出

这个坑发生在一个有10万个文件的Monorepo上。

用户让AI"看一下项目的整体结构",AI调用了git_structure工具,没传max_depth参数,默认是3层。问题是这个Monorepo的第一层就有40多个子项目,每个子项目下面又有几十个目录,递归到第三层时,返回的目录树对象有将近5万个节点。

这个JSON对象序列化后超过3MB,Claude Desktop收到后直接卡住了。更严重的是,构建这个目录树的过程中,Python进程的内存占用飙到了2GB,差点把服务器搞OOM。

我当时的修复是加了几个限制:

第一,默认max_depth从3改成2。大部分情况下用户只需要看到前两层目录就能了解项目结构。

第二,加了单次返回的节点数上限。超过500个节点就截断,返回一个提示说"目录过大,请指定子目录查看"。

第三,加了文件大小过滤。大于1MB的文件不出现在目录树里,只显示文件名和大小。

但真正治本的方案是让AI学会先看顶层结构,再逐层深入。我在工具描述里加了一句话:"对于大型仓库,建议先不传sub_dir查看根目录,然后选择感兴趣的子目录深入查看。"这样AI就会先看第一层,选一个子目录,再看第二层,逐步深入。

还有一个关于GitPython的坑。repo.iter_commits()在大型仓库上会很慢,因为它需要遍历整个提交历史。如果仓库有10万次提交,iter_commits(max_count=10000)可能要跑好几秒。解决方案是尽量用since参数限制时间范围,减少遍历的提交数。

修复前后的性能对比:

指标修复前(10万文件仓库)修复后
目录树构建时间15秒0.5秒(2层)
响应JSON大小3MB+50KB以内
内存峰值2GB100MB
提交历史查询5秒(1万条)0.3秒(限制100条)
代码搜索全仓库30秒限文件类型后3秒

小结

这篇写了Git仓库MCP Server的完整实现。七个工具覆盖了AI理解代码库的主要需求,目录浏览看结构,代码搜索找实现,提交历史了解开发动态,分支对比看差异。

GitPython是个好库,但在大仓库上要注意性能。iter_commits要加max_count和since限制,目录遍历要加深度和节点数限制。代码搜索要过滤文件类型,不要在二进制文件和node_modules里搜。

安全方面,这套代码默认是只读的。如果需要让AI提交代码或创建分支,需要额外加写入工具,并且强烈建议加确认机制。AI提交代码这事,宁可慢一点也别出事故。

下一篇写监控告警MCP Server,让AI能查看系统运行状态。


相关推荐

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐