Weather MCP Server

Instructions

Implementation Reference

• weather_api.py:43-98 (handler)

  Core implementation of the get_current_weather tool. Fetches current weather from OpenWeatherMap API with caching, mock data fallback, and comprehensive data formatting.
  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)
• main.py:30-43 (registration)

  Registers the get_current_weather tool with the MCP server, including name, description, and input schema.
  Tool( name="get_current_weather", description="Get current weather conditions for a specific location", inputSchema={ "type": "object", "properties": { "location": { "type": "string", "description": "City name or location (e.g., 'New York', 'London')" } }, "required": ["location"] } ),
• main.py:33-42 (schema)

  Defines the input schema for the get_current_weather tool, specifying the required 'location' parameter.
  inputSchema={ "type": "object", "properties": { "location": { "type": "string", "description": "City name or location (e.g., 'New York', 'London')" } }, "required": ["location"] }
• main.py:85-97 (handler)

  MCP server handler dispatch for get_current_weather tool call, validates input and delegates to WeatherAPI instance.
  if name == "get_current_weather": location = arguments.get("location", "") if not location: return [TextContent( type="text", text="Error: location parameter is required" )] weather_data = await weather_api.get_current_weather(location) return [TextContent( type="text", text=json.dumps(weather_data, indent=2) )]
• weather_api.py:209-226 (helper)

  Helper function providing mock weather data when API key is missing or on API errors.
  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" }