server.py

import logging, os, json, sys 
from typing import Any, Literal, Optional 
from pathlib import Path
import subprocess
from dotenv import load_dotenv
load_dotenv()  

# -----------
# Logging
# ------------
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

fmt = "%(asctime)s -- %(levelname)s -- %(name)s -- %(message)s"
file_handler = logging.FileHandler("multitools-server.log")
file_handler.setFormatter(logging.Formatter(fmt))

logger.addHandler(file_handler)

# Now import neccessary Packages
try:
    from mcp.server.fastmcp import FastMCP
    from bs4 import BeautifulSoup 
    import httpx 
except ImportError as e:
    logger.error("Got Error when Importing Packages: \n%s", e) 
    sys.exit(1) 
except Exception as e:
    logger.error("Got UnExcepted Error when Importing Packages: \n%s", e)
    sys.exit(1)

# -------------------------
# Initiating FastMCP server 
# -------------------------
mcp = FastMCP("multitools-server") 

# --------------
# Configuration
#---------------
BASE_CRICKET_URL = os.environ.get("BASE_CRICKET_URL")
logger.warning("Env variable BASE_CRICKET_URL Not-Found may cause error...") if not BASE_CRICKET_URL else logger.info("")

# PR template directory 
TEMPLATES_DIR = Path(__file__).parent / "templates"
logger.warning("TEMPLATES_DIR Not-Found may cause Error...") if not TEMPLATES_DIR else logger.info("TEMPLATES_DIR: \n%s", TEMPLATES_DIR)

# 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, want: 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"
        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 ain't 2xx it will raise HTTP error
    except httpx.HTTPError as e:
        return json.dumps({'error': str(e)})
    except Exception as 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
        container = html.find("div", class_= 'cb-col cb-col-100 cb-rank-tabs')
        if mode in ['live', 'upcomming'] and want == "text":
            text = container.get_text(separator=" ", strip= True) 
            return json.dumps({"output": str(text)})    
        elif mode == 'live' and want == 'herf':
            herfs_list = container.find_all("a", class_ = "cb-text-link cb-mtch-lnks") 
            herfs_string = ",".join(str(tag) for tag in herfs_list)
            return json.dumps({"output": herfs_string})
        else:
            return json.dumps({"error": f"Not Implemented for {mode} with {want}"})
        
    else:
        return json.dumps({"error": "No Available details right now!"})

@mcp.tool()
async def fetch_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(), want= 'text')
    return response


@mcp.tool()
async def live_cricket_scorecard_herf()-> str:
    """String of comma separated anchor tags contains herf attributes that pointing to live cricket scorecards """
    response = await cricket_source('live', 'herf')
    return response


@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: live cricket match scorecard endpoint
    """
    scorecard_url = f"{BASE_CRICKET_URL}{herf}"

    try:
        async with httpx.AsyncClient(timeout= 10.0) as client:
            response = await client.get(url = scorecard_url)
            response.raise_for_status()
    except httpx.HTTPError as e:
        return json.dumps({"error": str(e)})
    except Exception as 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=" ", 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__":
    logger.info("multitools-server is started 🚀🚀🚀")
    mcp.run(transport = 'stdio')