get_historical_data
Retrieve historical meteorological data for a specific station using its identifier and a date range.
Instructions
Obtain historical meteorological data for a specific station.
Args: station_id: Identifier of the station (e.g. "3195" for Madrid Retiro) start_date: Start date in format YYYYY-MM-DD end_date: End date in format YYYYY-MM-DD
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| station_id | Yes | ||
| start_date | Yes | ||
| end_date | Yes |
Implementation Reference
- src/aemet_mcp/server.py:248-261 (handler)The actual handler function for the get_historical_data tool. It constructs the AEMET API URL for daily climatological data and delegates to make_aemet_request for the HTTP call and response parsing.
@mcp.tool() async def get_historical_data(station_id: str, start_date: str, end_date: str): """Obtain historical meteorological data for a specific station. Args: station_id: Identifier of the station (e.g. "3195" for Madrid Retiro) start_date: Start date in format YYYYY-MM-DD end_date: End date in format YYYYY-MM-DD """ start = start_date + "T00:00:00UTC" end = end_date + "T23:59:59UTC" url = f"{AEMET_API_BASE}/valores/climatologicos/diarios/datos/fechaini/{start}/fechafin/{end}/estacion/{station_id}" return await make_aemet_request(url) - src/aemet_mcp/server.py:248-248 (registration)The tool is registered via the @mcp.tool() decorator on line 248, which is the FastMCP framework's way of registering an MCP tool.
@mcp.tool() - src/aemet_mcp/server.py:55-76 (helper)The make_aemet_request helper function that performs the HTTP request to the AEMET API using an API key, handles the redirect to the actual data URL, and decodes the response from latin1.
async def make_aemet_request(url: str) -> dict[str, Any] | list[Any] | None: logger.info(f"make_aemet_request") headers = { "api_key": API_KEY, "Accept": "application/json" } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() data_info = response.json() if data_info.get("estado") == 200: data_url = data_info.get("datos") if data_url: data_response = await client.get(data_url, timeout=30.0) data_response.raise_for_status() content = data_response.content.decode('latin1') return json.loads(content) return None except Exception as e: logger.error(f"Error connecting to AEMET: {str(e)}") return None