REE MCP Server

Instructions

Implementation Reference

• src/ree_mcp/interface/tool_services.py:444-493 (handler)

  Core handler logic in InternationalExchangeService.get_international_exchanges: fetches import/export data using DataFetcher for Andorra, Morocco, Portugal, France; computes net balances and totals.
  async def get_international_exchanges(self, start_date: str, end_date: str) -> dict[str, Any]: """Get international electricity exchanges. Args: start_date: Start datetime in ISO format end_date: End datetime in ISO format Returns: Exchange data by country with net balance """ exchanges = IndicatorIDs.get_international_exchanges() result: dict[str, Any] = { "datetime": start_date, "exchanges": {}, "totals": {"total_exports_mw": 0.0, "total_imports_mw": 0.0, "net_balance_mw": 0.0}, } for country, indicators in exchanges.items(): # Fetch export and import data export_mw = await self.data_fetcher.fetch_value_at_time( indicators["export"], start_date, end_date, "hour" ) import_mw = await self.data_fetcher.fetch_value_at_time( indicators["import"], start_date, end_date, "hour" ) if export_mw is not None and import_mw is not None: net_mw = import_mw - export_mw result["exchanges"][country] = { "export_mw": export_mw, "import_mw": import_mw, "net_balance_mw": net_mw, "net_flow": ( "import" if net_mw > 0 else "export" if net_mw < 0 else "balanced" ), } result["totals"]["total_exports_mw"] += export_mw result["totals"]["total_imports_mw"] += import_mw else: result["exchanges"][country] = {"error": "Could not fetch exchange data"} result["totals"]["net_balance_mw"] = ( result["totals"]["total_imports_mw"] - result["totals"]["total_exports_mw"] ) return result
• src/ree_mcp/interface/mcp_server.py:241-275 (registration)

  MCP tool registration via @mcp.tool() decorator. Wrapper function that builds datetime range, instantiates InternationalExchangeService, calls it, and formats response as JSON.
  @mcp.tool() async def get_international_exchanges(date: str, hour: str = "12") -> str: """Get international electricity exchanges at a specific time. Returns import/export data by country (Andorra, Morocco, Portugal, France) with net balance calculations. Args: date: Date in YYYY-MM-DD format hour: Hour in HH format (00-23, default: 12) Returns: JSON string with imports, exports, and net balance by country. Examples: Get exchanges at noon on Oct 8: >>> await get_international_exchanges("2025-10-08", "12") Get overnight exchanges: >>> await get_international_exchanges("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 = InternationalExchangeService(data_fetcher) result = await service.get_international_exchanges(start_datetime, end_datetime) return ResponseFormatter.success(result) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error getting exchanges")
• src/ree_mcp/interface/indicator_config.py:310-335 (helper)

  IndicatorIDs.get_international_exchanges(): Returns configuration dict of IndicatorMetadata for export/import pairs per country (Andorra, Morocco, Portugal, France), used by the service to know which indicators to fetch.
  @classmethod def get_international_exchanges(cls) -> dict[str, dict[str, IndicatorMetadata]]: """Get international exchange indicators by country. Returns: Dictionary mapping country names to export/import indicators. """ return { "andorra": { "export": cls.EXPORT_ANDORRA, "import": cls.IMPORT_ANDORRA, }, "morocco": { "export": cls.EXPORT_MOROCCO, "import": cls.IMPORT_MOROCCO, }, "portugal": { "export": cls.EXPORT_PORTUGAL, "import": cls.IMPORT_PORTUGAL, }, "france": { "export": cls.EXPORT_FRANCE, "import": cls.IMPORT_FRANCE, }, }