IPMA Weather MCP Server

Instructions

Implementation Reference

• weather.py:383-450 (handler)

  Core handler function for the get_station_observations MCP tool. Decorated with @mcp.tool() for automatic registration. Fetches and formats recent meteorological observations from IPMA weather stations, supports listing stations or querying specific ones.
  @mcp.tool() async def get_station_observations(station_id: str = "") -> str: """Get meteorological observations from weather stations (Observações últimas 24 horas). Args: station_id: ID of the weather station. Leave empty to see all stations. Returns hourly meteorological observations from the last 24 hours including: - Temperature - Humidity - Pressure - Wind speed and direction - Precipitation - Radiation """ # Get observations obs_url = f"{IPMA_API_BASE}/observation/meteorology/stations/observations.json" obs_data = await make_ipma_request(obs_url) if not obs_data: return "Unable to fetch meteorological observations." # Get station information stations_url = f"{IPMA_API_BASE}/observation/meteorology/stations/stations.json" stations_data = await make_ipma_request(stations_url) # Create station mapping station_map = {} if stations_data and isinstance(stations_data, list): for station in stations_data: props = station.get('properties', {}) station_map[str(props.get('idEstacao'))] = props.get('localEstacao', 'Unknown') if not station_id: # List available stations result = f"Available Weather Stations ({len(station_map)} total):\n\n" count = 0 for sid, name in list(station_map.items())[:20]: result += f"ID: {sid} - {name}\n" count += 1 result += f"\n... and {len(station_map) - count} more stations.\n" result += "\nUse station_id parameter to get observations for a specific station." return result # Get observations for specific station result = f"Meteorological Observations for Station {station_id}\n" result += f"Station Name: {station_map.get(station_id, 'Unknown')}\n\n" observations_found = False for timestamp, stations in obs_data.items(): if station_id in stations: observations_found = True obs = stations[station_id] result += f"""Time: {timestamp} UTC Temperature: {obs.get('temperatura', 'N/A')}°C Humidity: {obs.get('humidade', 'N/A')}% Pressure: {obs.get('pressao', 'N/A')} hPa Wind Speed: {obs.get('intensidadeVento', 'N/A')} m/s ({obs.get('intensidadeVentoKM', 'N/A')} km/h) Wind Direction: {obs.get('idDireccVento', 'N/A')} Accumulated Precipitation: {obs.get('precAcumulada', 'N/A')} mm Radiation: {obs.get('radiacao', 'N/A')} W/m² --- """ if not observations_found: result += f"No observations found for station {station_id} in the last 24 hours." return result
• weather.py:12-21 (helper)

  Utility helper function used by the tool to perform HTTP requests to the IPMA API endpoints, handling errors and returning JSON data or None.
  async def make_ipma_request(url: str) -> dict[str, Any] | list[dict[str, Any]] | None: """Make a request to the IPMA API with proper error handling.""" async with httpx.AsyncClient() as client: try: response = await client.get(url, timeout=30.0) response.raise_for_status() return response.json() except Exception: return None
• weather.py:527-532 (registration)

  Server entry point that runs the FastMCP server instance, automatically registering and exposing all @mcp.tool()-decorated functions including get_station_observations.
  def main(): # Initialize and run the server mcp.run(transport='stdio') if __name__ == "__main__": main()
• weather.py:10-10 (helper)

  Constant defining the base URL for IPMA open data API, used in constructing endpoints for the tool.
  IPMA_API_BASE = "https://api.ipma.pt/open-data"