REE MCP Server

Instructions

Get 5-minute wind generation data:
>>> await get_indicator_data(2038, "2025-10-08T00:00", "2025-10-08T03:00", "raw")

Implementation Reference

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

  The main MCP tool handler for 'get_indicator_data', decorated with @mcp.tool() which registers it. Handles input parameters, creates request DTO, executes the use case via ToolExecutor, and returns JSON response.

  @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/dtos/indicator_request.py:8-44 (schema)

  Pydantic model defining the input schema for GetIndicatorDataRequest with validation for fields, datetimes, and time granularity.

  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 model defining the output schema for indicator data response, including metadata, time-series values, and statistics. Converts from domain entity.

  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/application/use_cases/get_indicator_data.py:8-52 (helper)

  Core business logic use case that validates request, fetches data from repository using domain objects, and converts to response 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/interface/tool_helpers.py:208-215 (helper)

  ToolExecutor method that creates and returns the GetIndicatorDataUseCase instance with the repository dependency injected.

  def create_get_indicator_data_use_case(self) -> GetIndicatorDataUseCase:
      """Create a GetIndicatorDataUseCase instance.
  
      Returns:
          Configured use case instance
      """
      return GetIndicatorDataUseCase(self.repository)