REE MCP Server

Instructions

Get overnight generation mix:
>>> await get_generation_mix("2025-10-08", "02")

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:204-239 (handler)

  Primary MCP tool handler and registration for 'get_generation_mix'. Processes date/hour inputs, orchestrates data fetching via GenerationMixService, and returns formatted JSON response.

  @mcp.tool()
  async def get_generation_mix(date: str, hour: str = "12") -> str:
      """Get the electricity generation mix at a specific time.
  
      Returns the power generation breakdown by source (nuclear, wind, solar, etc.)
      for a specific hour.
  
      Args:
          date: Date in YYYY-MM-DD format
          hour: Hour in HH format (00-23, default: 12)
  
      Returns:
          JSON string with generation mix by source.
  
      Examples:
          Get generation mix at noon on Oct 8:
          >>> await get_generation_mix("2025-10-08", "12")
  
          Get overnight generation mix:
          >>> await get_generation_mix("2025-10-08", "02")
      """
      try:
          start_datetime, end_datetime = DateTimeHelper.build_datetime_range(date, hour)
  
          async with ToolExecutor() as executor:
              use_case = executor.create_get_indicator_data_use_case()
              data_fetcher = DataFetcher(use_case)
              service = GenerationMixService(data_fetcher)
  
              result = await service.get_generation_mix(start_datetime, end_datetime)
  
          return ResponseFormatter.success(result)
  
      except Exception as e:
          return ResponseFormatter.unexpected_error(e, context="Error getting generation mix")

• src/ree_mcp/interface/tool_services.py:133-167 (helper)

  Core logic in GenerationMixService.get_generation_mix that fetches parallel data for generation sources defined in IndicatorIDs.get_generation_mix_sources() and constructs the mix dictionary from the first value of each.

  async def get_generation_mix(self, start_date: str, end_date: str) -> dict[str, Any]:
      """Get generation mix for a specific time.
  
      Args:
          start_date: Start datetime in ISO format
          end_date: End datetime in ISO format
  
      Returns:
          Generation mix data with sources and values
      """
      sources = IndicatorIDs.get_generation_mix_sources()
      raw_data = await self.data_fetcher.fetch_multiple_indicators(
          sources, start_date, end_date, "hour"
      )
  
      generation_mix: dict[str, Any] = {
          "datetime": start_date,
          "sources": {},
      }
  
      for source_name, response_data in raw_data.items():
          if "error" in response_data:
              generation_mix["sources"][source_name] = response_data
          else:
              values = response_data.get("values", [])
              if values:
                  generation_mix["sources"][source_name] = {
                      "value_mw": values[0]["value"],
                      "unit": response_data["indicator"]["unit"],
                  }
              else:
                  generation_mix["sources"][source_name] = {"error": "No data available"}
  
      return generation_mix

• src/ree_mcp/interface/indicator_config.py:253-266 (helper)

  IndicatorIDs.get_generation_mix_sources() class method providing the mapping of generation source names to specific REE indicator metadata used by the service.

  def get_generation_mix_sources(cls) -> dict[str, IndicatorMetadata]:
      """Get indicators for generation mix analysis.
  
      Returns:
          Dictionary mapping source names to indicator metadata.
      """
      return {
          "nuclear": cls.NUCLEAR,
          "wind_national": cls.WIND_NATIONAL,
          "solar_pv_peninsular": cls.SOLAR_PV_PENINSULAR,
          "solar_thermal_peninsular": cls.SOLAR_THERMAL_PENINSULAR,
          "hydro_national": cls.HYDRO_NATIONAL,
          "combined_cycle_national": cls.COMBINED_CYCLE_NATIONAL,
      }