import os
import httpx
from dotenv import load_dotenv
from fastmcp import FastMCP, Client
import argparse
import asyncio
# Load environment variables from .env file
load_dotenv()
# define mcp server name
mcp = FastMCP(name="Weather Server")
async def _get_current_weather_data(location: str, units: str = "metric") -> dict:
"""
Internal function to get the current weather forecast for a specific location.
"""
api_key = os.getenv("OPENWEATHER_API_KEY")
if not api_key:
raise ValueError("OPENWEATHER_API_KEY not found in environment variables. Please set it in your .env file.")
base_url = "https://api.openweathermap.org/data/2.5/weather"
params = {
"q": location,
"units": units,
"appid": api_key
}
async with httpx.AsyncClient() as client:
try:
response = await client.get(base_url, params=params)
response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
weather_data = response.json()
# Extract relevant information
main_weather = weather_data.get("weather", [{}])[0].get("main", "N/A")
description = weather_data.get("weather", [{}])[0].get("description", "N/A")
temp = weather_data.get("main", {}).get("temp", "N/A")
feels_like = weather_data.get("main", {}).get("feels_like", "N/A")
humidity = weather_data.get("main", {}).get("humidity", "N/A")
wind_speed = weather_data.get("wind", {}).get("speed", "N/A")
city_name = weather_data.get("name", "N/A")
return {
"location": city_name,
"main_weather": main_weather,
"description": description,
"temperature": temp,
"feels_like": feels_like,
"humidity": humidity,
"wind_speed": wind_speed,
"units": units
}
except httpx.RequestError as exc:
raise ConnectionError(f"An error occurred while requesting {exc.request.url!r}: {exc}")
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 401:
raise ValueError("Invalid OpenWeatherMap API key. Please check your OPENWEATHER_API_KEY.")
elif exc.response.status_code == 404:
raise ValueError(f"Location '{location}' not found.")
else:
raise ValueError(f"HTTP error {exc.response.status_code} while fetching weather: {exc.response.text}")
except Exception as e:
raise RuntimeError(f"An unexpected error occurred: {e}")
@mcp.tool()
async def get_current_weather(location: str, units: str = "metric") -> dict:
"""
Gets the current weather forecast for a specific location.
Args:
location: The city name (e.g., "London", "New York").
units: The units of measurement. Can be "metric" (Celsius), "imperial" (Fahrenheit), or "standard" (Kelvin). Defaults to "metric".
"""
return await _get_current_weather_data(location, units)
async def run_tests():
print("Running tests for mcp_weather.py...")
# Test case 1: Valid location, default units
try:
print("Testing get_current_weather for London (metric)...")
weather = await _get_current_weather_data(location="London")
print(f"Test 1 Passed: Weather in {weather['location']} is {weather['main_weather']}, Temp: {weather['temperature']} {weather['units']}")
assert weather["location"] == "London"
assert isinstance(weather["temperature"], (int, float))
except Exception as e:
print(f"Test 1 Failed: {e}")
# Test case 2: Valid location, imperial units
try:
print("Testing get_current_weather for New York (imperial)...")
weather = await _get_current_weather_data(location="New York", units="imperial")
print(f"Test 2 Passed: Weather in {weather['location']} is {weather['main_weather']}, Temp: {weather['temperature']} {weather['units']}")
assert weather["location"] == "New York"
assert isinstance(weather["temperature"], (int, float))
except Exception as e:
print(f"Test 2 Failed: {e}")
# Test case 3: Invalid location
try:
print("Testing get_current_weather for NonExistentCity123 (should fail)...")
await _get_current_weather_data(location="NonExistentCity123")
print("Test 3 Failed: Expected ValueError for invalid location, but it passed.")
except ValueError as e:
print(f"Test 3 Passed (expected failure): {e}")
except Exception as e:
print(f"Test 3 Failed (unexpected error type): {e}")
print("\nTests finished.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="MCP Weather Server")
parser.add_argument("--test", action="store_true", help="Run tests for the weather tool.")
args = parser.parse_args()
if args.test:
asyncio.run(run_tests())
else:
# Add a small delay to allow the server to fully initialize
# This sleep might not be necessary if mcp.run blocks, but keeping for safety.
# await asyncio.sleep(1) # Removed as mcp.run handles the event loop
try:
mcp.run(transport="stdio")
except Exception as e:
import sys
print(f"Error running MCP server: {e}", file=sys.stderr)
