python-services/service2/mcp_server.py

from mcp.server.fastmcp import FastMCP
from typing import Dict, Any, List, Optional
import json
from datetime import datetime
import statistics

# Initialize FastMCP server
mcp = FastMCP(
    "DataAnalyzer",  # Name of the MCP server
    instructions="You are a data analysis assistant that can perform various types of data analysis, statistics, and data processing tasks.",
    host="0.0.0.0",
    port=8002,
)

@mcp.tool()
async def analyze_data(input_data: Dict[str, Any], operation: str = "default") -> str:
    """
    Perform data analysis on the provided input data.
    
    Args:
        input_data (Dict[str, Any]): Data to analyze
        operation (str): Type of analysis to perform
        
    Returns:
        str: Analysis results
    """
    try:
        analysis_result = {
            "input_size": len(str(input_data)),
            "operation_type": operation,
            "analysis_timestamp": datetime.now().isoformat(),
            "data_summary": f"Processed {len(input_data)} items",
            "analysis_results": {}
        }
        
        # Perform different types of analysis based on operation
        if operation == "statistics":
            if isinstance(input_data, dict):
                values = [v for v in input_data.values() if isinstance(v, (int, float))]
                if values:
                    analysis_result["analysis_results"] = {
                        "count": len(values),
                        "mean": statistics.mean(values),
                        "median": statistics.median(values),
                        "min": min(values),
                        "max": max(values),
                        "std_dev": statistics.stdev(values) if len(values) > 1 else 0
                    }
        
        elif operation == "text_analysis":
            text_content = str(input_data)
            analysis_result["analysis_results"] = {
                "character_count": len(text_content),
                "word_count": len(text_content.split()),
                "line_count": len(text_content.splitlines()),
                "unique_words": len(set(text_content.lower().split()))
            }
        
        elif operation == "structure_analysis":
            analysis_result["analysis_results"] = {
                "data_type": type(input_data).__name__,
                "is_nested": any(isinstance(v, (dict, list)) for v in input_data.values()) if isinstance(input_data, dict) else False,
                "key_types": {k: type(v).__name__ for k, v in input_data.items()} if isinstance(input_data, dict) else {},
                "depth": _get_nesting_depth(input_data)
            }
        
        return json.dumps(analysis_result, indent=2, ensure_ascii=False)
        
    except Exception as e:
        return f"Error analyzing data: {str(e)}"

@mcp.tool()
async def get_data_statistics(data: List[float]) -> str:
    """
    Calculate comprehensive statistics for a list of numerical data.
    
    Args:
        data (List[float]): List of numerical values
        
    Returns:
        str: Statistical analysis results
    """
    try:
        if not data:
            return "Error: No data provided"
        
        if not all(isinstance(x, (int, float)) for x in data):
            return "Error: All values must be numerical"
        
        stats = {
            "count": len(data),
            "sum": sum(data),
            "mean": statistics.mean(data),
            "median": statistics.median(data),
            "min": min(data),
            "max": max(data),
            "range": max(data) - min(data),
            "variance": statistics.variance(data) if len(data) > 1 else 0,
            "std_dev": statistics.stdev(data) if len(data) > 1 else 0,
            "quartiles": statistics.quantiles(data, n=4) if len(data) > 1 else []
        }
        
        return json.dumps(stats, indent=2, ensure_ascii=False)
        
    except Exception as e:
        return f"Error calculating statistics: {str(e)}"

@mcp.tool()
async def get_service_statistics() -> str:
    """
    Get statistics and information about the data analysis service.
    
    Returns:
        str: Service statistics and information
    """
    stats = {
        "service_name": "DataAnalyzer",
        "version": "1.0.0",
        "status": "running",
        "port": 8002,
        "endpoints": ["analyze_data", "get_data_statistics", "get_service_statistics"],
        "capabilities": [
            "statistical_analysis",
            "text_analysis",
            "structure_analysis",
            "data_processing"
        ],
        "description": "A comprehensive data analysis service providing statistical calculations and data insights"
    }
    
    return json.dumps(stats, indent=2, ensure_ascii=False)

def _get_nesting_depth(obj, current_depth=0):
    """Helper function to calculate nesting depth of data structures."""
    if isinstance(obj, dict):
        if not obj:
            return current_depth
        return max(_get_nesting_depth(v, current_depth + 1) for v in obj.values())
    elif isinstance(obj, list):
        if not obj:
            return current_depth
        return max(_get_nesting_depth(item, current_depth + 1) for item in obj)
    else:
        return current_depth

if __name__ == "__main__":
    # Start the MCP server with stdio transport
    mcp.run(transport="stdio")