REE MCP Server

Instructions

Find price indicators:
>>> await search_indicators("precio")

Find solar generation indicators:
>>> await search_indicators("solar")

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:117-155 (handler)

  MCP tool handler and registration for 'search_indicators'. Uses ToolExecutor to create use case, executes search, formats results as JSON or error response.

  @mcp.tool()
  async def search_indicators(keyword: str, limit: int | None = 20) -> str:
      """Search for indicators by keyword in their names.
  
      Searches through all available indicators and returns those matching
      the keyword in their name or short name.
  
      Args:
          keyword: Keyword to search for (e.g., "demanda", "precio", "solar")
          limit: Maximum number of results (default: 20)
  
      Returns:
          JSON string with matching indicator metadata.
  
      Examples:
          Find all demand-related indicators:
          >>> await search_indicators("demanda", limit=10)
  
          Find price indicators:
          >>> await search_indicators("precio")
  
          Find solar generation indicators:
          >>> await search_indicators("solar")
      """
      try:
          async with ToolExecutor() as executor:
              use_case = executor.create_search_indicators_use_case()
              indicators = await use_case.execute(keyword=keyword, limit=limit)
  
          result = {
              "keyword": keyword,
              "count": len(indicators),
              "indicators": [ind.model_dump() for ind in indicators],
          }
          return ResponseFormatter.success(result, ensure_ascii=False)
  
      except Exception as e:
          return ResponseFormatter.unexpected_error(e, context="Error searching indicators")

• src/ree_mcp/application/use_cases/search_indicators.py:7-48 (helper)

  SearchIndicatorsUseCase: core business logic that calls repository.search_indicators() and maps results to IndicatorMetadataResponse DTOs.

  class SearchIndicatorsUseCase:
      """Use case for searching indicators by keyword.
  
      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, keyword: str, limit: int | None = None
      ) -> list[IndicatorMetadataResponse]:
          """Execute the use case.
  
          Args:
              keyword: Keyword to search for
              limit: Maximum number of results
  
          Returns:
              List of matching indicator metadata.
          """
          indicators = await self.repository.search_indicators(keyword=keyword, limit=limit)
  
          return [
              IndicatorMetadataResponse(
                  id=int(ind.id),
                  name=ind.name,
                  short_name=ind.short_name,
                  description=ind.description,
                  unit=ind.unit.value,
                  frequency=ind.frequency,
                  geo_scope=ind.geo_scope.value,
              )
              for ind in indicators
          ]

• src/ree_mcp/interface/tool_helpers.py:224-231 (helper)

  ToolExecutor.create_search_indicators_use_case(): factory method to instantiate the use case with injected repository.

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

• src/ree_mcp/interface/mcp_server.py:29-29 (registration)

  FastMCP server initialization where all @mcp.tool() decorators register the tools including search_indicators.

  mcp = FastMCP("REE MCP Server", dependencies=["httpx", "pydantic", "pydantic-settings"])