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