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()