REE MCP Server

Instructions

Implementation Reference

• src/ree_mcp/interface/tool_services.py:243-312 (handler)

  Core implementation of get_renewable_summary in RenewableAnalysisService. Fetches data for multiple renewable indicators using DataFetcher, aggregates generation values distinguishing variable and synchronous renewables, calculates percentages relative to total demand, and returns structured summary.
  async def get_renewable_summary(self, start_date: str, end_date: str) -> dict[str, Any]: """Get renewable generation summary. Args: start_date: Start datetime in ISO format end_date: End datetime in ISO format Returns: Renewable summary with breakdowns and percentages """ renewable_sources = IndicatorIDs.get_renewable_sources() raw_data = await self.data_fetcher.fetch_multiple_indicators( renewable_sources, start_date, end_date, "hour" ) result: dict[str, Any] = { "datetime": start_date, "renewable_sources": {}, "summary": {}, } total_renewable_mw = 0.0 variable_renewable_mw = 0.0 # Process renewable sources for source_name, response_data in raw_data.items(): if "error" in response_data: result["renewable_sources"][source_name] = response_data else: values = response_data.get("values", []) if values: value_mw = values[0]["value"] is_variable = source_name in [ "wind_national", "solar_pv_national", "solar_thermal_national", ] result["renewable_sources"][source_name] = { "value_mw": value_mw, "type": "variable" if is_variable else "synchronous", } total_renewable_mw += value_mw if is_variable: variable_renewable_mw += value_mw else: result["renewable_sources"][source_name] = {"error": "No data available"} # Get total demand for percentage calculation demand_mw = await self.data_fetcher.fetch_value_at_time( IndicatorIDs.REAL_DEMAND_NATIONAL, start_date, end_date, "hour" ) if demand_mw and demand_mw > 0: renewable_pct = (total_renewable_mw / demand_mw) * 100 variable_pct = (variable_renewable_mw / demand_mw) * 100 result["summary"] = { "total_renewable_mw": round(total_renewable_mw, 2), "variable_renewable_mw": round(variable_renewable_mw, 2), "synchronous_renewable_mw": round(total_renewable_mw - variable_renewable_mw, 2), "total_demand_mw": round(demand_mw, 2), "renewable_percentage": round(renewable_pct, 2), "variable_renewable_percentage": round(variable_pct, 2), } else: result["summary"] = {"error": "Could not calculate percentages: No demand data"} return result
• src/ree_mcp/interface/mcp_server.py:278-312 (registration)

  MCP tool registration using @mcp.tool() decorator. Provides user-friendly interface with date and hour parameters, constructs datetime range using DateTimeHelper, instantiates RenewableAnalysisService via ToolExecutor, calls the core handler, and formats response as JSON string.
  @mcp.tool() async def get_renewable_summary(date: str, hour: str = "12") -> str: """Get renewable energy generation summary at a specific time. Aggregates wind, solar PV, solar thermal, and hydro generation with renewable percentage calculations. Args: date: Date in YYYY-MM-DD format hour: Hour in HH format (00-23, default: 12) Returns: JSON string with renewable generation breakdown and percentages. Examples: Get renewable summary at noon: >>> await get_renewable_summary("2025-10-08", "12") Get overnight renewable summary: >>> await get_renewable_summary("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 = RenewableAnalysisService(data_fetcher) result = await service.get_renewable_summary(start_datetime, end_datetime) return ResponseFormatter.success(result) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error getting renewable summary")