MCP Weather SSE Server

Instructions

Args:
    city: City name (e.g. 'London', 'New York')
    units: Units of measurement ('metric' or 'imperial')

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()