-356

pkgs/git-mcp/git-mcp.py

Reviewed

··· 1 1 - #!/usr/bin/env python3 2 2 - """ 3 3 - Git MCP Server - Advanced Git operations for AI assistants 4 4 - """ 5 5 - 6 6 - import asyncio 7 7 - from typing import Dict, List, Any 8 8 - import os 9 9 - 10 10 - from mcp.server import Server 11 11 - from mcp.server.stdio import stdio_server 12 12 - from mcp.types import Tool, TextContent 13 13 - 14 14 - 15 15 - class GitMCP: 16 16 - def __init__(self): 17 17 - self.repo_path = os.getcwd() 18 18 - 19 19 - async def run_git_command(self, cmd: List[str]) -> Dict[str, Any]: 20 20 - """Run a git command and return the result""" 21 21 - try: 22 22 - result = await asyncio.create_subprocess_exec( 23 23 - "git", 24 24 - *cmd, 25 25 - stdout=asyncio.subprocess.PIPE, 26 26 - stderr=asyncio.subprocess.PIPE, 27 27 - cwd=self.repo_path, 28 28 - ) 29 29 - stdout, stderr = await result.communicate() 30 30 - 31 31 - if result.returncode == 0: 32 32 - return {"output": stdout.decode("utf-8")} 33 33 - else: 34 34 - return {"error": stderr.decode("utf-8")} 35 35 - except Exception as e: 36 36 - return {"error": str(e)} 37 37 - 38 38 - async def git_status(self, args: Dict[str, Any]) -> str: 39 39 - """Get detailed git status""" 40 40 - cmd = ["status", "--porcelain"] 41 41 - if args.get("untracked", True): 42 42 - cmd.append("--untracked-files=all") 43 43 - if args.get("ignored"): 44 44 - cmd.append("--ignored") 45 45 - 46 46 - result = await self.run_git_command(cmd) 47 47 - if "error" in result: 48 48 - return f"Error: {result['error']}" 49 49 - 50 50 - return result["output"] 51 51 - 52 52 - async def git_diff(self, args: Dict[str, Any]) -> str: 53 53 - """Get diff with various options""" 54 54 - cmd = ["diff"] 55 55 - if args.get("staged"): 56 56 - cmd.append("--staged") 57 57 - if args.get("commit"): 58 58 - cmd.append(args["commit"]) 59 59 - if args.get("context"): 60 60 - cmd.extend(["-U", str(args["context"])]) 61 61 - if args.get("files"): 62 62 - cmd.extend(["--"] + args["files"]) 63 63 - 64 64 - result = await self.run_git_command(cmd) 65 65 - if "error" in result: 66 66 - return f"Error: {result['error']}" 67 67 - 68 68 - return result["output"] 69 69 - 70 70 - async def git_log(self, args: Dict[str, Any]) -> str: 71 71 - """Get commit history""" 72 72 - cmd = ["log", "--oneline"] 73 73 - if args.get("limit"): 74 74 - cmd.extend(["-n", str(args["limit"])]) 75 75 - if args.get("author"): 76 76 - cmd.extend(["--author", args["author"]]) 77 77 - if args.get("since"): 78 78 - cmd.extend(["--since", args["since"]]) 79 79 - if args.get("grep"): 80 80 - cmd.extend(["--grep", args["grep"]]) 81 81 - if args.get("files"): 82 82 - cmd.extend(["--"] + args["files"]) 83 83 - 84 84 - result = await self.run_git_command(cmd) 85 85 - if "error" in result: 86 86 - return f"Error: {result['error']}" 87 87 - 88 88 - return result["output"] 89 89 - 90 90 - async def git_branch(self, args: Dict[str, Any]) -> str: 91 91 - """Manage branches""" 92 92 - action = args.get("action", "list") 93 93 - 94 94 - if action == "list": 95 95 - cmd = ["branch", "-v"] 96 96 - if args.get("remote"): 97 97 - cmd.append("-r") 98 98 - elif action == "create": 99 99 - if not args.get("name"): 100 100 - return "Error: Branch name required" 101 101 - cmd = ["checkout", "-b", args["name"]] 102 102 - elif action == "delete": 103 103 - if not args.get("name"): 104 104 - return "Error: Branch name required" 105 105 - cmd = ["branch", "-d", args["name"]] 106 106 - else: 107 107 - return f"Error: Unknown action: {action}" 108 108 - 109 109 - result = await self.run_git_command(cmd) 110 110 - if "error" in result: 111 111 - return f"Error: {result['error']}" 112 112 - 113 113 - return result["output"] 114 114 - 115 115 - async def git_add(self, args: Dict[str, Any]) -> str: 116 116 - """Stage files""" 117 117 - files = args.get("files", []) 118 118 - if not files: 119 119 - return "Error: No files specified" 120 120 - 121 121 - cmd = ["add"] + files 122 122 - result = await self.run_git_command(cmd) 123 123 - if "error" in result: 124 124 - return f"Error: {result['error']}" 125 125 - 126 126 - return "Files staged successfully" 127 127 - 128 128 - async def git_commit(self, args: Dict[str, Any]) -> str: 129 129 - """Create a commit""" 130 130 - message = args.get("message") 131 131 - if not message: 132 132 - return "Error: Commit message required" 133 133 - 134 134 - cmd = ["commit", "-m", message] 135 135 - result = await self.run_git_command(cmd) 136 136 - if "error" in result: 137 137 - return f"Error: {result['error']}" 138 138 - 139 139 - return result["output"] 140 140 - 141 141 - async def git_push(self, args: Dict[str, Any]) -> str: 142 142 - """Push commits""" 143 143 - cmd = ["push"] 144 144 - if args.get("remote"): 145 145 - cmd.append(args["remote"]) 146 146 - if args.get("branch"): 147 147 - cmd.append(args["branch"]) 148 148 - 149 149 - result = await self.run_git_command(cmd) 150 150 - if "error" in result: 151 151 - return f"Error: {result['error']}" 152 152 - 153 153 - return result["output"] 154 154 - 155 155 - async def git_pull(self, args: Dict[str, Any]) -> str: 156 156 - """Pull commits""" 157 157 - cmd = ["pull"] 158 158 - if args.get("remote"): 159 159 - cmd.append(args["remote"]) 160 160 - if args.get("branch"): 161 161 - cmd.append(args["branch"]) 162 162 - 163 163 - result = await self.run_git_command(cmd) 164 164 - if "error" in result: 165 165 - return f"Error: {result['error']}" 166 166 - 167 167 - return result["output"] 168 168 - 169 169 - 170 170 - async def main(): 171 171 - git_mcp = GitMCP() 172 172 - 173 173 - # Create the MCP server 174 174 - server = Server("git-mcp") 175 175 - 176 176 - # Register tools 177 177 - @server.list_tools() 178 178 - async def list_tools() -> List[Tool]: 179 179 - return [ 180 180 - Tool( 181 181 - name="git_status", 182 182 - description="Get git repository status", 183 183 - inputSchema={ 184 184 - "type": "object", 185 185 - "properties": { 186 186 - "untracked": { 187 187 - "type": "boolean", 188 188 - "description": "Include untracked files", 189 189 - }, 190 190 - "ignored": { 191 191 - "type": "boolean", 192 192 - "description": "Include ignored files", 193 193 - }, 194 194 - }, 195 195 - "additionalProperties": False, 196 196 - }, 197 197 - ), 198 198 - Tool( 199 199 - name="git_diff", 200 200 - description="Get git diff", 201 201 - inputSchema={ 202 202 - "type": "object", 203 203 - "properties": { 204 204 - "staged": { 205 205 - "type": "boolean", 206 206 - "description": "Show staged changes", 207 207 - }, 208 208 - "commit": { 209 209 - "type": "string", 210 210 - "description": "Show diff for specific commit", 211 211 - }, 212 212 - "files": { 213 213 - "type": "array", 214 214 - "items": {"type": "string"}, 215 215 - "description": "Specific files to diff", 216 216 - }, 217 217 - "context": { 218 218 - "type": "number", 219 219 - "description": "Lines of context", 220 220 - }, 221 221 - }, 222 222 - "additionalProperties": False, 223 223 - }, 224 224 - ), 225 225 - Tool( 226 226 - name="git_log", 227 227 - description="Get commit history", 228 228 - inputSchema={ 229 229 - "type": "object", 230 230 - "properties": { 231 231 - "limit": {"type": "number", "description": "Number of commits"}, 232 232 - "author": {"type": "string", "description": "Filter by author"}, 233 233 - "since": {"type": "string", "description": "Since date"}, 234 234 - "grep": { 235 235 - "type": "string", 236 236 - "description": "Search commit messages", 237 237 - }, 238 238 - "files": { 239 239 - "type": "array", 240 240 - "items": {"type": "string"}, 241 241 - "description": "Show commits affecting files", 242 242 - }, 243 243 - }, 244 244 - "additionalProperties": False, 245 245 - }, 246 246 - ), 247 247 - Tool( 248 248 - name="git_branch", 249 249 - description="Manage branches", 250 250 - inputSchema={ 251 251 - "type": "object", 252 252 - "properties": { 253 253 - "action": { 254 254 - "type": "string", 255 255 - "enum": ["list", "create", "delete"], 256 256 - "description": "Action to perform", 257 257 - }, 258 258 - "name": {"type": "string", "description": "Branch name"}, 259 259 - "remote": { 260 260 - "type": "boolean", 261 261 - "description": "Include remote branches", 262 262 - }, 263 263 - }, 264 264 - "additionalProperties": False, 265 265 - }, 266 266 - ), 267 267 - Tool( 268 268 - name="git_add", 269 269 - description="Stage files for commit", 270 270 - inputSchema={ 271 271 - "type": "object", 272 272 - "properties": { 273 273 - "files": { 274 274 - "type": "array", 275 275 - "items": {"type": "string"}, 276 276 - "description": "Files to stage", 277 277 - } 278 278 - }, 279 279 - "required": ["files"], 280 280 - "additionalProperties": False, 281 281 - }, 282 282 - ), 283 283 - Tool( 284 284 - name="git_commit", 285 285 - description="Create a commit", 286 286 - inputSchema={ 287 287 - "type": "object", 288 288 - "properties": { 289 289 - "message": {"type": "string", "description": "Commit message"} 290 290 - }, 291 291 - "required": ["message"], 292 292 - "additionalProperties": False, 293 293 - }, 294 294 - ), 295 295 - Tool( 296 296 - name="git_push", 297 297 - description="Push commits to remote", 298 298 - inputSchema={ 299 299 - "type": "object", 300 300 - "properties": { 301 301 - "remote": {"type": "string", "description": "Remote name"}, 302 302 - "branch": {"type": "string", "description": "Branch name"}, 303 303 - }, 304 304 - "additionalProperties": False, 305 305 - }, 306 306 - ), 307 307 - Tool( 308 308 - name="git_pull", 309 309 - description="Pull commits from remote", 310 310 - inputSchema={ 311 311 - "type": "object", 312 312 - "properties": { 313 313 - "remote": {"type": "string", "description": "Remote name"}, 314 314 - "branch": {"type": "string", "description": "Branch name"}, 315 315 - }, 316 316 - "additionalProperties": False, 317 317 - }, 318 318 - ), 319 319 - ] 320 320 - 321 321 - @server.call_tool() 322 322 - async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: 323 323 - """Handle tool calls""" 324 324 - try: 325 325 - if name == "git_status": 326 326 - result = await git_mcp.git_status(arguments) 327 327 - elif name == "git_diff": 328 328 - result = await git_mcp.git_diff(arguments) 329 329 - elif name == "git_log": 330 330 - result = await git_mcp.git_log(arguments) 331 331 - elif name == "git_branch": 332 332 - result = await git_mcp.git_branch(arguments) 333 333 - elif name == "git_add": 334 334 - result = await git_mcp.git_add(arguments) 335 335 - elif name == "git_commit": 336 336 - result = await git_mcp.git_commit(arguments) 337 337 - elif name == "git_push": 338 338 - result = await git_mcp.git_push(arguments) 339 339 - elif name == "git_pull": 340 340 - result = await git_mcp.git_pull(arguments) 341 341 - else: 342 342 - result = f"Unknown tool: {name}" 343 343 - 344 344 - return [TextContent(type="text", text=result)] 345 345 - except Exception as e: 346 346 - return [TextContent(type="text", text=f"Error: {str(e)}")] 347 347 - 348 348 - # Run the server 349 349 - async with stdio_server() as (read_stream, write_stream): 350 350 - await server.run( 351 351 - read_stream, write_stream, server.create_initialization_options() 352 352 - ) 353 353 - 354 354 - 355 355 - if __name__ == "__main__": 356 356 - asyncio.run(main())

-18

pkgs/git-mcp/package.nix

Reviewed

··· 1 1 - { pkgs, ... }: 2 2 - let 3 3 - python = pkgs.python3.withPackages ( 4 4 - ps: with ps; [ 5 5 - mcp 6 6 - ] 7 7 - ); 8 8 - in 9 9 - pkgs.writeShellApplication { 10 10 - name = "git-mcp"; 11 11 - runtimeInputs = with pkgs; [ 12 12 - git 13 13 - python 14 14 - ]; 15 15 - text = '' 16 16 - ${python}/bin/python ${./git-mcp.py} "$@" 17 17 - ''; 18 18 - }