Weather MCP Server

Instructions

Implementation Reference

• weather.py:171-294 (handler)

  The @mcp.tool()-decorated handler function that implements the core logic for fetching and formatting the weather forecast from the National Weather Service (NWS) API using latitude and longitude coordinates.
  @mcp.tool() async def get_forecast(latitude: float, longitude: float, ctx: Context[ServerSession, None]) -> str: """Get weather forecast for a location. Args: latitude: Latitude of the location longitude: Longitude of the location """ await ctx.info(f"Fetching weather forecast for coordinates: {latitude}, {longitude}") logger.info(f"Processing forecast request for coordinates: {latitude}, {longitude}") # Validate coordinates if not (-90 <= latitude <= 90): error_msg = f"Invalid latitude: {latitude}. Must be between -90 and 90." await ctx.warning(error_msg) logger.warning(error_msg) return "Error: Latitude must be between -90 and 90 degrees." if not (-180 <= longitude <= 180): error_msg = f"Invalid longitude: {longitude}. Must be between -180 and 180." await ctx.warning(error_msg) logger.warning(error_msg) return "Error: Longitude must be between -180 and 180 degrees." # Step 1: Get the forecast grid endpoint await ctx.debug("Step 1: Getting forecast grid endpoint from NWS points API") points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}" await ctx.report_progress( progress=0.3, total=1.0, message="Getting grid information..." ) points_data = await make_nws_request(points_url, ctx) if not points_data: error_msg = f"Unable to fetch forecast grid data for coordinates: {latitude}, {longitude}" await ctx.error(error_msg) logger.error(error_msg) return "Unable to fetch forecast data for this location." try: # Get the forecast URL from the points response forecast_url = points_data["properties"]["forecast"] await ctx.debug(f"Retrieved forecast URL: {forecast_url}") logger.debug(f"Forecast URL: {forecast_url}") except KeyError: error_msg = f"Invalid points response format for coordinates: {latitude}, {longitude}" await ctx.error(error_msg) logger.error(error_msg) return "Invalid response format from weather service points API." # Step 2: Get the detailed forecast await ctx.debug("Step 2: Getting detailed forecast data") await ctx.report_progress( progress=0.6, total=1.0, message="Fetching detailed forecast..." ) forecast_data = await make_nws_request(forecast_url, ctx) if not forecast_data: error_msg = f"Unable to fetch detailed forecast for coordinates: {latitude}, {longitude}" await ctx.error(error_msg) logger.error(error_msg) return "Unable to fetch detailed forecast." try: # Format the periods into a readable forecast periods = forecast_data["properties"]["periods"] if not periods: await ctx.warning(f"No forecast periods available for coordinates: {latitude}, {longitude}") logger.warning(f"No forecast periods for coordinates: {latitude}, {longitude}") return "No forecast periods available for this location." period_count = min(len(periods), 5) # Only show next 5 periods await ctx.info(f"Processing {period_count} forecast periods") logger.info(f"Processing {period_count} forecast periods") await ctx.report_progress( progress=0.8, total=1.0, message=f"Formatting {period_count} forecast periods..." ) forecasts = [] for i, period in enumerate(periods[:5]): forecast = f""" {period['name']}: Temperature: {period['temperature']}°{period['temperatureUnit']} Wind: {period['windSpeed']} {period['windDirection']} Forecast: {period['detailedForecast']} """ forecasts.append(forecast) # Report progress for each period processed progress = 0.8 + (0.2 * (i + 1) / period_count) await ctx.report_progress( progress=progress, total=1.0, message=f"Processed period {i + 1}/{period_count}" ) await ctx.info(f"Successfully processed forecast for coordinates: {latitude}, {longitude}") logger.info(f"Successfully processed forecast for coordinates: {latitude}, {longitude}") return "\n---\n".join(forecasts) except KeyError as e: error_msg = f"Missing required field in forecast data: {e}" await ctx.error(error_msg) logger.exception("Missing required field in forecast data") return "Error: Invalid forecast data format." except Exception: error_msg = f"Unexpected error processing forecast for coordinates: {latitude}, {longitude}" await ctx.error(error_msg) logger.exception("Unexpected error processing forecast") return "Error: Could not process forecast data."
• weather.py:26-72 (helper)

  Shared helper function used by get_forecast (and get_alerts) to make authenticated HTTP requests to the NWS API with comprehensive error handling and MCP context logging.
  async def make_nws_request(url: str, ctx: Context[ServerSession, None] | None = None) -> dict[str, Any] | None: """Make a request to the NWS API with proper error handling.""" headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"} if ctx: await ctx.debug(f"Making NWS API request to: {url}") logger.debug(f"Making request to NWS API: {url}") async with httpx.AsyncClient(follow_redirects=True) as client: try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() if ctx: await ctx.debug(f"NWS API request successful, status: {response.status_code}") logger.debug(f"NWS API request successful: {response.status_code}") return response.json() except httpx.TimeoutException: error_msg = f"Request timeout for URL: {url}" if ctx: await ctx.error(error_msg) logger.exception("Request timeout occurred") return None except httpx.HTTPStatusError as e: error_msg = f"HTTP error {e.response.status_code} for URL: {url}" if ctx: await ctx.error(error_msg) logger.exception("HTTP status error occurred") return None except httpx.RequestError: error_msg = f"Network error for URL: {url}" if ctx: await ctx.error(error_msg) logger.exception("Network request error occurred") return None except Exception: error_msg = f"Unexpected error for URL: {url}" if ctx: await ctx.error(error_msg) logger.exception("Unexpected error during NWS API request") return None
• weather.py:307-307 (registration)

  List of available tools including get_forecast, exposed via the server_info tool.
  "available_tools": ["get_alerts", "get_forecast", "server_info"],