REE MCP Server

Instructions

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:1040-1075 (handler)

  MCP tool handler for 'get_daily_demand_statistics', decorated with @mcp.tool() which registers it. Orchestrates execution by setting up data fetcher and DemandAnalysisService, calls the service method, and formats the response.
  @mcp.tool() async def get_daily_demand_statistics(start_date: str, end_date: str) -> str: """Get daily demand statistics for a period. Provides comprehensive daily demand analysis including maximum, minimum, and sum of generation values for each day in the specified period. Args: start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format Returns: JSON string with daily statistics and overall summary. Examples: Get statistics for a week: >>> await get_daily_demand_statistics("2025-10-01", "2025-10-07") Get statistics for a month: >>> await get_daily_demand_statistics("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.get_daily_demand_statistics(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 getting demand statistics")
• src/ree_mcp/interface/tool_services.py:509-587 (helper)

  Core helper method implementing the daily demand statistics logic within DemandAnalysisService. Fetches max daily demand, min daily demand, and sum generation data, computes per-day stats (max, min, load factor, swing), and overall summary statistics.
  async def get_daily_demand_statistics(self, start_date: str, end_date: str) -> dict[str, Any]: """Get daily demand statistics. Args: start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format Returns: Daily demand statistics with max, min, and sum of generation """ # Fetch all three demand indicators indicators = { "max_daily": IndicatorIDs.MAX_DAILY_DEMAND, "min_daily": IndicatorIDs.MIN_DAILY_DEMAND, "sum_generation": IndicatorIDs.REAL_DEMAND_SUM_GENERATION, } 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_statistics": [], } # Extract values by date max_values = raw_data.get("max_daily", {}).get("values", []) min_values = raw_data.get("min_daily", {}).get("values", []) sum_values = raw_data.get("sum_generation", {}).get("values", []) # Combine data by date for i, max_point in enumerate(max_values): date = max_point["datetime"][:10] # Extract YYYY-MM-DD max_mw = max_point["value"] min_mw = min_values[i]["value"] if i < len(min_values) else None sum_mw = sum_values[i]["value"] if i < len(sum_values) else None daily_stat: dict[str, Any] = { "date": date, "max_demand_mw": max_mw, "min_demand_mw": min_mw, "sum_generation_mw": sum_mw, } # Calculate load factor if we have both max and min if min_mw is not None and max_mw > 0: daily_stat["load_factor"] = round((min_mw / max_mw) * 100, 2) daily_stat["daily_swing_mw"] = round(max_mw - min_mw, 2) else: daily_stat["load_factor"] = None daily_stat["daily_swing_mw"] = None result["daily_statistics"].append(daily_stat) # Calculate overall statistics if result["daily_statistics"]: max_values_list = [s["max_demand_mw"] for s in result["daily_statistics"]] min_values_list = [ s["min_demand_mw"] for s in result["daily_statistics"] if s["min_demand_mw"] is not None ] load_factors = [ s["load_factor"] for s in result["daily_statistics"] if s["load_factor"] is not None ] result["summary"] = { "peak_demand_mw": round(max(max_values_list), 2), "lowest_demand_mw": round(min(min_values_list), 2) if min_values_list else None, "average_max_demand_mw": round(sum(max_values_list) / len(max_values_list), 2), "average_load_factor_pct": ( round(sum(load_factors) / len(load_factors), 2) if load_factors else None ), "days_analyzed": len(result["daily_statistics"]), } return result