.gitignore

@@ -5,4 +5,5 @@ __pycache__/
 db_config.py
 base_config.py
 
-.idea/
+.idea/
+test/

README.md

@@ -1,5 +1,11 @@
-### TO DO ：
+# GitNexus-Core
 
-修改数据库内时间为13位毫秒时间戳
+先前往阿里云百炼平台创建应用并收集apikey等信息。
+
+编辑数据库及大模型配置文件：
+
+```
+cp db_config_example.py db_config.py
+cp base_config_example.py base_config.py
+```
 
-提取所有配置（apikey、appid）到统一的配置文件 并取消跟踪

api/aiRouter.py

@@ -1,5 +1,5 @@
 import os, json, time, asyncio
-from base_config import ai_key, path
+from base_config import ai_key, path, file_summary_app_id, commit_summary_app_id, filter_code_files_app_id, analysis_results_app_id
 from fastapi import APIRouter, BackgroundTasks
 from pathlib import Path
 from pydantic import BaseModel
@@ -9,7 +9,8 @@ from http import HTTPStatus
 from dashscope import Application
 
 from models.aiModels import Scan_Tasks, Commit_Summary_Tasks, File_Summary_Tasks
-from models.gitModels import Repos
+
+from mcp_.client import MCPClient
 
 airouter = APIRouter()
 class RequestCommit(BaseModel):
@@ -28,6 +29,10 @@ class RequestFile(BaseModel):
     repo_url: str
     file_path: str
 
+class RequestChat(BaseModel):
+    uuid: str
+    message: str
+
 def generate_repo_path(uuid, repo_url):
     repo_name = repo_url.split("/")[-1].replace(".git", "")
     base_path = os.path.join(path, uuid)
@@ -35,9 +40,8 @@ def generate_repo_path(uuid, repo_url):
 
 async def file_summary(content):
     response = Application.call(
-        # 若没有配置环境变量，可用百炼API Key将下行替换为：api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中，以减少API Key泄露风险。
         api_key=ai_key,
-        app_id='ef50d70cd4074a899a09875e6a6e36ea',
+        app_id=file_summary_app_id,
         prompt=content)
     if response.status_code == HTTPStatus.OK:
         try:
@@ -45,6 +49,7 @@ async def file_summary(content):
             print(json_data)
         except json.JSONDecodeError:
             print("返回内容不是有效的 JSON 格式！")
+            print(response.output.text)
             json_data = {"summary": []}
     else:
         print(f"请求失败: {response.message}")
@@ -53,9 +58,8 @@ async def file_summary(content):
 
 async def commit_summary(content):
     response = Application.call(
-        # 若没有配置环境变量，可用百炼API Key将下行替换为：api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中，以减少API Key泄露风险。
         api_key=ai_key,
-        app_id='88426cc2301b44bea5d28d41d187ebf2',
+        app_id=commit_summary_app_id,
         prompt=content)
     if response.status_code == HTTPStatus.OK:
         try:
@@ -63,6 +67,7 @@ async def commit_summary(content):
             print(json_data)
         except json.JSONDecodeError:
             print("返回内容不是有效的 JSON 格式！")
+            print(response.output.text)
             json_data = {"null": []}
     else:
         print(f"请求失败: {response.message}")
@@ -71,9 +76,8 @@ async def commit_summary(content):
 
 def filter_code_files(prompt):
     response = Application.call(
-        # 若没有配置环境变量，可用百炼API Key将下行替换为：api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中，以减少API Key泄露风险。
         api_key=ai_key,
-        app_id='b0725a23eafd4422bfa7d5eff278af7c',
+        app_id=filter_code_files_app_id,
         prompt=prompt)
     if response.status_code == HTTPStatus.OK:
         try:
@@ -81,6 +85,7 @@ def filter_code_files(prompt):
             print(json_data)
         except json.JSONDecodeError:
             print("返回内容不是有效的 JSON 格式！")
+            print(response.output.text)
             json_data={"files":[]}
     else:
         print(f"请求失败: {response.message}")
@@ -94,11 +99,9 @@ def analysis_results(local_path,path):
         for line_num, line in enumerate(f, start=1):
             prompt+=f"{line_num}\t{line}"
     response = Application.call(
-        # 若没有配置环境变量，可用百炼API Key将下行替换为：api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中，以减少API Key泄露风险。
         api_key=ai_key,
-        app_id='b6edb4f5ff1c49f9855af27b14a0e8b4',  # 替换为实际的应用 ID
+        app_id=analysis_results_app_id,
         prompt=prompt)
-
     if response.status_code == HTTPStatus.OK:
         try:
             json_data = json.loads(response.output.text)
@@ -110,7 +113,7 @@ def analysis_results(local_path,path):
     else:
         print(f"请求失败: {response.message}")
         json_data = {"summary":None}
-    json_data["path"]=file_path
+    json_data["path"]=path
 
     return json_data
 
@@ -146,7 +149,6 @@ async def get_code_files(path):
     chunks = [files[i * 500: (i + 1) * 500]
               for i in range(0, len(files) // 500 + 1)]
     # 提交所有批次任务
-    # futures = [executor.submit(process_batch1, chunk) for chunk in chunks]
     tasks = [process_batch1(chunk) for chunk in chunks]
     futures = await asyncio.gather(*tasks, return_exceptions=True)
     # 实时获取已完成任务的结果
@@ -171,9 +173,23 @@ async def process_batch2(local_path,path):
 
 async def analysis(local_path, task_id):
     file_list = await get_code_files(local_path)
-    print(file_list)
+    all_extensions = [
+         "adoc", "asm", "awk", "bas", "bat", "bib", "c", "cbl", "cls", "clj",
+        "cljc", "cljs", "cmd", "conf", "cpp", "cr", "cs", "css", "cxx", "dart",
+        "dockerfile", "edn", "el", "env", "erl", "ex", "exs", "f", "f90", "f95",
+        "fs", "fsscript", "fsi", "fsx", "g4", "gd", "gql", "graphql", "groovy",
+        "gsh", "gvy", "h", "hbs", "hcl", "hh", "hl", "hpp", "hrl", "hs", "htm",
+        "html", "hx", "ini", "jad", "jade", "java", "jl", "js", "json", "json5",
+        "jsx", "kt", "kts", "less", "lfe", "lgt", "lhs", "log", "ltx", "lua",
+        "m", "mjs", "ml", "mli", "mm", "nim", "nims", "nlogo", "pas", "php",
+        "pl", "plantuml", "pro", "ps1", "pug", "puml", "py", "qml", "r", "rb",
+        "re", "rei", "res", "resi", "rkt", "rs", "rst", "s", "sass", "scala",
+        "scm", "scss", "sed", "sh", "sol", "sql", "ss", "st", "squeak", "swift",
+        "tcl", "tex", "tf", "tfvars", "toml", "ts", "tsx", "txt", "v", "vb",
+        "vbs", "vh", "vhd", "vhdl", "vim", "vue", "xml", "yaml", "yang", "yml"]
+    file_list = [file for file in file_list if file.split(".")[-1]  in all_extensions]
     results = []
-    tasks = [process_batch2(local_path, file) for file in file_list]  # 假设process_batch2已改为异步函数
+    tasks = [process_batch2(local_path, file) for file in file_list]
     batch_results = await asyncio.gather(*tasks, return_exceptions=True)
 
     for result in batch_results:
@@ -223,7 +239,6 @@ async def summaryCommit(request: RequestCommit, background_tasks: BackgroundTask
     repo_commit_hash=repo_commit.repo_hash
     print(f"开始提交仓库: {repo_name}")
     await Commit_Summary_Tasks.filter(id=request.task_id).update(start_time=int(time.time() * 1000))
-    # commit_content = Repo(local_path).git.log('-1', '-p', '--pretty=format:%h %s')
     commit_content = Repo(local_path).git.diff(f"{repo_commit_hash}^", repo_commit_hash)
     background_tasks.add_task(commit_task,commit_content, request.task_id)
     return {"code": 200, "msg": "添加提交任务成功"}
@@ -234,6 +249,14 @@ async def summaryFile(request: RequestFile,background_tasks: BackgroundTasks):
     background_tasks.add_task(file_task, request.file_path, request.task_id)
     return {"code": 200, "msg": "添加提交任务成功"}
 
+@airouter.post("/chat")
+async def chat(request: RequestChat):
+    client = MCPClient(request.uuid)
+    await client.connect_to_server()
+    response = await client.process_query(request.message)
+    await client.cleanup()
+    return {"code": 200, "msg": response}
+
 
 
 

base_config_example.py

@@ -0,0 +1,11 @@
+path = "/www/gitnexus_repos/"
+avatar_url = "https://cravatar.cn/avatar/"
+
+ai_key = "sk-xxxxxxxxxxx"
+file_summary_app_id = "ef50d7xxxxxxxxx"
+commit_summary_app_id="88426cxxxxxxxxx"
+filter_code_files_app_id = "b0725a23eaxxxxx"
+analysis_results_app_id = "b6edbxxxxxx"
+
+mcp_key = "sk-xxxxxxxxx"
+mcp_app_id = "aeee454xxxxxxxx"

db_config_example.py

@@ -0,0 +1,25 @@
+TORTOISE_ORM = {
+    "connections": {
+        "default": {
+            "engine": "tortoise.backends.mysql",
+            "credentials": {
+                "host": "127.0.0.1",
+                "port": 3306,
+                "user": "gitnexus",
+                "password": "",
+                "database": "gitnexus",
+                "minsize": 3,    # 最小连接数
+                "maxsize": 20,   # 最大连接数
+                "charset": "utf8mb4"
+            }
+        }
+    },
+    "apps": {
+        "models": {
+            "models": ["models.gitModels"],  # 包含模型文件
+            "default_connection": "default"
+        }
+    },
+    "use_tz": False,
+    "timezone": "Asia/Shanghai"
+}

demo.py

@@ -15,19 +15,6 @@ from db_config import TORTOISE_ORM
 app = FastAPI()
 monkey_patch_for_docs_ui(app)
 register_tortoise(app=app, config=TORTOISE_ORM)
-
-@app.get("/user/{id}")
-async def test(id: int):
-    user= await Users.get(id=id)
-    print(type(user))
-    return user
-
-@app.get("/task/{id}")
-async def test(id: int):
-    task = await Scan_Tasks.create(repo_id=1, state=1,result={"a":1}, create_time=1234567890,scan_start_time=1234567890,scan_end_time=1234567890,create_user="admin",repo_hash="1234567890")
-    print(type(task))
-    return task
-
 app.include_router(gitrouter,prefix="/git")
 app.include_router(testapi,prefix="/test")
 

mcp_/client.py

@@ -0,0 +1,127 @@
+import json
+import asyncio
+import re
+from typing import Optional
+from contextlib import AsyncExitStack
+from http import HTTPStatus
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from dashscope import Application
+from base_config import mcp_key, mcp_app_id
+# from dotenv import load_dotenv
+
+# load_dotenv()
+
+
+
+
+class MCPClient:
+    def __init__(self,session_id):
+        self.session: Optional[ClientSession] = None
+        self.exit_stack = AsyncExitStack()
+        self.session_id = session_id
+        self.api_key = mcp_key
+        self.app_id = mcp_app_id
+        self.model = "qwen-max"
+        self.SCRIPT_PATH = "mcp_/server.py"
+        self.SYSTEM_PROMPT = """
+您是 GitNexus 网站中的智能助手“小吉”，具备调用指定工具执行 Git 操作的能力。以下是您的行为规范：
+
+可用工具简介：
+{tools_description}
+
+响应规则：
+
+响应规则如下：
+
+1. 当任务需要调用工具完成时，必须返回**严格符合 JSON 格式**的响应：
+{{
+    "tool": "工具名称",
+    "arguments": {{参数键值对}}
+}}
+2. 如果任务不需要工具支持，请直接用自然语言回答，无需 JSON。
+
+3. 所有 Git 仓库统一位于路径 `/www/gitnexus_repos/{uuid}/仓库名称`，但请注意：
+   **在回复中请勿直接透露仓库的物理路径。**
+
+4. 工具列表：
+{available_tools}"""
+
+    async def connect_to_server(self):
+        server_params = StdioServerParameters(
+            command="python",
+            args=[self.SCRIPT_PATH]
+        )
+        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
+        self.stdio, self.write = stdio_transport
+        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
+
+        await self.session.initialize()
+
+
+        response = await self.session.list_tools()
+        tools = response.tools
+        print("\nConnected to server with tools:", [tool.name for tool in tools])
+
+    async def _call_bailian_api(self, prompt: str) -> str:
+
+        try:
+            tool_response = await self.session.list_tools()
+            tools_desc = "\n".join([f"- {t.name}: {t.description}" for t in tool_response.tools])
+
+            full_prompt = self.SYSTEM_PROMPT.format(
+                tools_description=tools_desc,
+                available_tools=[t.name for t in tool_response.tools],
+                uuid=self.session_id
+            )
+
+            response = await asyncio.to_thread(
+                Application.call,
+                api_key=self.api_key,
+                app_id=self.app_id,
+                prompt=full_prompt + "\n用户提问：" + prompt,
+                session_id=self.session_id,
+                stream=False,
+                incremental_output=False,
+                enable_mcp=True,
+                tool_choice="auto"
+            )
+            if response.status_code == HTTPStatus.OK:
+                return response.output.text
+            return f"API Error: {response.message}"
+
+        except Exception as e:
+            print(e)
+            return f"调用异常: {str(e)}"
+
+    async def process_query(self, query: str) -> str:
+        bailian_response = await self._call_bailian_api(query)
+        # print("respone:"+bailian_response)
+        json_str = re.findall( r'\{.*\}', bailian_response, re.S)
+        # print(json_str)
+        if json_str:
+            tool_call=json.loads(json_str[0])
+            if "tool" in tool_call:
+                result = await self.session.call_tool(
+                    tool_call["tool"],
+                    tool_call["arguments"]
+                )
+                final_answer = await self._call_bailian_api("tool_response:"+result.content[0].text)
+                return final_answer
+        else:
+            return bailian_response
+
+    async def chat_loop(self):
+        print("\nMCP Client Started!")
+        try:
+            query = input("\nQuery: ").strip()
+            response = await self.process_query(query)
+            print("\n" + response)
+
+        except Exception as e:
+            print(f"\nError: {str(e)}")
+
+    async def cleanup(self):
+        await self.exit_stack.aclose()
+

mcp_/server.py

@@ -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.")

requirements.txt

@@ -0,0 +1,3 @@
+fastapi_cdn_host
+tortoise
+tortoise-orm