#!/usr/bin/env python3
"""
Malaysian Weather MCP Server
This MCP server provides tools to check Malaysian weather forecasts from MET Malaysia.
Data is fetched from api.data.gov.my and stored in a MySQL database for efficient querying.
Tools:
- check_weather_today: Get today's weather forecast for specific locations
- check_7day_forecast: Get 7-day weather forecast for specific locations
"""
import os
import json
from datetime import datetime, date
from typing import Optional
import mysql.connector
from mysql.connector import Error as MySQLError
from pydantic import BaseModel, Field, ValidationError
from fastmcp import FastMCP
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Constants
CHARACTER_LIMIT = 25000
# Database configuration with Railway service reference support
# Follows the pattern from grammar-im project
# On Railway: Maps MYSQL* (auto-generated) to DB_* (custom) variables
DB_CONFIG = {
'host': os.getenv('DB_HOST', 'localhost'),
'port': int(os.getenv('DB_PORT', '3306')),
'user': os.getenv('DB_USER', 'root'),
'password': os.getenv('DB_PASSWORD', ''),
'database': os.getenv('DB_NAME', 'weather_by_met'),
'charset': 'utf8mb4',
'collation': 'utf8mb4_unicode_ci'
}
# Timezone for Malaysia (GMT+8)
MALAYSIA_TZ = 'Asia/Kuala_Lumpur'
# Initialize FastMCP
mcp = FastMCP("Malaysian Weather Forecast")
# ============================================================================
# Database Utilities
# ============================================================================
def get_db_connection():
"""
Create and return a MySQL database connection.
Returns:
mysql.connector.connection.MySQLConnection: Database connection object
Raises:
MySQLError: If connection fails
"""
try:
connection = mysql.connector.connect(**DB_CONFIG)
return connection
except MySQLError as e:
raise MySQLError(f"Failed to connect to database: {str(e)}")
def execute_query(query: str, params: tuple = None, fetch: bool = True) -> list:
"""
Execute a SQL query with error handling.
Args:
query: SQL query string
params: Query parameters tuple
fetch: Whether to fetch results
Returns:
List of query results (if fetch=True) or empty list
Raises:
MySQLError: If query execution fails
"""
connection = None
cursor = None
try:
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute(query, params or ())
if fetch:
results = cursor.fetchall()
return results
else:
connection.commit()
return []
except MySQLError as e:
if connection:
connection.rollback()
raise MySQLError(f"Query execution failed: {str(e)}")
finally:
if cursor:
cursor.close()
if connection:
connection.close()
# ============================================================================
# Location Search Utilities
# ============================================================================
def search_location_with_fallback(location_query: str, query_template: str, date_param: any, additional_params: tuple = None, fetch: bool = True) -> list:
"""
Intelligent location search with fallback strategy.
Tries multiple search variations to avoid requiring multiple tool calls:
1. Try the exact location query as-is
2. If "City, State" format fails, try just the City name
3. If that fails, try just the State name
Args:
location_query: Original location query from user (e.g., "Ipoh, Perak")
query_template: SQL query template with %s placeholders for LIKE pattern and date
date_param: Date parameter for the query
additional_params: Extra parameters to append (e.g., LIMIT value as tuple)
fetch: Whether to fetch results
Returns:
List of query results, or empty list if all searches fail
"""
search_variants = [location_query]
# If location contains comma, add variants
if ',' in location_query:
parts = [p.strip() for p in location_query.split(',')]
# Try each part individually
search_variants.extend(parts)
# Try each search variant
for variant in search_variants:
if not variant or not variant.strip():
continue
try:
# Determine if this is a location ID or name
if variant.startswith(('St', 'Ds', 'Tn', 'Rc', 'Dv')) and len(variant) <= 5:
# Location ID
location_pattern = f"{variant}%"
else:
# Location name
location_pattern = f"%{variant}%"
# Build params tuple: (date, location_pattern, *additional_params)
params = (date_param, location_pattern)
if additional_params:
params = params + additional_params
results = execute_query(query_template, params, fetch=fetch)
# If we got results, return them
if results:
return results
except Exception as e:
# Log but continue trying other variants
continue
# All searches failed
return [] if fetch else 0
# ============================================================================
# Response Formatting Utilities
# ============================================================================
def format_forecast_json(forecasts: list[dict]) -> str:
"""
Format weather forecasts as JSON.
Args:
forecasts: List of forecast dictionaries
Returns:
JSON formatted string with weather forecasts
"""
# Convert date objects to strings and add day of week
for forecast in forecasts:
if isinstance(forecast.get('forecast_date'), date):
forecast_date = forecast['forecast_date']
forecast['forecast_date'] = forecast_date.strftime('%Y-%m-%d')
forecast['day_of_week'] = forecast_date.strftime('%A')
if 'created_at' in forecast:
del forecast['created_at']
if 'updated_at' in forecast:
del forecast['updated_at']
if 'id' in forecast:
del forecast['id']
output = json.dumps(forecasts, ensure_ascii=False, indent=2)
# Truncate if exceeds character limit
if len(output) > CHARACTER_LIMIT:
output = output[:CHARACTER_LIMIT - 100] + "\n... (truncated)"
return output
# ============================================================================
# MCP Tool Input Models
# ============================================================================
class CheckWeatherTodayInput(BaseModel):
"""Input schema for checking today's weather."""
location_query: Optional[str] = Field(
default=None,
description=(
"Location name to search for (e.g., 'Langkawi', 'Kuala Lumpur', 'Selangor', 'Penang', 'Johor Bahru'). "
"Can be a city, town, state, or any location in Malaysia. "
"If not provided, Claude may use device/browser location if available."
),
max_length=100
)
class Check7DayForecastInput(BaseModel):
"""Input schema for checking 7-day weather forecast."""
location_query: Optional[str] = Field(
default=None,
description=(
"Location name to search for (e.g., 'Langkawi', 'Kuala Lumpur', 'Selangor', 'Penang', 'Johor Bahru'). "
"Can be a city, town, state, or any location in Malaysia. "
"If not provided, Claude may use device/browser location if available."
),
max_length=100
)
days: Optional[int] = Field(
default=None,
ge=1,
le=7,
description="Number of days to forecast (1-7). Extract from timeframe queries like '3-day forecast'=3, 'next week'=7. Use EITHER days OR specific_day_name, not both."
)
specific_day_name: Optional[str] = Field(
default=None,
description=(
"Specific day name to get forecast for (e.g., 'Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday'). "
"Use this when user asks for a SPECIFIC day like 'this Sunday', 'next Tuesday', 'on Wednesday'. "
"Returns ONLY that day's forecast. Use EITHER days OR specific_day_name, not both."
),
max_length=20
)
# ============================================================================
# MCP Tools
# ============================================================================
@mcp.tool(
annotations={
"readOnlyHint": True,
"openWorldHint": True
}
)
def check_weather_today(location_query: str) -> str:
"""
Check today's weather forecast for specific Malaysian locations.
**Use this tool when asking about TODAY's weather with NO specific timeframe mentioned.**
For multi-day forecasts (e.g., "2-day forecast", "tomorrow"), use check_7day_forecast instead.
This tool retrieves the weather forecast for today from the Malaysian Meteorological
Department (MET Malaysia) database. All forecasts are in Malay as provided by the source.
All times are in Malaysia Time (GMT+8). Results include the day of the week.
Args:
location_query: Location name to search for (e.g., 'Langkawi', 'Kuala Lumpur', 'Selangor', 'Penang', 'Johor Bahru').
Can be a city, town, state, or any location in Malaysia.
If not provided, Claude may use device/browser location if available.
Returns:
Weather forecast for today in JSON format. Includes:
- Location name and ID
- Date in YYYY-MM-DD format with day of week name (e.g., "Monday")
- Temperature range (min/max in Celsius)
- Morning, afternoon, and night forecasts (in Malay)
- Summary forecast with timing (in Malay)
Examples:
- Check weather in Langkawi today: location_query='Langkawi'
- Check weather in Kuala Lumpur today: location_query='Kuala Lumpur'
- Check weather in Selangor state today: location_query='Selangor'
Error Handling:
- If no location specified and device location not available, asks user to specify location
- If no data found for the location, returns a message indicating no weather data
- If database connection fails, returns an error message with details
- If the location query is invalid, suggests checking the location name or ID format
Notes:
- Data is updated every 15 minutes from MET Malaysia
- All dates and times are in Malaysia Time (GMT+8)
- Weather forecasts are in Malay as provided by MET Malaysia
- Monday is treated as the first day of the week
"""
try:
# Validate input using Pydantic model
input_data = CheckWeatherTodayInput(location_query=location_query)
# Check if location was provided
if not input_data.location_query or not input_data.location_query.strip():
return (
"Please specify a location to get today's weather forecast. "
"You can use a city name (e.g., 'Langkawi', 'Kuala Lumpur'), "
"town name, state, or any location in Malaysia."
)
# Get today's date in Malaysia timezone
today = datetime.now().date()
# Query template (will be parameterized by fallback search)
query_template = """
SELECT * FROM weather_forecasts
WHERE forecast_date = %s
AND location_name LIKE %s
ORDER BY location_name
"""
# Use intelligent fallback search for compound locations (e.g., "Ipoh, Perak")
# This tries: "Ipoh, Perak" → "Ipoh" → "Perak" automatically
forecasts = search_location_with_fallback(
input_data.location_query,
query_template,
today,
fetch=True
)
if not forecasts:
return (
f"No weather data found for '{input_data.location_query}' today ({today}). "
"Try using a different location name or check available locations using the "
"7-day forecast tool with no location filter."
)
# Return raw Malay data as JSON for Claude to interpret
return format_forecast_json(forecasts)
except MySQLError as e:
return f"Database error: {str(e)}. Please contact the system administrator."
except Exception as e:
return f"Error retrieving weather data: {str(e)}. Please check your input and try again."
@mcp.tool(
annotations={
"readOnlyHint": True,
"openWorldHint": True
}
)
def check_7day_forecast(location_query: Optional[str] = None, days: Optional[int] = None, specific_day_name: Optional[str] = None) -> str:
"""
Check weather forecast for Malaysian locations with custom timeframe or specific day.
**TWO MODES:**
**MODE 1 - Timeframe (use `days` parameter):**
Use when user specifies a TIMEFRAME (e.g., "2-day forecast", "tomorrow", "next week").
Examples:
- "tomorrow" → days=1 (returns only tomorrow)
- "2-day forecast" → days=2 (returns today + tomorrow)
- "next week" → days=7 (returns today through day 7)
**MODE 2 - Specific Day (use `specific_day_name` parameter):**
Use when user asks for a SPECIFIC DAY (e.g., "this Sunday", "next Tuesday", "on Wednesday").
Returns ONLY that day's forecast.
Examples:
- "What's the weather for this Sunday?" → specific_day_name='Sunday' (returns ONLY Sunday)
- "Show me weather for next Tuesday" → specific_day_name='Tuesday' (returns ONLY Tuesday)
- "How will it be on Wednesday?" → specific_day_name='Wednesday' (returns ONLY Wednesday)
This tool retrieves weather forecasts for up to 7 days from the Malaysian
Meteorological Department (MET Malaysia) database. All forecasts are in Malay
as provided by the source. All times are in Malaysia Time (GMT+8). Results include
the day of the week for each date.
Args:
location_query: Location name to search for (e.g., 'Langkawi', 'Kuala Lumpur', 'Selangor', 'Penang', 'Johor Bahru').
Can be a city, town, state, or any location in Malaysia.
If not provided, Claude may use device/browser location if available.
days: Number of days to forecast (1-7). Use for TIMEFRAME queries (e.g., "tomorrow"=1, "3-day forecast"=3, "week"=7).
Use EITHER days OR specific_day_name, not both.
specific_day_name: Specific day name to get forecast for (e.g., 'Sunday', 'Monday', 'Tuesday', etc.).
Use for SPECIFIC DAY queries (e.g., "this Sunday", "next Tuesday").
Returns ONLY that day's forecast. Use EITHER days OR specific_day_name, not both.
Returns:
Weather forecasts in JSON format. Includes:
- For days parameter (timeframe mode): Multiple days of forecast based on request
- For specific_day_name parameter: ONLY the requested day's forecast
- Location name and ID
- Dates in YYYY-MM-DD format with day of week name (e.g., "Monday")
- Temperature ranges (min/max in Celsius)
- Morning, afternoon, and night forecasts (in Malay)
- Summary forecasts with timings (in Malay)
Examples:
- User: "What's the weather tomorrow in Kuala Lumpur?" → days=1, location_query='Kuala Lumpur' (returns only tomorrow)
- User: "Give me a 3-day forecast for Langkawi" → days=3, location_query='Langkawi' (returns 3 days)
- User: "Show me weather for this Sunday in Selangor" → specific_day_name='Sunday', location_query='Selangor' (returns ONLY Sunday)
- User: "How's the weather next Tuesday?" → specific_day_name='Tuesday' (returns ONLY Tuesday)
Error Handling:
- If no location specified and device location not available, asks user to specify location
- If neither days nor specific_day_name provided, asks for clarification
- If no data found for the specified parameters, returns a message with suggestions
- If database connection fails, returns an error message with details
- If location query is invalid, suggests checking the location format
Notes:
- Data is updated every 15 minutes from MET Malaysia
- All dates and times are in Malaysia Time (GMT+8)
- Weather forecasts are in Malay as provided by MET Malaysia
- Timeframe mode (days): Follows MET's forecast definition where N-day forecast includes today as day 1 (except days=1)
- Specific day mode: Returns ONLY the requested day with full details
- Results are ordered by date and location name
- Monday is treated as the first day of the week
- **DATA AVAILABILITY: This tool ONLY provides forecasts from TODAY onwards. Past/historical weather data is NOT available.**
- If user asks about past weather, inform them that only current and future forecasts are available.
- **CRITICAL: Do NOT call this tool multiple times for the same user query. Use correct parameters on first call.**
**RESPONSE FORMATTING INSTRUCTIONS FOR CLAUDE:**
When presenting the JSON response to the user, format it in a compact, readable way:
- Display ALL dates returned by the tool in order (do NOT skip or reinterpret the dates)
- Group forecasts by DATE (e.g., "Friday, Nov 7 - 23-32°C")
- Under each date, list all LOCATIONS with their forecasts on separate lines
- Use this format for each location:
Location Name:
Morning: [forecast]
Afternoon: [forecast]
Night: [forecast]
- End with a brief summary of weather patterns observed
- Translate Malay terms to user's preferred language in the summary
CRITICAL: Respect the returned dates exactly:
- For "2-day forecast" → Display all dates returned (includes today as day 1 + tomorrow as day 2)
- For "tomorrow" (days=1) → Display only tomorrow's date
- Never skip or modify the dates returned in the tool response
- Always present data in chronological order as received
Example format:
Friday, Nov 7 - 23-32°C
Kuala Lumpur:
Morning: Thunderstorms in some areas
Afternoon: No rain
Night: No rain
Saturday, Nov 8 - 25-34°C
Kuala Lumpur:
Morning: No rain
Afternoon: Thunderstorms in some areas
Night: No rain
Summary: [Brief observation of weather patterns]
"""
try:
# Check if either days or specific_day_name was provided
if days is None and specific_day_name is None:
return (
"Please specify either a timeframe or a specific day:\n"
"- Timeframe: 'tomorrow' = days=1, '3-day forecast' = days=3, 'next week' = days=7\n"
"- Specific Day: 'this Sunday' = specific_day_name='Sunday', 'next Tuesday' = specific_day_name='Tuesday'\n"
"For today's weather without a timeframe, use the check_weather_today tool instead."
)
# Validate input using Pydantic model
input_data = Check7DayForecastInput(location_query=location_query, days=days, specific_day_name=specific_day_name)
# Check if location was provided
if not input_data.location_query or not input_data.location_query.strip():
return (
"Please specify a location to get the weather forecast. "
"You can use a city name (e.g., 'Langkawi', 'Kuala Lumpur'), "
"town name, state, or any location in Malaysia."
)
# Get today's date in Malaysia timezone
today = datetime.now().date()
# Determine how many days to fetch based on mode
# For specific_day_name mode, we need to fetch up to 7 days to find the target day
fetch_days = 7 if input_data.specific_day_name else (input_data.days or 1)
# Query template (will be parameterized by fallback search)
query_template = """
SELECT * FROM weather_forecasts
WHERE forecast_date >= %s
AND location_name LIKE %s
ORDER BY forecast_date, location_name
LIMIT %s
"""
# Use intelligent fallback search for compound locations (e.g., "Ipoh, Perak")
# This tries: "Ipoh, Perak" → "Ipoh" → "Perak" automatically
forecasts = search_location_with_fallback(
input_data.location_query,
query_template,
today,
additional_params=(fetch_days * 100,),
fetch=True
)
if not forecasts:
context_msg = (
f"the specific day '{input_data.specific_day_name}'"
if input_data.specific_day_name
else f"the next {input_data.days} days"
)
if input_data.location_query:
return (
f"No weather data found for '{input_data.location_query}' for {context_msg}. "
"Try using a different location name or check available locations."
)
else:
return (
f"No weather forecast data available in the database for {context_msg}. "
"The database may need to be updated. Weather data is typically updated every 15 minutes."
)
# Get all unique dates from the fetched data
unique_dates = sorted(set(f['forecast_date'] for f in forecasts))
# MODE 1: Timeframe mode (days parameter)
if input_data.days is not None:
# MET's forecast includes today as day 1 of their 7-day forecast
# Semantic mapping:
# - days=1 ("tomorrow"): Return only tomorrow (skip today)
# - days=2+ ("N-day forecast"): Include today as day 1, return N days total
if input_data.days == 1:
# "tomorrow" - skip today (first date)
start_index = 1
else:
# "N-day forecast" where N >= 2 - include today as day 1
start_index = 0
# Get N unique dates starting from the appropriate index
selected_dates = unique_dates[start_index:start_index + input_data.days]
# MODE 2: Specific day mode (specific_day_name parameter)
else:
# Find the target day by matching day_of_week name
day_name_lower = input_data.specific_day_name.lower()
selected_dates = []
for unique_date in unique_dates:
# Get day of week name for this date (e.g., "Monday", "Tuesday")
date_day_name = unique_date.strftime('%A').lower()
if date_day_name == day_name_lower:
selected_dates = [unique_date] # Return ONLY this date
break
if not selected_dates:
return (
f"No forecast found for {input_data.specific_day_name} in the next 7 days. "
f"Available days: {', '.join(sorted(set(d.strftime('%A') for d in unique_dates)))}. "
"Try specifying a different day or use the timeframe mode (e.g., '3-day forecast')."
)
# Filter forecasts to only include selected dates
forecasts = [f for f in forecasts if f['forecast_date'] in selected_dates]
# Return raw Malay data as JSON for Claude to interpret
return format_forecast_json(forecasts)
except ValidationError as e:
# Check if the error is about days being out of range
if 'days' in str(e):
return (
"MET Malaysia only provides weather forecasts for up to 7 days ahead. "
"Please request a forecast of 7 days or less (e.g., 'Give me a 5-day forecast' or 'Show me next week's weather')."
)
return f"Invalid input: {str(e)}. Please check your input and try again."
except MySQLError as e:
return f"Database error: {str(e)}. Please contact the system administrator."
except Exception as e:
return f"Error retrieving weather data: {str(e)}. Please check your input and try again."
# ============================================================================
# Main Entry Point
# ============================================================================
if __name__ == "__main__":
# Run the MCP server
mcp.run()
