Weather MCP Server

Instructions

Implementation Reference

• weather.py:108-168 (handler)

  The main execution logic for the 'get_alerts' tool. Fetches active weather alerts for a given US state using the NWS API, validates input, handles errors, processes multiple alerts with progress reporting, and formats the output using helper functions.
  @mcp.tool() async def get_alerts(state: str, ctx: Context[ServerSession, None]) -> str: """Get weather alerts for a US state. Args: state: Two-letter US state code (e.g. CA, NY) """ await ctx.info(f"Fetching weather alerts for state: {state}") logger.info(f"Processing alerts request for state: {state}") # Validate state code if len(state) != 2: error_msg = f"Invalid state code format: {state}. Expected 2-letter code." await ctx.warning(error_msg) logger.warning(error_msg) return "Error: State code must be exactly 2 letters (e.g., CA, NY)" state = state.upper() url = f"{NWS_API_BASE}/alerts/active/area/{state}" await ctx.debug(f"Requesting alerts from URL: {url}") data = await make_nws_request(url, ctx) if not data: error_msg = f"Unable to fetch alerts for state: {state}" await ctx.error(error_msg) logger.error(error_msg) return "Unable to fetch alerts or no alerts found." if "features" not in data: error_msg = f"Invalid response format for state: {state}" await ctx.error(error_msg) logger.error(error_msg) return "Invalid response format from weather service." if not data["features"]: await ctx.info(f"No active alerts found for state: {state}") logger.info(f"No active alerts for state: {state}") return "No active alerts for this state." alert_count = len(data["features"]) await ctx.info(f"Processing {alert_count} alerts for state: {state}") logger.info(f"Found {alert_count} alerts for state: {state}") # Process alerts with progress reporting alerts = [] for i, feature in enumerate(data["features"]): progress = (i + 1) / alert_count await ctx.report_progress( progress=progress, total=1.0, message=f"Processing alert {i + 1}/{alert_count}" ) formatted_alert = await format_alert(feature, ctx) alerts.append(formatted_alert) await ctx.info(f"Successfully processed {alert_count} alerts for state: {state}") logger.info(f"Successfully processed {alert_count} alerts for state: {state}") return "\n---\n".join(alerts)
• weather.py:74-105 (helper)

  Helper function to format an individual weather alert feature into a human-readable string, extracting key properties like event, area, severity, description, and instructions.
  async def format_alert(feature: dict, ctx: Context[ServerSession, None] | None = None) -> str: """Format an alert feature into a readable string.""" try: props = feature["properties"] alert_event = props.get('event', 'Unknown') if ctx: await ctx.debug(f"Formatting alert for event: {alert_event}") logger.debug(f"Formatting alert for event: {alert_event}") formatted_alert = f""" Event: {alert_event} Area: {props.get('areaDesc', 'Unknown')} Severity: {props.get('severity', 'Unknown')} Description: {props.get('description', 'No description available')} Instructions: {props.get('instruction', 'No specific instructions provided')} """ return formatted_alert except KeyError as e: error_msg = f"Missing required field in alert data: {e}" if ctx: await ctx.warning(error_msg) logger.warning(error_msg) return "Error: Invalid alert data format" except Exception: error_msg = "Unexpected error while formatting alert" if ctx: await ctx.error(error_msg) logger.exception("Unexpected error while formatting alert") return "Error: Could not format alert"
• weather.py:26-71 (helper)

  Shared helper function for making HTTP requests to the NWS API, with comprehensive error handling for timeouts, HTTP errors, network issues, and logging via MCP context.
  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