Weather MCP Server

"""Weather API client for fetching weather data.""" import os from typing import Dict, List, Optional from datetime import datetime, timedelta import httpx from dotenv import load_dotenv load_dotenv() class WeatherAPI: """Client for fetching weather data from external APIs.""" def __init__(self): """Initialize the weather API client.""" self.api_key = os.getenv("WEATHER_API_KEY") self.base_url = "https://api.openweathermap.org/data/2.5" self.cache: Dict[str, tuple] = {} # location -> (data, timestamp) self.cache_ttl = timedelta(minutes=10) # Cache for 10 minutes def _is_cache_valid(self, location: str) -> bool: """Check if cached data for location is still valid.""" if location not in self.cache: return False _, timestamp = self.cache[location] return datetime.now() - timestamp < self.cache_ttl def _get_from_cache(self, location: str) -> Optional[Dict]: """Get cached data if valid.""" if self._is_cache_valid(location): data, _ = self.cache[location] return data return None def _cache_data(self, location: str, data: Dict): """Cache weather data with current timestamp.""" self.cache[location] = (data, datetime.now()) def _normalize_location(self, location: str) -> str: """Normalize location string for API calls.""" return location.strip().title() async def get_current_weather(self, location: str) -> Dict: """ Get current weather for a location. Args: location: City name or location string Returns: Dictionary with weather data """ location = self._normalize_location(location) # Check cache first cached = self._get_from_cache(f"current_{location}") if cached: return cached # If no API key, return mock data if not self.api_key: return self._get_mock_current_weather(location) try: async with httpx.AsyncClient() as client: url = f"{self.base_url}/weather" params = { "q": location, "appid": self.api_key, "units": "metric" } response = await client.get(url, params=params, timeout=10.0) response.raise_for_status() data = response.json() # Format the response result = { "location": data.get("name", location), "country": data.get("sys", {}).get("country", ""), "temperature": data.get("main", {}).get("temp", 0), "feels_like": data.get("main", {}).get("feels_like", 0), "humidity": data.get("main", {}).get("humidity", 0), "pressure": data.get("main", {}).get("pressure", 0), "description": data.get("weather", [{}])[0].get("description", ""), "wind_speed": data.get("wind", {}).get("speed", 0), "wind_direction": data.get("wind", {}).get("deg", 0), "visibility": data.get("visibility", 0) / 1000 if data.get("visibility") else None, "clouds": data.get("clouds", {}).get("all", 0), "timestamp": datetime.now().isoformat() } self._cache_data(f"current_{location}", result) return result except httpx.HTTPError as e: # Fallback to mock data on API error return self._get_mock_current_weather(location) async def get_forecast(self, location: str, days: int = 5) -> Dict: """ Get weather forecast for a location. Args: location: City name or location string days: Number of days to forecast (1-5) Returns: Dictionary with forecast data """ location = self._normalize_location(location) days = max(1, min(5, days)) # Clamp between 1 and 5 # Check cache cache_key = f"forecast_{location}_{days}" cached = self._get_from_cache(cache_key) if cached: return cached # If no API key, return mock data if not self.api_key: return self._get_mock_forecast(location, days) try: async with httpx.AsyncClient() as client: url = f"{self.base_url}/forecast" params = { "q": location, "appid": self.api_key, "units": "metric", "cnt": days * 8 # 8 forecasts per day (3-hour intervals) } response = await client.get(url, params=params, timeout=10.0) response.raise_for_status() data = response.json() # Format the response forecasts = [] for item in data.get("list", [])[:days * 8]: forecasts.append({ "datetime": item.get("dt_txt", ""), "temperature": item.get("main", {}).get("temp", 0), "feels_like": item.get("main", {}).get("feels_like", 0), "humidity": item.get("main", {}).get("humidity", 0), "pressure": item.get("main", {}).get("pressure", 0), "description": item.get("weather", [{}])[0].get("description", ""), "wind_speed": item.get("wind", {}).get("speed", 0), "clouds": item.get("clouds", {}).get("all", 0), }) result = { "location": data.get("city", {}).get("name", location), "country": data.get("city", {}).get("country", ""), "forecasts": forecasts, "days": days, "timestamp": datetime.now().isoformat() } self._cache_data(cache_key, result) return result except httpx.HTTPError as e: # Fallback to mock data on API error return self._get_mock_forecast(location, days) async def search_locations(self, query: str) -> List[Dict]: """ Search for locations matching the query. Args: query: Search query string Returns: List of matching locations """ query = query.strip() # If no API key, return mock locations if not self.api_key: return self._get_mock_locations(query) try: async with httpx.AsyncClient() as client: url = "https://api.openweathermap.org/geo/1.0/direct" params = { "q": query, "limit": 5, "appid": self.api_key } response = await client.get(url, params=params, timeout=10.0) response.raise_for_status() data = response.json() locations = [] for item in data: locations.append({ "name": item.get("name", ""), "country": item.get("country", ""), "state": item.get("state", ""), "lat": item.get("lat", 0), "lon": item.get("lon", 0), }) return locations except httpx.HTTPError: # Fallback to mock locations on API error return self._get_mock_locations(query) def _get_mock_current_weather(self, location: str) -> Dict: """Return mock current weather data for demo purposes.""" import random return { "location": location, "country": "US", "temperature": round(random.uniform(15, 30), 1), "feels_like": round(random.uniform(14, 29), 1), "humidity": random.randint(40, 80), "pressure": random.randint(1000, 1020), "description": random.choice(["clear sky", "few clouds", "scattered clouds", "broken clouds", "shower rain", "rain", "thunderstorm", "snow", "mist"]), "wind_speed": round(random.uniform(0, 15), 1), "wind_direction": random.randint(0, 360), "visibility": round(random.uniform(5, 10), 1), "clouds": random.randint(0, 100), "timestamp": datetime.now().isoformat(), "note": "Mock data - set WEATHER_API_KEY in .env for real data" } def _get_mock_forecast(self, location: str, days: int) -> Dict: """Return mock forecast data for demo purposes.""" import random forecasts = [] base_time = datetime.now() for i in range(days * 8): forecast_time = base_time + timedelta(hours=i * 3) forecasts.append({ "datetime": forecast_time.strftime("%Y-%m-%d %H:%M:%S"), "temperature": round(random.uniform(15, 30), 1), "feels_like": round(random.uniform(14, 29), 1), "humidity": random.randint(40, 80), "pressure": random.randint(1000, 1020), "description": random.choice(["clear sky", "few clouds", "scattered clouds", "broken clouds", "shower rain", "rain"]), "wind_speed": round(random.uniform(0, 15), 1), "clouds": random.randint(0, 100), }) return { "location": location, "country": "US", "forecasts": forecasts, "days": days, "timestamp": datetime.now().isoformat(), "note": "Mock data - set WEATHER_API_KEY in .env for real data" } def _get_mock_locations(self, query: str) -> List[Dict]: """Return mock location search results for demo purposes.""" # Common cities that might match common_cities = [ {"name": "New York", "country": "US", "state": "New York", "lat": 40.7128, "lon": -74.0060}, {"name": "London", "country": "GB", "state": "", "lat": 51.5074, "lon": -0.1278}, {"name": "Paris", "country": "FR", "state": "", "lat": 48.8566, "lon": 2.3522}, {"name": "Tokyo", "country": "JP", "state": "", "lat": 35.6762, "lon": 139.6503}, {"name": "San Francisco", "country": "US", "state": "California", "lat": 37.7749, "lon": -122.4194}, ] # Filter by query (case-insensitive) query_lower = query.lower() matches = [city for city in common_cities if query_lower in city["name"].lower()] # If no matches, return all cities return matches if matches else common_cities[:3]