search_indicators
Search for electricity grid indicators by keyword to find demand, generation, price, or emissions data from Spain's electrical system.
Instructions
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")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| keyword | Yes | ||
| limit | No |
Implementation Reference
- 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")
- 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 ]
- 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"])