REE MCP Server

Instructions

Get peak analysis for a month:
>>> await get_peak_analysis("2025-10-01", "2025-10-31")

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:874-973 (handler)

  The core handler function implementing the get_peak_analysis MCP tool. Decorated with @mcp.tool() for automatic registration in FastMCP. Fetches daily max and min demand data from REE API using specific indicator IDs, processes it to compute daily peak-to-valley differences, load factors, and period-wide statistics. Returns formatted JSON with analysis results and interpretations.

  @mcp.tool()
  async def get_peak_analysis(start_date: str, end_date: str) -> str:
      """Get peak demand analysis over a period.
  
      Analyzes daily maximum and minimum demand to identify patterns and
      calculate load factors.
  
      Args:
          start_date: Start date in YYYY-MM-DD format
          end_date: End date in YYYY-MM-DD format
  
      Returns:
          JSON string with peak demand analysis.
  
      Examples:
          Get peak analysis for a week:
          >>> await get_peak_analysis("2025-10-01", "2025-10-07")
  
          Get peak analysis for a month:
          >>> await get_peak_analysis("2025-10-01", "2025-10-31")
      """
      try:
          async with ToolExecutor() as executor:
              use_case = executor.create_get_indicator_data_use_case()
  
              # Get max demand
              max_request = GetIndicatorDataRequest(
                  indicator_id=IndicatorIDs.MAX_DAILY_DEMAND.id,
                  start_date=start_date,
                  end_date=end_date,
                  time_granularity="day",
              )
              max_response = await use_case.execute(max_request)
              max_data = max_response.model_dump()
  
              # Get min demand
              min_request = GetIndicatorDataRequest(
                  indicator_id=IndicatorIDs.MIN_DAILY_DEMAND.id,
                  start_date=start_date,
                  end_date=end_date,
                  time_granularity="day",
              )
              min_response = await use_case.execute(min_request)
              min_data = min_response.model_dump()
  
          # Combine data
          max_values = max_data.get("values", [])
          min_values = min_data.get("values", [])
  
          daily_analysis = []
          peak_demands = []
          load_factors = []
  
          for max_val, min_val in zip(max_values, min_values, strict=False):
              max_mw = max_val["value"]
              min_mw = min_val["value"]
              avg_mw = (max_mw + min_mw) / 2
              peak_to_valley = max_mw - min_mw
              load_factor = (avg_mw / max_mw * 100) if max_mw > 0 else 0
  
              daily_analysis.append(
                  {
                      "date": max_val["datetime"][:10],
                      "peak_demand_mw": max_mw,
                      "minimum_demand_mw": min_mw,
                      "average_demand_mw": round(avg_mw, 2),
                      "peak_to_valley_mw": round(peak_to_valley, 2),
                      "load_factor_percentage": round(load_factor, 2),
                  }
              )
  
              peak_demands.append(max_mw)
              load_factors.append(load_factor)
  
          # Calculate period statistics
          period_stats = {}
          if peak_demands:
              period_stats = {
                  "highest_peak_mw": max(peak_demands),
                  "lowest_peak_mw": min(peak_demands),
                  "average_peak_mw": round(sum(peak_demands) / len(peak_demands), 2),
                  "average_load_factor_percentage": round(sum(load_factors) / len(load_factors), 2),
                  "interpretation": {
                      "high_load_factor": "> 70% (efficient, stable demand)",
                      "medium_load_factor": "50-70% (moderate variability)",
                      "low_load_factor": "< 50% (high variability, inefficient)",
                  },
              }
  
          result = {
              "period": {"start": start_date, "end": end_date},
              "daily_analysis": daily_analysis,
              "period_statistics": period_stats,
          }
  
          return ResponseFormatter.success(result)
  
      except Exception as e:
          return ResponseFormatter.unexpected_error(e, context="Error analyzing peaks")