Spaces:
Sleeping
Sleeping
File size: 5,366 Bytes
2de095a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
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") |