MCP Weather SSE Server

Instructions

Implementation Reference

• weather.py:232-258 (handler)

  The main handler function for the 'get_current_weather' MCP tool. Decorated with @mcp.tool() for automatic registration and schema inference from type hints and docstring. Fetches current weather data from OpenWeatherMap API, formats it, and returns as JSON.
  @mcp.tool() async def get_current_weather(city: str, units: str = "metric") -> str: """Get current weather for a city using OpenWeatherMap. Args: city: City name (e.g. 'London', 'New York') units: Units of measurement ('metric' or 'imperial') """ print(f"get_current_weather called with city: {city}, units: {units}", file=sys.stderr) if not OPENWEATHER_API_KEY: return "OpenWeatherMap API key not configured. Please set OPENWEATHER_API_KEY environment variable." url = f"{OPENWEATHER_API_BASE}/weather" params = { "q": city, "units": units } data = await make_openweather_request(url, params) if not data: return "Unable to fetch current weather data." result = format_current_weather(data, units) return json.dumps(result, indent=2)
• weather.py:67-116 (helper)

  Helper function used by get_current_weather to format the raw API response into a structured dictionary with location and current weather details.
  def format_current_weather(data: dict, units: str) -> dict: """Format current weather data from OpenWeatherMap.""" temp_unit = "°C" if units == "metric" else "°F" speed_unit = "m/s" if units == "metric" else "mph" try: weather = { "location": { "name": data.get("name", "Unknown"), "country": data.get("sys", {}).get("country", "Unknown"), "coordinates": { "latitude": data.get("coord", {}).get("lat", 0), "longitude": data.get("coord", {}).get("lon", 0) } }, "current": { "temperature": f"{data.get('main', {}).get('temp', 0)}{temp_unit}", "feels_like": f"{data.get('main', {}).get('feels_like', 0)}{temp_unit}", "humidity": f"{data.get('main', {}).get('humidity', 0)}%", "pressure": f"{data.get('main', {}).get('pressure', 0)} hPa", "wind": { "speed": f"{data.get('wind', {}).get('speed', 0)} {speed_unit}", "direction": data.get('wind', {}).get('deg', 0) }, "weather": { "main": data.get('weather', [{}])[0].get('main', "Unknown"), "description": data.get('weather', [{}])[0].get('description', "Unknown"), "icon": data.get('weather', [{}])[0].get('icon', "Unknown") }, "visibility": f"{data.get('visibility', 0) / 1000} km", "cloudiness": f"{data.get('clouds', {}).get('all', 0)}%", "sunrise": data.get('sys', {}).get('sunrise', 0), "sunset": data.get('sys', {}).get('sunset', 0) } } if 'rain' in data: weather['current']['rain'] = { "1h": f"{data['rain'].get('1h', 0)} mm" } if 'snow' in data: weather['current']['snow'] = { "1h": f"{data['snow'].get('1h', 0)} mm" } return weather except Exception as e: print(f"Error formatting weather data: {e}", file=sys.stderr) return {"error": "Error formatting weather data"}
• weather.py:39-54 (helper)

  Helper function used by get_current_weather to make HTTP requests to the OpenWeatherMap API, including API key handling and error management.
  async def make_openweather_request(url: str, params: dict) -> dict[str, Any] | None: """Make a request to OpenWeatherMap API with proper error handling.""" if not OPENWEATHER_API_KEY: print("OpenWeatherMap API key not found", file=sys.stderr) return None params["appid"] = OPENWEATHER_API_KEY async with httpx.AsyncClient() as client: try: response = await client.get(url, params=params, timeout=30.0) response.raise_for_status() return response.json() except Exception as e: print(f"OpenWeatherMap API error: {e}", file=sys.stderr) return None
• weather.py:232-232 (registration)

  The @mcp.tool() decorator on the get_current_weather function registers it as an MCP tool and defines its schema based on function signature and documentation.
  @mcp.tool()