"""MCP Server for News Fetching""" import requests import feedparser from typing import List, Dict, Any from datetime import datetime import time class NewsMCPServer: """MCP Server for fetching and curating news""" def __init__(self): self.name = "news_server" self.description = "Fetches latest news from various sources" # More reliable RSS feeds with fallbacks self.news_feeds = { "technology": [ "https://rss.cnn.com/rss/edition.rss", # CNN Tech "https://feeds.feedburner.com/oreilly/radar", # O'Reilly Radar "https://www.wired.com/feed/rss" # Wired ], "world": [ "https://rss.cnn.com/rss/edition.rss", # CNN World "https://feeds.reuters.com/reuters/topNews", # Reuters Top News "https://rss.nytimes.com/services/xml/rss/nyt/World.xml" # NYT World ], "business": [ "https://feeds.reuters.com/reuters/businessNews", # Reuters Business "https://rss.cnn.com/rss/money_latest.rss", # CNN Money ], "entertainment": [ "https://rss.cnn.com/rss/edition_entertainment.rss", # CNN Entertainment "https://www.rollingstone.com/feed", # Rolling Stone ], "science": [ "https://rss.cnn.com/rss/edition_space.rss", # CNN Science "https://feeds.feedburner.com/sciencealert-latestnews", # Science Alert ] } # NewsAPI.org (free tier - 100 requests/day) # You can get a free API key from https://newsapi.org/ self.newsapi_key = None # Optional - can be added later self.newsapi_base = "https://newsapi.org/v2" def fetch_news(self, category: str = "world", limit: int = 5) -> List[Dict[str, Any]]: """ Fetch latest news from RSS feeds with better error handling Args: category: News category (technology, world, business, entertainment, science) limit: Number of news items to return Returns: List of news items """ all_news = [] # Try RSS feeds first feeds = self.news_feeds.get(category, self.news_feeds["world"]) for feed_url in feeds: try: # Add timeout and headers for better reliability headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } # Parse feed with timeout feed = feedparser.parse(feed_url) # Check if feed is valid if not feed.entries: continue for entry in feed.entries[:limit]: # Safely extract fields title = str(entry.get("title", "")).strip() summary = str(entry.get("summary", entry.get("description", ""))).strip() link = str(entry.get("link", "")).strip() published = str(entry.get("published", entry.get("updated", ""))).strip() if not title: # Skip if no title continue # Truncate summary safely if summary: summary = summary[:200] + "..." if len(summary) > 200 else summary else: summary = "No summary available." news_item = { "title": title, "summary": summary, "link": link if link else "#", "published": published if published else datetime.now().strftime("%Y-%m-%d"), "category": category } all_news.append(news_item) if len(all_news) >= limit: break if len(all_news) >= limit: break except Exception as e: print(f"Error fetching from {feed_url}: {e}") continue # If we got some news, return it if all_news: return all_news[:limit] # Fallback to demo news if all feeds fail print(f"All feeds failed for {category}, using demo news") return self._get_demo_news(category, limit) def _get_demo_news(self, category: str, limit: int) -> List[Dict[str, Any]]: """Return demo news items when RSS feeds fail""" demo_news = { "technology": [ { "title": "AI Breakthrough: New Language Model Achieves Human-Level Understanding", "summary": "Researchers announce significant advancement in artificial intelligence...", "link": "#", "published": datetime.now().strftime("%Y-%m-%d"), "category": "technology" }, { "title": "Tech Giants Unite for Sustainable Computing Initiative", "summary": "Major technology companies announce partnership to reduce carbon footprint...", "link": "#", "published": datetime.now().strftime("%Y-%m-%d"), "category": "technology" } ], "world": [ { "title": "Global Leaders Meet for Climate Summit", "summary": "World leaders gather to discuss climate action and sustainability...", "link": "#", "published": datetime.now().strftime("%Y-%m-%d"), "category": "world" } ], "business": [ { "title": "Markets Show Strong Growth in Tech Sector", "summary": "Technology stocks lead market gains as investors show confidence...", "link": "#", "published": datetime.now().strftime("%Y-%m-%d"), "category": "business" } ], "entertainment": [ { "title": "New Music Festival Announces Stellar Lineup", "summary": "Major artists confirmed for summer music festival...", "link": "#", "published": datetime.now().strftime("%Y-%m-%d"), "category": "entertainment" } ], "science": [ { "title": "Scientists Discover New Species in Deep Ocean", "summary": "Marine biologists announce discovery of previously unknown deep-sea creatures...", "link": "#", "published": datetime.now().strftime("%Y-%m-%d"), "category": "science" } ] } news_items = demo_news.get(category, demo_news["world"]) return news_items[:limit] def get_personalized_news(self, user_preferences: Dict[str, Any]) -> List[Dict[str, Any]]: """ Get personalized news based on user interests Args: user_preferences: Dictionary with user's news preferences Returns: List of personalized news items """ interests = user_preferences.get("interests", ["world"]) news_items = [] for interest in interests[:3]: # Top 3 interests items = self.fetch_news(category=interest, limit=2) news_items.extend(items) return news_items def get_tools_definition(self) -> List[Dict[str, Any]]: """Return MCP tools definition for this server""" return [ { "name": "fetch_news", "description": "Fetch latest news from various categories", "parameters": { "type": "object", "properties": { "category": { "type": "string", "description": "News category (technology, world, business, entertainment, science)" }, "limit": { "type": "integer", "description": "Number of news items to return" } }, "required": ["category"] } }, { "name": "get_personalized_news", "description": "Get personalized news based on user interests", "parameters": { "type": "object", "properties": { "user_preferences": { "type": "object", "description": "User's news preferences including interests" } }, "required": ["user_preferences"] } } ]