weather_current
Fetch current weather data and air quality information for any location using city name, coordinates, or postal code. Designed for AI assistants via Weather MCP Server.
Instructions
Get current weather for a location
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| aqi | No | Include air quality data ('yes' or 'no') | no |
| q | Yes | Location query (city name, lat/lon, postal code, etc) |
Implementation Reference
- server.py:177-183 (handler)Handler logic for the 'weather_current' tool: validates input parameters, calls the shared 'fetch' helper with the 'current.json' endpoint, and formats the result as a JSON string.if tool_name == "weather_current": q = arguments.get("q") aqi = arguments.get("aqi", "no") if not q: raise ValueError("Location (q) is required") result = await fetch("current.json", {"q": q, "aqi": aqi}) content = json.dumps(result, indent=2)
- server.py:112-126 (schema)Input schema for 'weather_current' tool defining parameters 'q' (required location query) and optional 'aqi'."inputSchema": { "type": "object", "properties": { "q": { "type": "string", "description": "Location query (city name, lat/lon, postal code, etc)" }, "aqi": { "type": "string", "description": "Include air quality data ('yes' or 'no')", "default": "no" } }, "required": ["q"] }
- server.py:109-127 (registration)Registration of the 'weather_current' tool in the 'tools/list' MCP method response.{ "name": "weather_current", "description": "Get current weather for a location", "inputSchema": { "type": "object", "properties": { "q": { "type": "string", "description": "Location query (city name, lat/lon, postal code, etc)" }, "aqi": { "type": "string", "description": "Include air quality data ('yes' or 'no')", "default": "no" } }, "required": ["q"] } },
- server.py:42-79 (helper)Shared helper function that performs the actual HTTP request to WeatherAPI.com using httpx, adds API key, handles errors, and returns parsed JSON data. Used by all weather tools including 'weather_current'.async def fetch(endpoint: str, params: dict) -> dict: """Perform async GET to WeatherAPI and return JSON.""" logger.debug(f"fetch() called with endpoint: {endpoint}, params: {params}") if not WEATHER_API_KEY: logger.error("Weather API key not set.") raise Exception("Weather API key not set.") params["key"] = WEATHER_API_KEY url = f"https://api.weatherapi.com/v1/{endpoint}" logger.info(f"Requesting {url}") async with httpx.AsyncClient() as client: logger.debug("HTTPx client created") try: resp = await client.get(url, params=params) logger.debug(f"HTTP response received: status={resp.status_code}") if resp.status_code != 200: try: error_data = resp.json() detail = error_data.get("error", {}).get("message", resp.text) except: detail = resp.text logger.error(f"WeatherAPI error {resp.status_code}: {detail}") raise Exception(f"WeatherAPI error {resp.status_code}: {detail}") data = resp.json() logger.debug(f"JSON parsing successful") logger.info(f"WeatherAPI success: {url}") return data except httpx.RequestError as e: logger.error(f"HTTPX request error: {e}") raise Exception(f"Request error: {e}") except Exception as e: logger.error(f"Unexpected error: {e}") raise Exception(f"Unexpected error: {e}")