REE MCP Server

Instructions

Get statistics for a month:
>>> await get_daily_demand_statistics("2025-10-01", "2025-10-31")

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