Add MediaStack and NewsData.io connectors for custom news sources

.env.example

@@ -2,3 +2,5 @@ POLYGON_WALLET_PRIVATE_KEY=""
 OPENAI_API_KEY=""
 TAVILY_API_KEY=""
 NEWSAPI_API_KEY=""
+MEDIASTACK_API_KEY=""
+NEWSDATA_API_KEY=""

agents/connectors/mediastack.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from datetime import datetime
+import os
+
+import requests
+
+from agents.utils.objects import Article
+
+
+class MediaStackNews:
+    def __init__(self) -> None:
+        self.api_key = os.getenv("MEDIASTACK_API_KEY")
+        self.base_url = "http://api.mediastack.com/v1/news"
+        self.language = "en"
+        self.country = "us"
+
+    def get_articles_for_cli_keywords(self, keywords: str) -> "list[Article]":
+        query_words = keywords.split(",")
+        all_articles = self.get_articles_for_options(query_words)
+        article_objects: list[Article] = []
+        for _, articles in all_articles.items():
+            for article in articles:
+                article_objects.append(Article(**article))
+        return article_objects
+
+    def get_articles_for_options(
+        self,
+        market_options: "list[str]",
+        date_start: datetime = None,
+        date_end: datetime = None,
+    ) -> dict[str, list[dict]]:
+        all_articles: dict[str, list[dict]] = {}
+        for option in market_options:
+            params = {
+                "access_key": self.api_key,
+                "keywords": option.strip(),
+                "languages": self.language,
+                "countries": self.country,
+                "sort": "published_desc",
+                "limit": 10,
+            }
+            if date_start and date_end:
+                params["date"] = f"{date_start:%Y-%m-%d},{date_end:%Y-%m-%d}"
+            elif date_start:
+                params["date"] = f"{date_start:%Y-%m-%d}"
+
+            response = requests.get(self.base_url, params=params, timeout=10)
+            response.raise_for_status()
+            data = response.json().get("data", [])
+            all_articles[option] = [self._normalize_article(item) for item in data]
+
+        return all_articles
+
+    def _normalize_article(self, item: dict) -> dict:
+        source_name = item.get("source")
+        return {
+            "source": {"name": source_name} if source_name else None,
+            "author": item.get("author"),
+            "title": item.get("title"),
+            "description": item.get("description"),
+            "url": item.get("url"),
+            "urlToImage": item.get("image"),
+            "publishedAt": item.get("published_at"),
+            "content": None,
+        }

agents/connectors/newsdata.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from datetime import datetime
+import os
+
+import requests
+
+from agents.utils.objects import Article
+
+
+class NewsDataNews:
+    def __init__(self) -> None:
+        self.api_key = os.getenv("NEWSDATA_API_KEY")
+        self.base_url = "https://newsdata.io/api/1/news"
+        self.language = "en"
+        self.country = "us"
+
+    def get_articles_for_cli_keywords(self, keywords: str) -> "list[Article]":
+        query_words = keywords.split(",")
+        all_articles = self.get_articles_for_options(query_words)
+        article_objects: list[Article] = []
+        for _, articles in all_articles.items():
+            for article in articles:
+                article_objects.append(Article(**article))
+        return article_objects
+
+    def get_articles_for_options(
+        self,
+        market_options: "list[str]",
+        date_start: datetime = None,
+        date_end: datetime = None,
+    ) -> dict[str, list[dict]]:
+        all_articles: dict[str, list[dict]] = {}
+        for option in market_options:
+            params = {
+                "apikey": self.api_key,
+                "q": option.strip(),
+                "language": self.language,
+                "country": self.country,
+                "size": 10,
+            }
+            if date_start:
+                params["from_date"] = f"{date_start:%Y-%m-%d}"
+            if date_end:
+                params["to_date"] = f"{date_end:%Y-%m-%d}"
+
+            response = requests.get(self.base_url, params=params, timeout=10)
+            response.raise_for_status()
+            results = response.json().get("results", [])
+            all_articles[option] = [self._normalize_article(item) for item in results]
+
+        return all_articles
+
+    def _normalize_article(self, item: dict) -> dict:
+        creator = item.get("creator")
+        if isinstance(creator, list):
+            creator = creator[0] if creator else None
+        source_id = item.get("source_id")
+        return {
+            "source": {"name": source_id} if source_id else None,
+            "author": creator,
+            "title": item.get("title"),
+            "description": item.get("description"),
+            "url": item.get("link"),
+            "urlToImage": item.get("image_url"),
+            "publishedAt": item.get("pubDate"),
+            "content": item.get("content"),
+        }

scripts/python/cli.py

@@ -4,13 +4,17 @@
 from agents.polymarket.polymarket import Polymarket
 from agents.connectors.chroma import PolymarketRAG
 from agents.connectors.news import News
+from agents.connectors.mediastack import MediaStackNews
+from agents.connectors.newsdata import NewsDataNews
 from agents.application.trade import Trader
 from agents.application.executor import Executor
 from agents.application.creator import Creator
 
 app = typer.Typer()
 polymarket = Polymarket()
 newsapi_client = News()
+mediastack_client = MediaStackNews()
+newsdata_client = NewsDataNews()
 polymarket_rag = PolymarketRAG()
 
 
@@ -37,6 +41,24 @@ def get_relevant_news(keywords: str) -> None:
     pprint(articles)
 
 
+@app.command()
+def get_mediastack_news(keywords: str) -> None:
+    """
+    Use MediaStack to query the internet
+    """
+    articles = mediastack_client.get_articles_for_cli_keywords(keywords)
+    pprint(articles)
+
+
+@app.command()
+def get_newsdata_news(keywords: str) -> None:
+    """
+    Use NewsData.io to query the internet
+    """
+    articles = newsdata_client.get_articles_for_cli_keywords(keywords)
+    pprint(articles)
+
+
 @app.command()
 def get_all_events(limit: int = 5, sort_by: str = "number_of_markets") -> None:
     """