From 91ba1558edc564eec86ba40f62c2402f58300931 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Thu, 5 Feb 2026 12:33:31 +0000 Subject: [PATCH] Add tool implementations (file, git, shell, custom) --- src/mcp_server_cli/tools/git_tools.py | 332 ++++++++++++++++++++++++++ 1 file changed, 332 insertions(+) create mode 100644 src/mcp_server_cli/tools/git_tools.py diff --git a/src/mcp_server_cli/tools/git_tools.py b/src/mcp_server_cli/tools/git_tools.py new file mode 100644 index 0000000..735db25 --- /dev/null +++ b/src/mcp_server_cli/tools/git_tools.py @@ -0,0 +1,332 @@ +"""Git integration tools for MCP Server CLI.""" + +import os +from pathlib import Path +from typing import Any, Dict, List, Optional +import subprocess + +from mcp_server_cli.tools.base import ToolBase, ToolResult +from mcp_server_cli.models import ToolSchema, ToolParameter + + +class GitTools(ToolBase): + """Built-in git operations.""" + + def __init__(self): + super().__init__( + name="git_tools", + description="Git operations: status, diff, log, commit, branch", + ) + + def _create_input_schema(self) -> ToolSchema: + return ToolSchema( + properties={ + "operation": ToolParameter( + name="operation", + type="string", + description="Git operation: status, diff, log, commit, branch", + required=True, + enum=["status", "diff", "log", "commit", "branch", "add", "checkout"], + ), + "path": ToolParameter( + name="path", + type="string", + description="Repository path (defaults to current directory)", + ), + "args": ToolParameter( + name="args", + type="string", + description="Additional arguments for the operation", + ), + }, + required=["operation"], + ) + + async def execute(self, arguments: Dict[str, Any]) -> ToolResult: + """Execute git operation.""" + operation = arguments.get("operation") + path = arguments.get("path", ".") + + repo = self._find_git_repo(path) + if not repo: + return ToolResult(success=False, output="", error="Not in a git repository") + + try: + if operation == "status": + return await self._status(repo) + elif operation == "diff": + return await self._diff(repo, arguments.get("args", "")) + elif operation == "log": + return await self._log(repo, arguments.get("args", "-10")) + elif operation == "commit": + return await self._commit(repo, arguments.get("args", "")) + elif operation == "branch": + return await self._branch(repo) + elif operation == "add": + return await self._add(repo, arguments.get("args", ".")) + elif operation == "checkout": + return await self._checkout(repo, arguments.get("args", "")) + else: + return ToolResult(success=False, output="", error=f"Unknown operation: {operation}") + except Exception as e: + return ToolResult(success=False, output="", error=str(e)) + + def _find_git_repo(self, path: str) -> Optional[Path]: + """Find the git repository root.""" + start_path = Path(path).absolute() + + if not start_path.exists(): + return None + + if start_path.is_file(): + start_path = start_path.parent + + current = start_path + while current != current.parent: + if (current / ".git").exists(): + return current + current = current.parent + + return None + + async def _run_git(self, repo: Path, *args) -> str: + """Run a git command.""" + env = os.environ.copy() + env["GIT_TERMINAL_PROMPT"] = "0" + + result = subprocess.run( + ["git"] + list(args), + cwd=repo, + capture_output=True, + text=True, + env=env, + timeout=30, + ) + + if result.returncode != 0: + raise RuntimeError(f"Git command failed: {result.stderr}") + + return result.stdout.strip() + + async def _status(self, repo: Path) -> ToolResult: + """Get git status.""" + output = await self._run_git(repo, "status", "--short") + if not output: + output = "Working tree is clean" + return ToolResult(success=True, output=output) + + async def _diff(self, repo: Path, args: str) -> ToolResult: + """Get git diff.""" + cmd = ["diff"] + if args: + cmd.extend(args.split()) + output = await self._run_git(repo, *cmd) + return ToolResult(success=True, output=output or "No changes") + + async def _log(self, repo: Path, args: str) -> ToolResult: + """Get git log.""" + cmd = ["log", "--oneline"] + if args: + cmd.extend(args.split()) + output = await self._run_git(repo, *cmd) + return ToolResult(success=True, output=output or "No commits") + + async def _commit(self, repo: Path, message: str) -> ToolResult: + """Create a commit.""" + if not message: + return ToolResult(success=False, output="", error="Commit message is required") + + output = await self._run_git(repo, "commit", "-m", message) + return ToolResult(success=True, output=f"Committed: {output}") + + async def _branch(self, repo: Path) -> ToolResult: + """List git branches.""" + output = await self._run_git(repo, "branch", "-a") + return ToolResult(success=True, output=output or "No branches") + + async def _add(self, repo: Path, pattern: str) -> ToolResult: + """Stage files.""" + output = await self._run_git(repo, "add", pattern) + return ToolResult(success=True, output=f"Staged: {pattern}") + + async def _checkout(self, repo: Path, branch: str) -> ToolResult: + """Checkout a branch.""" + if not branch: + return ToolResult(success=False, output="", error="Branch name is required") + + output = await self._run_git(repo, "checkout", branch) + return ToolResult(success=True, output=f"Switched to: {branch}") + + +class GitStatusTool(ToolBase): + """Tool for checking git status.""" + + def __init__(self): + super().__init__( + name="git_status", + description="Show working tree status", + ) + + def _create_input_schema(self) -> ToolSchema: + return ToolSchema( + properties={ + "path": ToolParameter( + name="path", + type="string", + description="Repository path (defaults to current directory)", + ), + "short": ToolParameter( + name="short", + type="boolean", + description="Use short format", + default=False, + ), + }, + ) + + async def execute(self, arguments: Dict[str, Any]) -> ToolResult: + """Get git status.""" + path = arguments.get("path", ".") + use_short = arguments.get("short", False) + + repo = Path(path).absolute() + if not (repo / ".git").exists(): + repo = repo.parent + while repo != repo.parent and not (repo / ".git").exists(): + repo = repo.parent + if not (repo / ".git").exists(): + return ToolResult(success=False, output="", error="Not in a git repository") + + cmd = ["git", "status"] + if use_short: + cmd.append("--short") + + result = subprocess.run( + cmd, + cwd=repo, + capture_output=True, + text=True, + timeout=10, + ) + + if result.returncode != 0: + return ToolResult(success=False, output="", error=result.stderr) + + return ToolResult(success=True, output=result.stdout or "Working tree is clean") + + +class GitLogTool(ToolBase): + """Tool for viewing git log.""" + + def __init__(self): + super().__init__( + name="git_log", + description="Show commit history", + ) + + def _create_input_schema(self) -> ToolSchema: + return ToolSchema( + properties={ + "path": ToolParameter( + name="path", + type="string", + description="Repository path", + ), + "n": ToolParameter( + name="n", + type="integer", + description="Number of commits to show", + default=10, + ), + "oneline": ToolParameter( + name="oneline", + type="boolean", + description="Show in oneline format", + default=True, + ), + }, + ) + + async def execute(self, arguments: Dict[str, Any]) -> ToolResult: + """Get git log.""" + path = arguments.get("path", ".") + n = arguments.get("n", 10) + oneline = arguments.get("oneline", True) + + repo = Path(path).absolute() + while repo != repo.parent and not (repo / ".git").exists(): + repo = repo.parent + if not (repo / ".git").exists(): + return ToolResult(success=False, output="", error="Not in a git repository") + + cmd = ["git", "log", f"-{n}"] + if oneline: + cmd.append("--oneline") + + result = subprocess.run( + cmd, + cwd=repo, + capture_output=True, + text=True, + timeout=10, + ) + + if result.returncode != 0: + return ToolResult(success=False, output="", error=result.stderr) + + return ToolResult(success=True, output=result.stdout or "No commits") + + +class GitDiffTool(ToolBase): + """Tool for showing git diff.""" + + def __init__(self): + super().__init__( + name="git_diff", + description="Show changes between commits", + ) + + def _create_input_schema(self) -> ToolSchema: + return ToolSchema( + properties={ + "path": ToolParameter( + name="path", + type="string", + description="Repository path", + ), + "cached": ToolParameter( + name="cached", + type="boolean", + description="Show staged changes", + default=False, + ), + }, + ) + + async def execute(self, arguments: Dict[str, Any]) -> ToolResult: + """Get git diff.""" + path = arguments.get("path", ".") + cached = arguments.get("cached", False) + + repo = Path(path).absolute() + while repo != repo.parent and not (repo / ".git").exists(): + repo = repo.parent + if not (repo / ".git").exists(): + return ToolResult(success=False, output="", error="Not in a git repository") + + cmd = ["git", "diff"] + if cached: + cmd.append("--cached") + + result = subprocess.run( + cmd, + cwd=repo, + capture_output=True, + text=True, + timeout=10, + ) + + if result.returncode != 0: + return ToolResult(success=False, output="", error=result.stderr) + + return ToolResult(success=True, output=result.stdout or "No changes")