MCP Server Template

""" Weather Tools Module Provides weather information tools using OpenWeatherMap API. Supports both real API calls and mock data for development/testing. """ import json import logging from typing import Any, Dict, Optional import httpx from fastmcp import FastMCP from .base import BaseTool from ..config.settings import get_settings class WeatherTools(BaseTool): """ Collection of weather-related tools. Provides weather information using OpenWeatherMap API with fallback to mock data for development and testing purposes. """ def __init__(self): super().__init__(name="weather") self.settings = get_settings() def _get_api_key(self) -> Optional[str]: """Get weather API key from environment.""" return getattr(self.settings, 'weather_api_key', None) def _use_mock_data(self) -> bool: """Check if mock data should be used.""" return getattr(self.settings, 'use_mock_weather_data', True) def _get_mock_weather_data(self, location: str) -> Dict[str, Any]: """Return realistic mock weather data for testing.""" return { "location": location, "temperature": 22.5, "temperature_unit": "°C", "humidity": 65, "humidity_unit": "%", "pressure": 1013.25, "pressure_unit": "hPa", "wind_speed": 3.2, "wind_speed_unit": "m/s", "wind_direction": 180, "wind_direction_unit": "degrees", "weather": "Partly cloudy", "description": "Few clouds", "visibility": 10000, "visibility_unit": "meters", "source": "mock_data", "timestamp": "2024-01-15T12:00:00Z" } async def _fetch_real_weather_data(self, location: str) -> Dict[str, Any]: """Fetch real weather data from OpenWeatherMap API.""" api_key = self._get_api_key() if not api_key: raise ValueError("WEATHER_API_KEY environment variable is required for real weather data") # Validate API key format (OpenWeatherMap keys are 32 hexadecimal characters) if not isinstance(api_key, str) or len(api_key) != 32: raise ValueError(f"Invalid API key format: expected 32 characters, got {len(api_key) if api_key else 0}") url = "https://api.openweathermap.org/data/2.5/weather" params = { "q": location, "appid": api_key, "units": "metric" } self.logger.debug(f"Making weather API request to {url} with key ending in ...{api_key[-4:]}") try: async with httpx.AsyncClient() as client: response = await client.get(url, params=params, timeout=10.0) response.raise_for_status() data = response.json() # Transform API response to standardized format weather_data = { "location": f"{data['name']}, {data['sys']['country']}", "temperature": data['main']['temp'], "temperature_unit": "°C", "humidity": data['main']['humidity'], "humidity_unit": "%", "pressure": data['main']['pressure'], "pressure_unit": "hPa", "wind_speed": data.get('wind', {}).get('speed', 0), "wind_speed_unit": "m/s", "wind_direction": data.get('wind', {}).get('deg', 0), "wind_direction_unit": "degrees", "weather": data['weather'][0]['main'], "description": data['weather'][0]['description'], "visibility": data.get('visibility', 0), "visibility_unit": "meters", "source": "openweathermap", "timestamp": data.get('dt', 0) } return weather_data except httpx.HTTPStatusError as e: if e.response.status_code == 401: # Provide more detailed error message for API key issues self.logger.error(f"API key authentication failed - key ending in ...{api_key[-4:]}") raise ValueError(f"Invalid API key provided for weather service. Please verify your OpenWeatherMap API key is active and correct.") elif e.response.status_code == 404: raise ValueError(f"Location '{location}' not found") else: raise ValueError(f"Weather API error: HTTP {e.response.status_code}") except httpx.TimeoutException: raise ValueError("Weather API request timed out") except httpx.RequestError as e: raise ValueError(f"Weather API request failed: {str(e)}") except (KeyError, json.JSONDecodeError) as e: raise ValueError(f"Invalid weather API response format: {str(e)}") def register_with_mcp(self, mcp: FastMCP) -> None: """ Register weather tools with the FastMCP instance. Args: mcp: The FastMCP instance to register with """ @mcp.tool() async def get_current_weather(location: str) -> Dict[str, Any]: """ Get current weather information for a specific location. Args: location: The location to get weather for (city name, "city,country", etc.) Returns: Dictionary containing current weather information including: - location: Formatted location name - temperature: Current temperature in Celsius - humidity: Humidity percentage - pressure: Atmospheric pressure in hPa - wind_speed: Wind speed in m/s - wind_direction: Wind direction in degrees - weather: Main weather condition - description: Detailed weather description - visibility: Visibility in meters - source: Data source (mock_data or openweathermap) - timestamp: Data timestamp Raises: ValueError: If location is invalid or API request fails """ self._log_tool_call("get_current_weather", location=location) if not location or not location.strip(): raise ValueError("Location cannot be empty") location = location.strip() try: if self._use_mock_data(): self.logger.info(f"Returning mock weather data for: {location}") return self._get_mock_weather_data(location) else: self.logger.info(f"Fetching real weather data for: {location}") return await self._fetch_real_weather_data(location) except Exception as e: self.logger.error(f"Error getting weather for {location}: {str(e)}") raise self.logger.info("Registered weather tools: get_current_weather")