get_air_quality

Instructions

Implementation Reference

• src/mcp_weather_server/tools/tools_air_quality.py:67-129 (handler)

  The main handler function for the 'get_air_quality' tool. It validates inputs (city, optional variables), geocodes the city to coordinates, fetches air quality data from the service, extracts current values, and formats a comprehensive response with field descriptions for AI analysis.

  async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
      """
      Execute the air quality tool.
      """
      try:
          self.validate_required_args(args, ["city"])
  
          city = args["city"]
          variables = args.get("variables", [
              "pm10", "pm2_5", "ozone", "nitrogen_dioxide", "carbon_monoxide"
          ])
  
          logger.info(f"Getting air quality for: {city} with variables: {variables}")
  
          # Get coordinates for the city
          latitude, longitude = await self.weather_service.get_coordinates(city)
  
          # Get air quality data
          aq_data = await self.air_quality_service.get_air_quality(
              latitude, longitude, variables
          )
  
          # Get current air quality values
          current_aq = self.air_quality_service.get_current_air_quality_index(aq_data)
  
          # Build comprehensive response data for AI comprehension
          response_data = {
              "city": city,
              "latitude": latitude,
              "longitude": longitude,
              "current_air_quality": current_aq,
              "full_data": aq_data
          }
  
          # Format the response with comprehensive field descriptions
          formatted_response = self.air_quality_service.format_air_quality_comprehensive(
              response_data
          )
  
          return [
              TextContent(
                  type="text",
                  text=formatted_response
              )
          ]
  
      except ValueError as e:
          logger.error(f"Air quality service error: {str(e)}")
          return [
              TextContent(
                  type="text",
                  text=f"Error: {str(e)}"
              )
          ]
      except Exception as e:
          logger.exception(f"Unexpected error in get_air_quality: {str(e)}")
          return [
              TextContent(
                  type="text",
                  text=f"Unexpected error occurred: {str(e)}"
              )
          ]

• src/mcp_weather_server/tools/tools_air_quality.py:28-65 (handler)

  Schema/input validation definition for the 'get_air_quality' tool. Defines the tool name, description, and input schema with required 'city' (string) and optional 'variables' (array of pollutant enums).

  def get_tool_description(self) -> Tool:
      """
      Return the tool description for air quality lookup.
      """
      return Tool(
          name=self.name,
          description="""Get air quality information for a specified city including PM2.5, PM10,
          ozone, nitrogen dioxide, carbon monoxide, and other pollutants. Provides health
          advisories based on current air quality levels.""",
          inputSchema={
              "type": "object",
              "properties": {
                  "city": {
                      "type": "string",
                      "description": "The name of the city to fetch air quality information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function."
                  },
                  "variables": {
                      "type": "array",
                      "items": {
                          "type": "string",
                          "enum": [
                              "pm10",
                              "pm2_5",
                              "carbon_monoxide",
                              "nitrogen_dioxide",
                              "ozone",
                              "sulphur_dioxide",
                              "ammonia",
                              "dust",
                              "aerosol_optical_depth"
                          ]
                      },
                      "description": "Air quality variables to retrieve. If not specified, defaults to pm10, pm2_5, ozone, nitrogen_dioxide, and carbon_monoxide."
                  }
              },
              "required": ["city"]
          }
      )

• src/mcp_weather_server/server.py:95-97 (registration)

  Registration of the GetAirQualityToolHandler in the central register_all_tools() function, binding the handler to the server's tool handler registry.

  # Air quality tools
  add_tool_handler(GetAirQualityToolHandler())
  add_tool_handler(GetAirQualityDetailsToolHandler())

• src/mcp_weather_server/tools/air_quality_service.py:28-74 (helper)

  The AirQualityService.get_air_quality() method that actually calls the Open-Meteo Air Quality API to fetch pollutant data for given coordinates.

  async def get_air_quality(
      self,
      latitude: float,
      longitude: float,
      hourly_vars: List[str] = None
  ) -> Dict[str, Any]:
      """
      Get air quality data for given coordinates.
  
      Args:
          latitude: Latitude coordinate
          longitude: Longitude coordinate
          hourly_vars: List of hourly variables to retrieve
  
      Returns:
          Air quality data dictionary
  
      Raises:
          ValueError: If air quality data cannot be retrieved
      """
      if hourly_vars is None:
          hourly_vars = ["pm10", "pm2_5", "ozone", "nitrogen_dioxide", "carbon_monoxide"]
  
      hourly_str = ",".join(hourly_vars)
  
      url = (
          f"{self.BASE_AIR_QUALITY_URL}"
          f"?latitude={latitude}&longitude={longitude}"
          f"&hourly={hourly_str}"
          f"&timezone=GMT"
      )
  
      logger.info(f"Fetching air quality data from: {url}")
  
      try:
          async with httpx.AsyncClient() as client:
              response = await client.get(url)
  
              if response.status_code != 200:
                  raise ValueError(f"Air Quality API returned status {response.status_code}")
  
              return response.json()
  
      except httpx.RequestError as e:
          raise ValueError(f"Network error while fetching air quality data: {str(e)}")
      except (KeyError, IndexError) as e:
          raise ValueError(f"Invalid response format from air quality API: {str(e)}")