ai-demo/git_mcp_server/git_checkout_tool.py

112 lines
3.5 KiB
Python

"""Git 切换分支工具"""
import subprocess
import logging
from typing import Dict, Any
from base import BaseTool, ToolResult
logger = logging.getLogger(__name__)
class GitCheckoutTool(BaseTool):
"""切换或创建 Git 分支"""
@property
def parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"repo_path": {
"type": "string",
"description": "Git 仓库路径"
},
"branch_name": {
"type": "string",
"description": "分支名称"
},
"create_new": {
"type": "boolean",
"description": "是否创建新分支",
"default": True
},
"timeout": {
"type": "integer",
"description": "超时时间(秒)",
"default": 30
}
},
"required": ["repo_path", "branch_name"]
}
def execute(self,
repo_path: str,
branch_name: str,
create_new: bool = True,
timeout: int = 30,
**kwargs) -> ToolResult:
"""
切换或创建分支
Args:
repo_path: 仓库路径
branch_name: 分支名称
create_new: 是否创建新分支
timeout: 超时时间(秒)
Returns:
ToolResult: 包含切换结果的工具返回对象
"""
# 参数验证
if not repo_path or not repo_path.strip():
error_msg = "仓库路径不能为空"
logger.error(error_msg)
raise ValueError(error_msg)
if not branch_name or not branch_name.strip():
error_msg = "分支名称不能为空"
logger.error(error_msg)
raise ValueError(error_msg)
try:
# 构建命令
if create_new:
checkout_cmd = ["git", "checkout", "-b", branch_name]
else:
checkout_cmd = ["git", "checkout", branch_name]
result = subprocess.run(
checkout_cmd,
capture_output=True,
encoding='utf-8',
cwd=repo_path,
timeout=timeout
)
if result.returncode != 0:
error_msg = f"切换分支失败: {result.stderr}"
logger.error(error_msg)
raise RuntimeError(error_msg)
logger.info(f"已切换到分支: {branch_name}")
return ToolResult(
success=True,
data={
"branch_name": branch_name,
"repo_path": repo_path,
"created_new": create_new
},
message=f"已切换到分支: {branch_name}"
)
except subprocess.TimeoutExpired:
error_msg = f"git checkout 超时({timeout}秒)"
logger.error(error_msg)
raise RuntimeError(error_msg)
except FileNotFoundError:
error_msg = "未找到 git 命令,请确认已安装 git"
logger.error(error_msg)
raise RuntimeError(error_msg)
except Exception as e:
error_msg = f"git checkout 执行异常: {str(e)}"
logger.error(error_msg)
raise RuntimeError(error_msg)