tools for git pr suggestion

main.py

@@ -1,13 +1,14 @@
 from mcp.server.fastmcp import FastMCP 
 import httpx 
 
-from typing import Any, Literal
+from pathlib import Path
+import subprocess
+from typing import Any, Literal, Optional
 from bs4 import BeautifulSoup
 import logging, os, json
 from dotenv import load_dotenv
 load_dotenv()
 
-
 # -----------
 # Logging
 # ------------
@@ -26,16 +27,47 @@ file_handler = logging.FileHandler(filename= "multitools-server.log")
 logger.addHandler(file_handler.setFormatter(fmt))
 
 
+# -------------------------
+# Initiating FastMCP server 
+# -------------------------
+mcp = FastMCP("multitools-server") 
+
+
 # --------------
 # Configuration
 #---------------
 BASE_CRICKET_URL = os.environ.get("BASE_CRICKET_URI", "False")
 
+# PR template directory (shared across all modules)
+TEMPLATES_DIR = Path(__file__).parent / "templates"
 
-# -------------------------
-# Initiating FastMCP server 
-# -------------------------
-mcp = FastMCP("multitools-server") 
+# Default PR templates
+DEFAULT_TEMPLATES = {
+    "bug.md": "Bug Fix",
+    "feature.md": "Feature",
+    "docs.md": "Documentation",
+    "refactor.md": "Refactor",
+    "test.md": "Test",
+    "performance.md": "Performance",
+    "security.md": "Security"
+}
+
+# Type mapping for PR templates
+TYPE_MAPPING = {
+    "bug": "bug.md",
+    "fix": "bug.md",
+    "feature": "feature.md",
+    "enhancement": "feature.md",
+    "docs": "docs.md",
+    "documentation": "docs.md",
+    "refactor": "refactor.md",
+    "cleanup": "refactor.md",
+    "test": "test.md",
+    "testing": "test.md",
+    "performance": "performance.md",
+    "optimization": "performance.md",
+    "security": "security.md"
+}
 
 
 # ----------------------
@@ -70,7 +102,7 @@ async def cricket_source(mode: str) -> str:
         html = BeautifulSoup(response.content, "html.parser")
         # find where the content is
         content = html.find("div", class_= 'cb-col cb-col-100 cb-rank-tabs')
-        return content
+        return json.dumps({'output': content})
     else:
         return json.dumps({"error": "No Available details right now!"})
 
@@ -82,19 +114,22 @@ async def fetch_live_cricket_details(mode: Literal["live", "upcomming"])-> str:
     """
 
     response = await cricket_source(mode.strip().lower())
-    if response['error']:
+    data = json.loads(response)
+
+    if data['error']:
         return response
-    live_details = response.get_text(separator = "\n", strip = True)
+    live_details = data['content'].get_text(separator = "\n", strip = True)
     return json.dumps({'output': str(live_details)}) 
 
-@mcp.resource("resource://scorecard_herf")
+@mcp.tools()
 async def live_cricket_scorecard_herf()-> str:
     """Returns string of comma separated anchor tags contains herf attributes that pointing to live cricket scorecards """
 
     response = await cricket_source("live")
-    if response['error']:
+    data = json.loads(response)
+    if data['error']:
         return response
-    herfs_list = response.find_all("a", class_ = "cb-text-link cb-mtch-lnks")
+    herfs_list = data["content"].find_all("a", class_ = "cb-text-link cb-mtch-lnks")      # here don't know is it possible
     herfs_string = ",".join(str(tag) for tag in herfs_list)
     return json.dumps({'output': herfs_string}) 
 
@@ -129,4 +164,175 @@ async def live_cricket_scorecard(herf: str)-> str:
     else:
         return json.dumps({'error': "No Available details right now"})
 
+
+@mcp.tool()
+async def analyze_file_changes(
+    base_branch: str = "main",
+    include_diff: bool = True,
+    max_diff_lines: int = 400,
+    working_directory: Optional[str] = None
+    ) -> str:
+    """Get the full diff and list of changed files in the current git repository.
+    
+    Args:
+        base_branch: Base branch to compare against (default: main)
+        include_diff: Include the full diff content (default: true)
+        max_diff_lines: Maximum number of diff lines to include (default: 400)
+        working_directory: Directory to run git commands in (default: current directory)
+    """
+    try:
+        # Try to get working directory from roots first
+        if working_directory is None:
+            try:
+                context = mcp.get_context()
+                roots_result = await context.session.list_roots()
+                # Get the first root - Claude Code sets this to the CWD
+                root = roots_result.roots[0]
+                # FileUrl object has a .path property that gives us the path directly
+                working_directory = root.uri.path
+            except Exception:
+                # If we can't get roots, fall back to current directory
+                pass
+        
+        # Use provided working directory or current directory
+        cwd = working_directory if working_directory else os.getcwd()
+        
+        # Debug output
+        debug_info = {
+            "provided_working_directory": working_directory,
+            "actual_cwd": cwd,
+            "server_process_cwd": os.getcwd(),
+            "server_file_location": str(Path(__file__).parent),
+            "roots_check": None
+        }
+        
+        # Add roots debug info
+        try:
+            context = mcp.get_context()
+            roots_result = await context.session.list_roots()
+            debug_info["roots_check"] = {
+                "found": True,
+                "count": len(roots_result.roots),
+                "roots": [str(root.uri) for root in roots_result.roots]
+            }
+        except Exception as e:
+            debug_info["roots_check"] = {
+                "found": False,
+                "error": str(e)
+            }
+        
+        # Get list of changed files
+        files_result = subprocess.run(
+            ["git", "diff", "--name-status", f"{base_branch}...HEAD"],
+            capture_output=True,
+            text=True,
+            check=True,
+            cwd=cwd
+        )
+        
+        # Get diff statistics
+        stat_result = subprocess.run(
+            ["git", "diff", "--stat", f"{base_branch}...HEAD"],
+            capture_output=True,
+            text=True,
+            cwd=cwd
+        )
+        
+        # Get the actual diff if requested
+        diff_content = ""
+        truncated = False
+        if include_diff:
+            diff_result = subprocess.run(
+                ["git", "diff", f"{base_branch}...HEAD"],
+                capture_output=True,
+                text=True,
+                cwd=cwd
+            )
+            diff_lines = diff_result.stdout.split('\n')
+            
+            # Check if we need to truncate
+            if len(diff_lines) > max_diff_lines:
+                diff_content = '\n'.join(diff_lines[:max_diff_lines])
+                diff_content += f"\n\n... Output truncated. Showing {max_diff_lines} of {len(diff_lines)} lines ..."
+                diff_content += "\n... Use max_diff_lines parameter to see more ..."
+                truncated = True
+            else:
+                diff_content = diff_result.stdout
+        
+        # Get commit messages for context
+        commits_result = subprocess.run(
+            ["git", "log", "--oneline", f"{base_branch}..HEAD"],
+            capture_output=True,
+            text=True,
+            cwd=cwd
+        )
+        
+        analysis = {
+            "base_branch": base_branch,
+            "files_changed": files_result.stdout,
+            "statistics": stat_result.stdout,
+            "commits": commits_result.stdout,
+            "diff": diff_content if include_diff else "Diff not included (set include_diff=true to see full diff)",
+            "truncated": truncated,
+            "total_diff_lines": len(diff_lines) if include_diff else 0,
+            "_debug": debug_info
+        }
+        
+        return json.dumps(analysis, indent=2)
+        
+    except subprocess.CalledProcessError as e:
+        return json.dumps({"error": f"Git error: {e.stderr}"})
+    except Exception as e:
+        return json.dumps({"error": str(e)})
+  
+
+
+@mcp.tool()
+async def get_pr_templates() -> str:
+    """List available PR templates with their content."""
+    templates = [
+        {
+            "filename": filename,
+            "type": template_type,
+            "content": (TEMPLATES_DIR / filename).read_text()
+        }
+        for filename, template_type in DEFAULT_TEMPLATES.items()
+    ]
+    
+    return json.dumps(templates, indent=2)
+
+
+
+@mcp.tool()
+async def suggest_template(changes_summary: str, change_type: str) -> str:
+    """Let LLM analyze the changes and suggest the most appropriate PR template.
+    
+    Args:
+        changes_summary: Your analysis of what the changes do
+        change_type: The type of change you've identified (bug, feature, docs, refactor, test, etc.)
+    """
+    
+    # Get available templates
+    templates_response = await get_pr_templates()
+    templates = json.loads(templates_response)
+    
+    # Find matching template
+    template_file = TYPE_MAPPING.get(change_type.lower(), "feature.md")
+    selected_template = next(
+        (t for t in templates if t["filename"] == template_file),
+        templates[0]  # Default to first template if no match
+    )
+    
+    suggestion = {
+        "recommended_template": selected_template,
+        "reasoning": f"Based on your analysis: '{changes_summary}', this appears to be a {change_type} change.",
+        "template_content": selected_template["content"],
+        "usage_hint": "LLM can help you fill out this template based on the specific changes in your PR."
+    }
+    
+    return json.dumps(suggestion, indent=2)
+
+if __name__ == "__main__":
+    mcp.run()
+