ModelContextProtocol

# Using pip
pip install mcp

# Using uv (recommended)
uv init myproject
cd myproject
uv venv
# Activate virtual environment
source .venv/bin/activate  # On macOS/Linux
.venv\Scripts\activate     # On Windows
uv add mcp

# Using npm
npm install @modelcontextprotocol/sdk

# Using yarn
yarn add @modelcontextprotocol/sdk

# Using pnpm
pnpm add @modelcontextprotocol/sdk

# Add to pom.xml (Maven)
<dependency>
    <groupId>io.modelcontextprotocol</groupId>
    <artifactId>mcp-sdk</artifactId>
    <version>latest.version</version>
</dependency>

# Or build.gradle (Gradle)
implementation 'io.modelcontextprotocol:mcp-sdk:latest.version'

# Add to pom.xml (Maven)
<dependency>
    <groupId>org.springframework.ai</groupId>
    <artifactId>spring-ai-mcp-client-spring-boot-starter</artifactId>
    <version>latest.version</version>
</dependency>

mkdir code-analyzer-mcp
cd code-analyzer-mcp

python -m venv venv
source venv/bin/activate  # On macOS/Linux
venv\Scripts\activate     # On Windows
pip install mcp pylint radon mypy

import asyncio
import os
import tempfile
import json
import subprocess
from typing import Any, Dict, List

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# Initialize the server
server = Server("code-analyzer")

@server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
    """List available code analysis tools."""
    return [
        types.Tool(
            name="analyze-code",
            description="Analyze Python code for quality issues and provide suggestions",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "Python code to analyze"
                    },
                    "analysis_types": {
                        "type": "array",
                        "items": {
                            "type": "string",
                            "enum": ["lint", "complexity", "type"]
                        },
                        "description": "Types of analysis to perform"
                    }
                },
                "required": ["code"]
            }
        ),
        types.Tool(
            name="fix-common-issues",
            description="Automatically fix common code issues",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "Python code to fix"
                    }
                },
                "required": ["code"]
            }
        )
    ]

@server.execute_tool("analyze-code")
async def handle_analyze_code(args: Dict[str, Any]) -> Any:
    """Handle code analysis tool execution."""
    code = args.get("code", "")
    analysis_types = args.get("analysis_types", ["lint", "complexity", "type"])

    if not code:
        return {"error": "No code provided"}

    results = {}

    # Create a temporary file for the code
    with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as temp_file:
        temp_file_path = temp_file.name
        temp_file.write(code.encode('utf-8'))

    try:
        # Run different analysis based on requested types
        if "lint" in analysis_types:
            results["lint_issues"] = await run_pylint(temp_file_path)

        if "complexity" in analysis_types:
            results["complexity"] = await run_complexity_analysis(temp_file_path)

        if "type" in analysis_types:
            results["type_issues"] = await run_type_analysis(temp_file_path)

        # Add a summary of the results
        total_issues = 0
        if "lint_issues" in results:
            total_issues += len(results["lint_issues"])
        if "type_issues" in results:
            total_issues += len(results["type_issues"])

        results["summary"] = {
            "total_issues": total_issues,
            "health_score": calculate_health_score(results)
        }

        return results

    finally:
        # Clean up the temporary file
        if os.path.exists(temp_file_path):
            os.unlink(temp_file_path)

@server.execute_tool("fix-common-issues")
async def handle_fix_common_issues(args: Dict[str, Any]) -> Any:
    """Handle automatic code fixing tool execution."""
    code = args.get("code", "")

    if not code:
        return {"error": "No code provided"}

    # Create a temporary file for the code
    with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as temp_file:
        temp_file_path = temp_file.name
        temp_file.write(code.encode('utf-8'))

    try:
        # Apply autoformatting using Black
        process = await asyncio.create_subprocess_exec(
            "python", "-m", "black", temp_file_path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        stdout, stderr = await process.communicate()

        # Read the modified file
        with open(temp_file_path, 'r', encoding='utf-8') as f:
            fixed_code = f.read()

        # Perform initial analysis to compare
        before_analysis = await handle_analyze_code({"code": code, "analysis_types": ["lint"]})
        after_analysis = await handle_analyze_code({"code": fixed_code, "analysis_types": ["lint"]})

        # Calculate improvement
        before_issues = len(before_analysis.get("lint_issues", []))
        after_issues = len(after_analysis.get("lint_issues", []))

        return {
            "fixed_code": fixed_code,
            "improvement": {
                "issues_before": before_issues,
                "issues_after": after_issues,
                "issues_fixed": before_issues - after_issues,
                "percent_improved": round(((before_issues - after_issues) / max(before_issues, 1)) * 100, 2)
            }
        }

    finally:
        # Clean up the temporary file
        if os.path.exists(temp_file_path):
            os.unlink(temp_file_path)

async def run_pylint(file_path):
    """Run Pylint on the given file."""
    process = await asyncio.create_subprocess_exec(
        "pylint", "--output-format=json", file_path,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    stdout, stderr = await process.communicate()

    try:
        issues = json.loads(stdout.decode())
        # Transform the data for better usability
        return [{
            "line": issue.get("line"),
            "column": issue.get("column"),
            "type": issue.get("type"),
            "symbol": issue.get("symbol"),
            "message": issue.get("message")
        } for issue in issues]
    except json.JSONDecodeError:
        return []

async def run_complexity_analysis(file_path):
    """Run complexity analysis using Radon."""
    process = await asyncio.create_subprocess_exec(
        "radon", "cc", "--json", file_path,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    stdout, stderr = await process.communicate()

    try:
        complexity_data = json.loads(stdout.decode())
        functions = []
        for filename, file_data in complexity_data.items():
            for func in file_data:
                functions.append({
                    "name": func.get("name"),
                    "complexity": func.get("complexity"),
                    "rank": func.get("rank"),
                    "line_number": func.get("lineno"),
                    "end_line": func.get("endline")
                })

        return {
            "functions": functions,
            "average_complexity": sum(f["complexity"] for f in functions) / max(len(functions), 1)
        }
    except json.JSONDecodeError:
        return {"functions": [], "average_complexity": 0}

async def run_type_analysis(file_path):
    """Run type analysis using Mypy."""
    process = await asyncio.create_subprocess_exec(
        "mypy", "--json-report", "-", file_path,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    stdout, stderr = await process.communicate()

    try:
        data = json.loads(stderr.decode())
        errors = []
        for error in data.get("errors", []):
            errors.append({
                "line": error.get("line"),
                "column": error.get("column"),
                "message": error.get("message"),
                "severity": error.get("severity")
            })
        return errors
    except json.JSONDecodeError:
        return []

def calculate_health_score(results):
    """Calculate a health score from 0-100 based on analysis results."""
    # Start with a perfect score
    score = 100

    # Deduct points for lint issues
    lint_issues = results.get("lint_issues", [])
    if lint_issues:
        score -= min(40, len(lint_issues) * 2)

    # Deduct points for complexity
    complexity = results.get("complexity", {})
    avg_complexity = complexity.get("average_complexity", 0)
    if avg_complexity > 5:
        score -= min(30, (avg_complexity - 5) * 5)

    # Deduct points for type issues
    type_issues = results.get("type_issues", [])
    if type_issues:
        score -= min(30, len(type_issues) * 3)

    return max(0, score)

async def main():
    # Run the server using stdin/stdout streams
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="code-analyzer",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "code-analyzer": {
      "command": "python",
      "args": ["/path/to/your/code-analyzer-mcp/server.py"]
    }
  }
}

def process_data(data):
    results = []
    for i in range(len(data)):
        item = data[i]
        if item['status'] == 'active':
            if item['value'] > 100:
                results.append({
                    'id': item['id'],
                    'normalized': item['value'] / 100,
                    'category': item.get('category', 'unknown')
                })
    return results