main.py•12.6 kB
#!/usr/bin/env python3
"""
Weather MCP Application - Main entry point for weather queries.
"""
import os
import sys
import argparse
from pathlib import Path
# Load .env file if exists
try:
from dotenv import load_dotenv
env_path = Path(__file__).parent / ".env"
if env_path.exists():
load_dotenv(env_path)
print(f"✅ Loaded environment from {env_path}")
except ImportError:
pass # dotenv is optional
# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent / "src"))
from weather_mcp import WeatherAgent, WeatherMCPClient, create_visualizer
from weather_mcp.visualization import plot_weather_trends_horizontal
def setup_environment():
"""Setup environment variables and check dependencies."""
# Check for Google API key
if not os.getenv("GOOGLE_API_KEY"):
print("⚠️ Warning: GOOGLE_API_KEY not found in environment variables")
print(" Set it with: export GOOGLE_API_KEY='your_api_key'")
print(" Or use --api-key parameter")
return False
return True
def run_server():
"""Run the MCP server."""
from weather_mcp.server import main as server_main
print("🚀 Starting Weather MCP Server...")
server_main()
def run_client_demo():
"""Run a simple client demonstration."""
print("🔧 Running MCP Client Demo...")
client = WeatherMCPClient()
# Test city geocoding
print("\n📍 Testing geocoding...")
geo_result = client.geocode_city("Tokyo")
print(f"Tokyo coordinates: {geo_result}")
if "coord" in geo_result:
lat, lon = geo_result["coord"]["lat"], geo_result["coord"]["lon"]
# Test weather query
print("\n🌤️ Testing weather query...")
weather_result = client.get_weather(lat, lon, hours=6)
print(f"Weather data keys: {list(weather_result.keys())}")
# Test combined query
print("\n🏙️ Testing combined city weather...")
combined_result = client.get_city_weather("Tokyo", hours=6)
print(f"Combined result keys: {list(combined_result.keys())}")
def run_agent_demo(api_key=None):
"""Run agent demonstration."""
print("🤖 Running Weather Agent Demo...")
if not api_key and not os.getenv("GOOGLE_API_KEY"):
print("❌ Error: Google API key required for agent demo")
return
try:
agent = WeatherAgent(api_key=api_key)
# Test questions
questions = [
"Tell me the weather in Tokyo for the next 8 hours",
"日本未來6小時的溫度趨勢如何？",
"What's the weather like in London?"
]
for question in questions:
print(f"\n❓ Question: {question}")
answer = agent.ask_weather(question, verbose=True)
print(f"✅ Answer: {answer[:200]}...")
except Exception as e:
print(f"❌ Agent demo failed: {e}")
def run_visualization_demo(city="Tokyo", target_date="", hours=None):
"""Run visualization demonstration.
Args:
city: City name for weather query
target_date: Target date in YYYY-MM-DD format (empty for today)
hours: Number of hours to show (None for auto-calculate)
"""
print("📊 Running Visualization Demo...")
try:
from datetime import datetime
client = WeatherMCPClient()
# Auto-calculate hours if not specified
if hours is None:
if not target_date or target_date == datetime.now().strftime("%Y-%m-%d"):
# Today: show from midnight to current time
current_hour = datetime.now().hour
hours = current_hour + 1
else:
# Other dates: show full day (24 hours)
hours = 24
# Get weather data for visualization
result = client.get_city_weather(city, hours=hours, target_date=target_date)
if "weather" in result:
weather_data = result["weather"]
city_name = result.get("display_name", city)
# Create title based on date
if target_date:
title = f"Weather in {city_name} ({target_date})"
else:
current_hour = datetime.now().hour
title = f"Weather in {city_name} (00:00 - {current_hour:02d}:00)"
print(f"📈 Creating visualization for {title}...")
# Create visualizer and plot
visualizer = create_visualizer()
fig = visualizer.plot_weather_overview(weather_data, title)
# Save plot
output_file = f"weather_{city.replace(' ', '_')}_{target_date or 'today'}.png"
fig.savefig(output_file, dpi=150, bbox_inches='tight')
print(f"💾 Chart saved as: {output_file}")
# Show summary
summary = visualizer.get_weather_summary(weather_data)
print(f"📋 Weather Summary:")
print(f" Temperature: {summary['temperature']['min']:.1f}°C - {summary['temperature']['max']:.1f}°C")
print(f" Humidity: {summary['humidity']['min']:.0f}% - {summary['humidity']['max']:.0f}%")
print(f" Max precipitation chance: {summary['precipitation']['max']:.0f}%")
else:
print("❌ Failed to get weather data for visualization")
except Exception as e:
print(f"❌ Visualization demo failed: {e}")
def interactive_mode(api_key=None):
"""Run interactive weather query mode."""
print("🗣️ Interactive Weather Assistant")
print(" Type 'quit' to exit, 'help' for commands")
if not api_key and not os.getenv("GOOGLE_API_KEY"):
print("⚠️ Agent mode disabled (no API key). Using basic client only.")
agent = None
client = WeatherMCPClient()
else:
try:
agent = WeatherAgent(api_key=api_key)
client = agent.client
print("✅ Agent mode enabled")
except Exception as e:
print(f"⚠️ Agent initialization failed: {e}")
print(" Falling back to basic client mode")
agent = None
client = WeatherMCPClient()
while True:
try:
user_input = input("\n🌤️ Weather query: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("👋 Goodbye!")
break
elif user_input.lower() == 'help':
print("📚 Available commands:")
print(" - Ask any weather question (e.g., 'Weather in Tokyo')")
print(" - 'viz [city] [date]' - Show weather visualization (date: YYYY-MM-DD, optional)")
print(" - 'raw [city] [date]' - Show raw weather data (date: YYYY-MM-DD, optional)")
print(" - 'quit' - Exit")
print("\n💡 Examples:")
print(" viz Tokyo # Today's weather")
print(" viz Tokyo 2025-10-01 # Specific date")
print(" raw London 2025-09-30 # Raw data for specific date")
continue
elif user_input.startswith('viz '):
parts = user_input[4:].strip().split()
if parts:
city = parts[0]
target_date = parts[1] if len(parts) > 1 else ""
try:
from datetime import datetime
# Auto-calculate hours
if not target_date or target_date == datetime.now().strftime("%Y-%m-%d"):
# Today: show from midnight to current time
current_hour = datetime.now().hour
hours_to_show = current_hour + 1
else:
# Other dates: show full day
hours_to_show = 24
result = client.get_city_weather(city, hours=hours_to_show, target_date=target_date)
if "weather" in result:
city_name = result.get('display_name', city)
if target_date:
title = f"Weather in {city_name} ({target_date})"
else:
current_hour = datetime.now().hour
title = f"Weather in {city_name} (00:00 - {current_hour:02d}:00)"
plot_weather_trends_horizontal(result["weather"], title)
else:
print(f"❌ Could not get weather data for {city}")
except Exception as e:
print(f"❌ Visualization error: {e}")
else:
print("❌ Please specify a city: viz [city] [date]")
continue
elif user_input.startswith('raw '):
parts = user_input[4:].strip().split()
if parts:
city = parts[0]
target_date = parts[1] if len(parts) > 1 else ""
try:
# Use 12 hours for better data coverage
result = client.get_city_weather(city, hours=12, target_date=target_date)
if target_date:
print(f"📋 Raw data for {city} ({target_date}):")
else:
print(f"📋 Raw data for {city}:")
print(result)
except Exception as e:
print(f"❌ Query error: {e}")
else:
print("❌ Please specify a city: raw [city] [date]")
continue
if not user_input:
continue
# Process weather query
if agent:
answer = agent.ask_weather(user_input)
print(f"🤖 {answer}")
else:
# Basic mode - try to extract city and show weather
words = user_input.split()
cities = ["Tokyo", "London", "Paris", "New York", "Beijing"]
found_city = None
for word in words:
if word.capitalize() in cities:
found_city = word.capitalize()
break
if found_city:
result = client.get_city_weather(found_city, hours=6)
if "weather" in result:
weather = result["weather"]["hourly"]
temp_range = f"{min(weather['temperature_2m']):.1f}°C - {max(weather['temperature_2m']):.1f}°C"
print(f"🌡️ {found_city} temperature range (next 6h): {temp_range}")
else:
print(f"❌ Could not get weather for {found_city}")
else:
print("🤔 Please specify a city or use agent mode for natural language queries")
except KeyboardInterrupt:
print("\n👋 Goodbye!")
break
except Exception as e:
print(f"❌ Error: {e}")
def main():
"""Main application entry point."""
parser = argparse.ArgumentParser(description="Weather MCP Application")
parser.add_argument("--mode", choices=["server", "client", "agent", "viz", "interactive"],
default="interactive", help="Application mode")
parser.add_argument("--api-key", help="Google API key (overrides environment variable)")
parser.add_argument("--city", default="Tokyo", help="City name for demos")
parser.add_argument("--hours", type=int, default=None, help="Hours to forecast (default: auto-calculate)")
parser.add_argument("--date", default="", help="Target date in YYYY-MM-DD format (for viz mode)")
args = parser.parse_args()
print("🌤️ Weather MCP Application")
print("=" * 50)
# Check environment
env_ok = setup_environment()
if args.mode == "server":
run_server()
elif args.mode == "client":
run_client_demo()
elif args.mode == "agent":
if args.api_key or env_ok:
run_agent_demo(args.api_key)
else:
print("❌ Google API key required for agent mode")
elif args.mode == "viz":
run_visualization_demo(args.city, args.date, args.hours)
elif args.mode == "interactive":
interactive_mode(args.api_key)
else:
print(f"❌ Unknown mode: {args.mode}")
if __name__ == "__main__":
main()