REE MCP Server

Instructions

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:1077-1112 (handler)

  MCP tool handler decorated with @mcp.tool(). Instantiates DemandAnalysisService and calls its analyze_demand_volatility method. Handles exceptions and formats response as JSON string.
  @mcp.tool() async def analyze_demand_volatility(start_date: str, end_date: str) -> str: """Analyze demand volatility patterns over a period. Calculates daily demand swings, load factors, and volatility levels to identify high-variability days and overall stability patterns. Args: start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format Returns: JSON string with volatility analysis and stability assessment. Examples: Analyze volatility for a week: >>> await analyze_demand_volatility("2025-10-01", "2025-10-07") Analyze volatility for a month: >>> await analyze_demand_volatility("2025-10-01", "2025-10-31") """ try: async with ToolExecutor() as executor: use_case = executor.create_get_indicator_data_use_case() data_fetcher = DataFetcher(use_case) service = DemandAnalysisService(data_fetcher) result = await service.analyze_demand_volatility(start_date, end_date) return ResponseFormatter.success(result) except DomainException as e: return ResponseFormatter.domain_exception(e) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error analyzing demand volatility")
• src/ree_mcp/interface/tool_services.py:589-696 (helper)

  Main business logic for analyzing demand volatility within DemandAnalysisService class. Fetches daily max and min demand data, computes daily swings, percentages, classifies volatility levels, and provides overall analysis including stability assessment.
  async def analyze_demand_volatility(self, start_date: str, end_date: str) -> dict[str, Any]: """Analyze demand volatility patterns. Args: start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format Returns: Volatility analysis with swings and stability metrics """ # Fetch max and min daily demand indicators = { "max_daily": IndicatorIDs.MAX_DAILY_DEMAND, "min_daily": IndicatorIDs.MIN_DAILY_DEMAND, } raw_data = await self.data_fetcher.fetch_multiple_indicators( indicators, start_date, end_date, "day" ) result: dict[str, Any] = { "period": {"start": start_date, "end": end_date}, "daily_volatility": [], "analysis": {}, } max_values = raw_data.get("max_daily", {}).get("values", []) min_values = raw_data.get("min_daily", {}).get("values", []) daily_swings = [] load_factors = [] # Calculate daily volatility for i, max_point in enumerate(max_values): if i >= len(min_values): break date = max_point["datetime"][:10] max_mw = max_point["value"] min_mw = min_values[i]["value"] swing_mw = max_mw - min_mw swing_pct = (swing_mw / max_mw) * 100 if max_mw > 0 else 0 load_factor = (min_mw / max_mw) * 100 if max_mw > 0 else 0 volatility_data = { "date": date, "max_demand_mw": round(max_mw, 2), "min_demand_mw": round(min_mw, 2), "daily_swing_mw": round(swing_mw, 2), "swing_percentage": round(swing_pct, 2), "load_factor_pct": round(load_factor, 2), } # Volatility classification if swing_pct < 20: volatility_data["volatility_level"] = "low" elif swing_pct < 40: volatility_data["volatility_level"] = "moderate" elif swing_pct < 60: volatility_data["volatility_level"] = "high" else: volatility_data["volatility_level"] = "very_high" result["daily_volatility"].append(volatility_data) daily_swings.append(swing_mw) load_factors.append(load_factor) # Overall analysis if daily_swings: avg_swing = sum(daily_swings) / len(daily_swings) max_swing = max(daily_swings) min_swing = min(daily_swings) avg_load_factor = sum(load_factors) / len(load_factors) # Count days by volatility level volatility_counts = {"low": 0, "moderate": 0, "high": 0, "very_high": 0} for day_data in result["daily_volatility"]: level = day_data["volatility_level"] volatility_counts[level] += 1 result["analysis"] = { "average_daily_swing_mw": round(avg_swing, 2), "max_daily_swing_mw": round(max_swing, 2), "min_daily_swing_mw": round(min_swing, 2), "average_load_factor_pct": round(avg_load_factor, 2), "volatility_distribution": volatility_counts, "stability_assessment": ( "excellent" if avg_load_factor >= 80 else "good" if avg_load_factor >= 70 else "moderate" if avg_load_factor >= 60 else "concerning" ), "interpretation": { "load_factor": "Higher load factor (closer to 100%) indicates more stable demand", "volatility_levels": { "low": "<20% swing", "moderate": "20-40% swing", "high": "40-60% swing", "very_high": ">60% swing", }, }, } return result