MCP Weather Sample

Instructions

獲取特定位置的預報資料

Args:
    latitude (float): 緯度
    longitude (float): 經度

Implementation Reference

• src/servers/weather/stdio/weather.py:67-101 (handler)

  Main handler for get_forecast tool in stdio transport - fetches weather forecast data from NOAA NWS API and formats it for display. Takes latitude and longitude parameters, fetches grid points endpoint, then retrieves and formats forecast periods.

  async def get_forecast(latitude: float, longitude: float) -> str:
      """
      獲取特定位置的預報資料
  
      Args:
          latitude (float): 緯度
          longitude (float): 經度
      """
      # 首先取得預測網格端點
      points_url = f"{NWS_API_BASE_URL}/points/{latitude},{longitude}"
      points_data = await make_nws_request(points_url)
  
      if not points_data:
          return "Unable to fetch forecast data for this location."
      
      # 從端點取得詳細預報
      forecast_url = points_data["properties"]["forecast"]
      forecast_data = await make_nws_request(forecast_url)
  
      if not forecast_data:
          return "Unable to fetch detailed forecast."
      
      periods = forecast_data["properties"]["periods"]
      forecasts = []
  
      for period in periods[:5]:
          forecast = f"""
  {period['name']}:
  Temperature: {period['temperature']}°{period['temperatureUnit']}
  Wind: {period['windSpeed']} {period['windDirection']}
  Forecast: {period['detailedForecast']}
  """
          forecasts.append(forecast)
  
      return "\n\n".join(forecasts)

• src/servers/weather/streamable-http/mcp-weather.py:130-165 (handler)

  Main handler for get_forecast tool in streamable-http transport - similar to stdio version but includes Context parameter for logging via ctx.info(). Uses '\n---\n' separator between forecast periods.

  async def get_forecast(latitude: float, longitude: float, ctx: Context) -> str:
      """Get weather forecast for a location.
  
      Args:
          latitude: Latitude of the location
          longitude: Longitude of the location
      """
      # First get the forecast grid endpoint
      points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
      points_data = await make_nws_request(points_url)
  
      await ctx.info(f"Received data: {points_data}")
  
      if not points_data:
          return "Unable to fetch forecast data for this location."
  
      # Get the forecast URL from the points response
      forecast_url = points_data["properties"]["forecast"]
      forecast_data = await make_nws_request(forecast_url)
  
      if not forecast_data:
          return "Unable to fetch detailed forecast."
  
      # Format the periods into a readable forecast
      periods = forecast_data["properties"]["periods"]
      forecasts = []
      for period in periods[:5]:  # Only show next 5 periods
          forecast = f"""
  {period['name']}:
  Temperature: {period['temperature']}°{period['temperatureUnit']}
  Wind: {period['windSpeed']} {period['windDirection']}
  Forecast: {period['detailedForecast']}
  """
          forecasts.append(forecast)
  
      return "\n---\n".join(forecasts)

• src/servers/weather/sse/mcp-weather.py:130-165 (handler)

  Main handler for get_forecast tool in SSE transport - nearly identical to streamable-http version with Context parameter and logging capability. Fetches forecast from NOAA NWS API and returns formatted weather periods.

  async def get_forecast(latitude: float, longitude: float, ctx: Context) -> str:
      """Get weather forecast for a location.
  
      Args:
          latitude: Latitude of the location
          longitude: Longitude of the location
      """
      # First get the forecast grid endpoint
      points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
      points_data = await make_nws_request(points_url)
  
      await ctx.info(f"Received data: {points_data}")
  
      if not points_data:
          return "Unable to fetch forecast data for this location."
  
      # Get the forecast URL from the points response
      forecast_url = points_data["properties"]["forecast"]
      forecast_data = await make_nws_request(forecast_url)
  
      if not forecast_data:
          return "Unable to fetch detailed forecast."
  
      # Format the periods into a readable forecast
      periods = forecast_data["properties"]["periods"]
      forecasts = []
      for period in periods[:5]:  # Only show next 5 periods
          forecast = f"""
  {period['name']}:
  Temperature: {period['temperature']}°{period['temperatureUnit']}
  Wind: {period['windSpeed']} {period['windDirection']}
  Forecast: {period['detailedForecast']}
  """
          forecasts.append(forecast)
  
      return "\n---\n".join(forecasts)

• src/servers/weather/stdio/weather.py:12-28 (helper)

  Helper function make_nws_request - performs HTTP requests to NOAA National Weather Service API with proper headers, error handling, and timeout configuration. Returns parsed JSON or None on error.

  async def make_nws_request(url: str) -> dict[str, Any] | None:
      """
      發送 HTTP 請求到 NOAA 天氣 API 並返回 JSON 響應
      """
      headers = {
          "User-Agent": USER_AGENT,
          "Accept": "application/geo+json",
      }
  
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, headers=headers, timeout=30.0)
              response.raise_for_status()
              return response.json()
          except httpx.HTTPStatusError as e:
              print(f"HTTP error occurred: {e}")
              return None

• src/servers/weather/streamable-http/mcp-weather.py:79-91 (helper)

  Helper function make_nws_request (streamable-http version) - performs HTTP requests to NOAA NWS API with User-Agent and Accept headers, 30-second timeout, and exception handling. Returns JSON response or None.

  async def make_nws_request(url: str) -> dict[str, Any] | None:
      """Make a request to the NWS API with proper error handling."""
      headers = {
          "User-Agent": USER_AGENT,
          "Accept": "application/geo+json"
      }
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, headers=headers, timeout=30.0)
              response.raise_for_status()
              return response.json()
          except Exception:
              return None