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, sys from dotenv import load_dotenv load_dotenv() # ----------- # Logging # ------------ logger = logging.getLogger(__name__) # logger.setLevel("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.tool() 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__": print("multitools-server is running 🚀🚀🚀", file = sys.stderr) mcp.run(transport = 'stdio')