Spaces:
Sleeping
Sleeping
| 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, | |
| ) | |
| 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)}" | |
| 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)}" | |
| 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") |