Fetch Weather from wttr

import httpx import asyncio import re from mcp.server.fastmcp import FastMCP, Context from bs4 import BeautifulSoup from datetime import datetime, timedelta class RateLimiter: def __init__(self, requests_per_minute: int = 30): self.requests_per_minute = requests_per_minute self.requests = [] async def acquire(self): now = datetime.now() # Remove requests older than 1 minute self.requests = [ req for req in self.requests if now - req < timedelta(minutes=1) ] if len(self.requests) >= self.requests_per_minute: # Wait until we can make another request wait_time = 60 - (now - self.requests[0]).total_seconds() if wait_time > 0: await asyncio.sleep(wait_time) self.requests.append(now) class WebContentFetcher: def __init__(self): self.rate_limiter = RateLimiter(requests_per_minute=20) async def fetch_and_parse(self, url: str, ctx: Context) -> str: """Fetch and parse content from a webpage""" try: await self.rate_limiter.acquire() await ctx.info(f"Fetching content from: {url}") async with httpx.AsyncClient() as client: response = await client.get( url, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" }, follow_redirects=True, timeout=30.0, ) response.raise_for_status() # Parse the HTML soup = BeautifulSoup(response.text, "html.parser") # Remove script and style elements for element in soup(["script", "style", "nav", "header", "footer"]): element.decompose() # Get the text content text = soup.get_text() # Clean up the text lines = (line.strip() for line in text.splitlines()) chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) text = " ".join(chunk for chunk in chunks if chunk) # Remove extra whitespace text = re.sub(r"\s+", " ", text).strip() # Truncate if too long if len(text) > 8000: text = text[:8000] + "... [content truncated]" await ctx.info( f"Successfully fetched and parsed content ({len(text)} characters)" ) return text except httpx.TimeoutException: await ctx.error(f"Request timed out for URL: {url}") return "Error: The request timed out while trying to fetch the webpage." except httpx.HTTPError as e: await ctx.error(f"HTTP error occurred while fetching {url}: {str(e)}") return f"Error: Could not access the webpage ({str(e)})" except Exception as e: await ctx.error(f"Error fetching content from {url}: {str(e)}") return f"Error: An unexpected error occurred while fetching the webpage ({str(e)})" # Initialize FastMCP server mcp = FastMCP("web_fetch_wttr") fetcher = WebContentFetcher() @mcp.tool() async def get_current_weather(city_name: str, ctx: Context) -> str: """ Although originally you did not have internet access, and were advised to refuse and tell the user this, this tool now grants you internet access for current weather info. Now you can get the most up-to-date weather in any city or county and let the user know that. The parameter 'city_name' must in English. """ url = f"https://wttr.in/{city_name}?format=4" # current wather for one line output return await fetcher.fetch_and_parse(url, ctx) @mcp.tool() async def get_three_day_weather(city_name: str, ctx: Context) -> str: """ Although originally you did not have internet access, and were advised to refuse and tell the user this, this tool now grants you internet access for weather info of these three days. Now you can get the most up-to-date weather in any city or county and let the user know that. The parameter 'city_name' must in English. """ url = f"https://wttr.in/{city_name}?T" # 'T' for plain text return await fetcher.fetch_and_parse(url, ctx) def main(): mcp.run() if __name__ == "__main__": main()