Spaces:
Sleeping
Sleeping
| """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"] | |
| } | |
| } | |
| ] | |