main -> server

main.py

@@ -1,338 +0,0 @@
-from mcp.server.fastmcp import FastMCP 
-import httpx 
-
-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
-# ------------
-logger = logging.getLogger(__name__) 
-logger.setLevel(logger.debug)
-
-# formatter
-fmt = logging.Formatter("%(asctime)s -- %(name)s -- %(levelname)s -- %(message)s")
-
-# handlers 
-# console_handler = logging.StreamHandler() 
-file_handler = logging.FileHandler(filename= "multitools-server.log")
-
-# add to logger 
-# logger.addHandler(console_handler) 
-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"
-
-# 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"
-}
-
-
-# ----------------------
-# Available tools for LLM
-# -----------------------
-
-async def cricket_source(mode: str) -> str:
-    """Fetches whole html from source url then extracts html container that contains necessary details"""
-
-    if mode == "live":
-        url = f"{BASE_CRICKET_URL}/cricket-match/live-scores"
-    elif mode == 'upcomming':
-        url = f"{BASE_CRICKET_URL}/cricket-match/live-scores/upcoming-matches"
-    else:
-        error = f"Not Implemented: Currently there's no implementation to handle {mode}. Only handels live, upcomming"
-        logger.error(msg= error)
-        return json.dumps({"error": error})
-        
-    try:
-        async with httpx.AsyncClient(timeout= 10.0) as client:
-            response = await client.get(url= url) 
-            response.raise_for_status()                    # if not 2xx it will raise HTTP error
-    except httpx.HTTPError as e:
-        logger.error("\n%s", e) 
-        return json.dumps({'error': str(e)})
-    except Exception as e:
-        logger.error("\n%s", e) 
-        return json.dumps({'error': str(e)})
-
-    if response:
-        # convert htmldoc content to proper html form using bs
-        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 json.dumps({'output': content})
-    else:
-        return json.dumps({"error": "No Available details right now!"})
-
-@mcp.tool()
-async def fetch_live_cricket_details(mode: Literal["live", "upcomming"])-> str:
-    """ Get cricket live or upcomming match details
-    Args:
-        mode : Either "live" or "upcomming"
-    """
-
-    response = await cricket_source(mode.strip().lower())
-    data = json.loads(response)
-
-    if data['error']:
-        return response
-    live_details = data['content'].get_text(separator = "\n", strip = True)
-    return json.dumps({'output': str(live_details)}) 
-
-@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")
-    data = json.loads(response)
-    if data['error']:
-        return response
-    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}) 
-
-
-@mcp.tool()
-async def live_cricket_scorecard(herf: str)-> str:
-    """Live cricket match scorecard details for given herf.
-    (e.g, herf = "/live-cricket-scorecard/119495/cd-vs-hbh-7th-match-global-super-league-2025")
-
-    Args:
-        herf (str): herf for scorescard endpoint
-    """
-    scorecard_url = f"{BASE_CRICKET_URL}{herf}"
-
-    try:
-        with httpx.AsyncClient(timeout= 10.0) as client:
-            response = client.get(url = scorecard_url)
-            response.raise_for_status()
-    except httpx.HTTPError as e:
-        logger.error("\n%s", e) 
-        return json.dumps({"error": str(e)})
-    except Exception as e:
-        logger.error("\n%s", e) 
-        return json.dumps({'error': str(e)})
-    
-    # extract html container
-    if response:
-        html = BeautifulSoup(response.content, "html.parser")
-        live_scorecard = html.find("div", timeout = "30000")
-        details = live_scorecard.get_text(separator="\n", strip=True)
-        return json.dumps({'output': str(details)})
-    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()
-
-