REE MCP Server

Instructions

Get all countries' prices for comparison:
>>> await get_price_analysis("2025-10-08T00:00", "2025-10-08T23:59")

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:693-774 (handler)

  The handler function for the 'get_price_analysis' MCP tool. It fetches SPOT market electricity price data for a given date range, optionally filters by geographic scope, groups prices by country, computes statistics (min, max, avg per country), and returns a formatted JSON response with analysis.

  async def get_price_analysis(
      start_date: str, end_date: str, geo_filter: str | None = None
  ) -> str:
      """Get electricity price analysis over time.
  
      Analyzes SPOT market prices with statistics and multi-country comparison.
      Note: SPOT price indicator returns data for multiple European countries.
      Use geo_filter to focus on a specific market.
  
      Args:
          start_date: Start datetime in ISO format (YYYY-MM-DDTHH:MM)
          end_date: End datetime in ISO format (YYYY-MM-DDTHH:MM)
          geo_filter: Optional geographic filter (e.g., "Península", "Portugal", "France")
                     If not specified, returns all countries
  
      Returns:
          JSON string with price data and analysis.
  
      Examples:
          Get Spanish hourly prices for a day:
          >>> await get_price_analysis("2025-10-08T00:00", "2025-10-08T23:59", "Península")
  
          Get all countries' prices for comparison:
          >>> await get_price_analysis("2025-10-08T00:00", "2025-10-08T23:59")
      """
      try:
          async with ToolExecutor() as executor:
              use_case = executor.create_get_indicator_data_use_case()
  
              # Get SPOT price data
              request = GetIndicatorDataRequest(
                  indicator_id=IndicatorIDs.SPOT_MARKET_PRICE.id,
                  start_date=start_date,
                  end_date=end_date,
                  time_granularity="hour",
              )
              response = await use_case.execute(request)
              price_data = response.model_dump()
  
          values = price_data.get("values", [])
  
          # Filter by geography if requested
          if geo_filter:
              values = [v for v in values if v["geo_scope"] == geo_filter]
  
          # Group by country
          countries: dict[str, list[dict[str, Any]]] = {}
          for value_point in values:
              geo = value_point["geo_scope"]
              if geo not in countries:
                  countries[geo] = []
              countries[geo].append(
                  {
                      "datetime": value_point["datetime"],
                      "price_eur_per_mwh": value_point["value"],
                  }
              )
  
          # Calculate statistics per country
          country_stats = {}
          for country, prices in countries.items():
              price_values = [p["price_eur_per_mwh"] for p in prices]
              if price_values:
                  country_stats[country] = {
                      "min_eur_per_mwh": round(min(price_values), 2),
                      "max_eur_per_mwh": round(max(price_values), 2),
                      "avg_eur_per_mwh": round(sum(price_values) / len(price_values), 2),
                      "count": len(price_values),
                  }
  
          result = {
              "period": {"start": start_date, "end": end_date},
              "countries": countries,
              "statistics_by_country": country_stats,
              "unit": price_data["indicator"]["unit"],
          }
  
          return ResponseFormatter.success(result)
  
      except Exception as e:
          return ResponseFormatter.unexpected_error(e, context="Error analyzing prices")