REE MCP Server

Instructions

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:33-79 (handler)

  The primary MCP tool handler for 'get_indicator_data'. Decorated with @mcp.tool(), it handles input parameters, creates the GetIndicatorDataRequest DTO, executes the use case via ToolExecutor, and returns formatted JSON response or error.
  @mcp.tool() async def get_indicator_data( indicator_id: int, start_date: str, end_date: str, time_granularity: str = "raw", ) -> str: """Get time-series data for a specific electricity indicator. Retrieves historical data for any REE indicator (demand, generation, prices, etc.) for a specified date range. Returns the data with statistical summary. Args: indicator_id: The indicator ID (e.g., 1293 for real demand, 549 for nuclear) start_date: Start datetime in ISO format (YYYY-MM-DDTHH:MM) end_date: End datetime in ISO format (YYYY-MM-DDTHH:MM) time_granularity: Time aggregation level (raw, hour, day, fifteen_minutes) Returns: JSON string with indicator metadata, time-series values, and statistics. Examples: Get hourly real demand for Oct 8, 2025: >>> await get_indicator_data(1293, "2025-10-08T00:00", "2025-10-08T23:59", "hour") Get 5-minute wind generation data: >>> await get_indicator_data(2038, "2025-10-08T00:00", "2025-10-08T03:00", "raw") """ try: request = GetIndicatorDataRequest( indicator_id=indicator_id, start_date=start_date, end_date=end_date, time_granularity=time_granularity, ) async with ToolExecutor() as executor: use_case = executor.create_get_indicator_data_use_case() response = await use_case.execute(request) return response.model_dump_json(indent=2) except DomainException as e: return ResponseFormatter.domain_exception(e) except Exception as e: return ResponseFormatter.unexpected_error(e)
• src/ree_mcp/application/use_cases/get_indicator_data.py:8-52 (handler)

  Core business logic handler implementing GetIndicatorDataUseCase.execute(). Converts input DTO to domain value objects, fetches data from repository, and maps to output DTO.
  class GetIndicatorDataUseCase: """Use case for retrieving indicator time-series data. Attributes: repository: Indicator repository implementation """ def __init__(self, repository: IndicatorRepository) -> None: """Initialize use case. Args: repository: Indicator repository """ self.repository = repository async def execute(self, request: GetIndicatorDataRequest) -> IndicatorDataResponse: """Execute the use case. Args: request: Request with indicator ID and date range Returns: Response with indicator data and statistics. Raises: InvalidIndicatorIdError: If indicator ID is invalid InvalidDateRangeError: If date range is invalid IndicatorNotFoundError: If indicator doesn't exist NoDataAvailableError: If no data available """ # Convert request to domain objects indicator_id = IndicatorId(request.indicator_id) date_range = DateTimeRange.from_iso_strings(request.start_date, request.end_date) time_granularity = TimeGranularity(request.time_granularity) # Get data from repository indicator_data = await self.repository.get_indicator_data( indicator_id=indicator_id, date_range=date_range, time_granularity=time_granularity, ) # Convert to response DTO return IndicatorDataResponse.from_domain(indicator_data)
• src/ree_mcp/application/dtos/indicator_request.py:8-44 (schema)

  Pydantic input schema (DTO) for get_indicator_data with validation for indicator_id, dates (ISO format), and time_granularity enum.
  class GetIndicatorDataRequest(BaseModel): """Request to get indicator data. Attributes: indicator_id: The indicator ID start_date: Start datetime (ISO format) end_date: End datetime (ISO format) time_granularity: Time aggregation (raw, hour, day, fifteen_minutes) """ indicator_id: int = Field(..., gt=0, description="Indicator ID (must be positive)") start_date: str = Field(..., description="Start datetime in ISO format") end_date: str = Field(..., description="End datetime in ISO format") time_granularity: str = Field( default="raw", description="Time granularity (raw, hour, day, fifteen_minutes)", ) @field_validator("time_granularity") @classmethod def validate_time_granularity(cls, v: str) -> str: """Validate time granularity.""" allowed = {"raw", "hour", "day", "fifteen_minutes"} if v not in allowed: raise ValueError(f"time_granularity must be one of {allowed}") return v @field_validator("start_date", "end_date") @classmethod def validate_datetime(cls, v: str) -> str: """Validate datetime format.""" try: datetime.fromisoformat(v.replace("Z", "+00:00")) except (ValueError, AttributeError) as e: raise ValueError(f"Invalid datetime format: {e}") from e return v
• src/ree_mcp/application/dtos/indicator_response.py:46-103 (schema)

  Pydantic output schema (DTO) for get_indicator_data response, including indicator metadata, list of values, and computed statistics. Includes from_domain factory method.
  class IndicatorDataResponse(BaseModel): """Complete indicator data response. Attributes: indicator: Indicator metadata values: List of time-series values statistics: Optional statistics about the data """ indicator: IndicatorMetadataResponse values: list[IndicatorValueResponse] statistics: dict[str, float | None] = Field( default_factory=dict, description="Statistical summary of values" ) @classmethod def from_domain(cls, data: IndicatorData) -> "IndicatorDataResponse": """Create response from domain entity. Args: data: Domain IndicatorData entity Returns: IndicatorDataResponse instance. """ indicator_response = IndicatorMetadataResponse( id=int(data.indicator.id), name=data.indicator.name, short_name=data.indicator.short_name, description=data.indicator.description, unit=data.indicator.unit.value, frequency=data.indicator.frequency, geo_scope=data.indicator.geo_scope.value, ) values_response = [ IndicatorValueResponse( value=val.value, datetime=val.datetime.isoformat(), datetime_utc=val.datetime_utc.isoformat(), geo_scope=val.geo_scope.value, ) for val in data.values ] statistics = { "count": len(data.values), "min": data.min_value(), "max": data.max_value(), "avg": data.avg_value(), } return cls( indicator=indicator_response, values=values_response, statistics=statistics, )
• src/ree_mcp/interface/tool_helpers.py:208-215 (helper)

  ToolExecutor helper method to create and return the GetIndicatorDataUseCase instance with injected repository dependency.
  def create_get_indicator_data_use_case(self) -> GetIndicatorDataUseCase: """Create a GetIndicatorDataUseCase instance. Returns: Configured use case instance """ return GetIndicatorDataUseCase(self.repository)