333 lines
11 KiB
Python
333 lines
11 KiB
Python
"""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")
|