A news aggregator that combines Google News articles, Reddit discussions, and YouTube videos for any topic gives a more complete picture than any single source. Google News provides editorial coverage, Reddit surfaces community reactions, and YouTube captures video commentary. This tutorial builds a multi-source aggregator using the Scavio API that queries all three platforms, normalizes the results into a common format, deduplicates by URL, and ranks by a combined relevance score.
Prerequisites
- Python 3.10 or higher
- requests library installed
- A Scavio API key
- Topics or keywords to aggregate news for
Walkthrough
Step 1: Query all three sources
Fetch Google News results, Reddit posts, and YouTube videos for the same topic using the Scavio API.
Python
from concurrent.futures import ThreadPoolExecutor
def fetch_google_news(topic: str) -> list[dict]:
r = requests.post(ENDPOINT, headers={"Authorization": f"Bearer {API_KEY}"},
json={"query": f"{topic} news", "gl": "us"})
r.raise_for_status()
return r.json().get("news_results", r.json().get("organic_results", []))
def fetch_reddit(topic: str) -> list[dict]:
r = requests.post(ENDPOINT, headers={"Authorization": f"Bearer {API_KEY}"},
json={"platform": "reddit", "query": topic})
r.raise_for_status()
return r.json().get("data", {}).get("posts", [])
def fetch_youtube(topic: str) -> list[dict]:
r = requests.post(ENDPOINT, headers={"Authorization": f"Bearer {API_KEY}"},
json={"platform": "youtube", "query": topic})
r.raise_for_status()
return r.json().get("videos", [])Step 2: Normalize results into a common format
Transform results from each platform into a uniform structure with title, url, source, and snippet.
Python
def normalize_google(item: dict) -> dict:
return {"title": item.get("title"), "url": item.get("link"), "source": "google", "snippet": item.get("snippet", ""), "date": item.get("date")}
def normalize_reddit(post: dict) -> dict:
return {"title": post.get("title"), "url": post.get("url"), "source": "reddit", "snippet": f"r/{post.get('subreddit', '')}", "date": post.get("timestamp")}
def normalize_youtube(video: dict) -> dict:
return {"title": video.get("title"), "url": video.get("url"), "source": "youtube", "snippet": video.get("description", "")[:100], "date": video.get("published_at")}Step 3: Deduplicate by URL
Remove duplicate entries that appear across sources using URL as the deduplication key.
Python
def deduplicate(items: list[dict]) -> list[dict]:
seen = {}
for item in items:
url = item.get("url", "")
if url and url not in seen:
seen[url] = item
return list(seen.values())Step 4: Output the aggregated feed
Print the combined, deduplicated feed grouped by source for easy consumption.
Python
def aggregate(topic: str) -> list[dict]:
with ThreadPoolExecutor(max_workers=3) as ex:
g = ex.submit(fetch_google_news, topic)
r = ex.submit(fetch_reddit, topic)
y = ex.submit(fetch_youtube, topic)
items = [normalize_google(i) for i in g.result()[:5]]
items += [normalize_reddit(i) for i in r.result()[:5]]
items += [normalize_youtube(i) for i in y.result()[:5]]
return deduplicate(items)Python Example
Python
import os
import requests
from concurrent.futures import ThreadPoolExecutor
# Scavio has one endpoint per platform - there is no dispatcher endpoint and no
# `platform` request param, so the selector lives in your code.
SCAVIO = "https://api.scavio.dev"
SCAVIO_ENDPOINTS = {
"google": "/api/v2/google",
"reddit": "/api/v1/reddit/search",
"youtube": "/api/v1/youtube/search",
"amazon": "/api/v1/amazon/search",
"walmart": "/api/v1/walmart/search",
}
SCAVIO_QUERY_KEY = {"youtube": "search"}
SCAVIO_RESULTS_KEY = {"google": "organic_results", "reddit": "results",
"youtube": "results", "amazon": "products", "walmart": "products"}
def scavio_url(platform):
return SCAVIO + SCAVIO_ENDPOINTS[platform or "google"]
def scavio_body(platform, query):
return {SCAVIO_QUERY_KEY.get(platform or "google", "query"): query}
def scavio_payload(payload, platform="google"):
"""Google v2 passes Google's response through as-is; every other endpoint
wraps its payload in `data`. Item fields differ per platform (see
https://scavio.dev/docs), so only the result list is normalised here."""
platform = platform or "google"
out = payload if platform == "google" else payload["data"]
return {**out, "results": out.get(SCAVIO_RESULTS_KEY[platform], [])}
API_KEY = os.environ.get("SCAVIO_API_KEY", "your_scavio_api_key")
def fetch(body: dict) -> dict:
r = requests.post(scavio_url(body.get("platform")), headers={"Authorization": f"Bearer {API_KEY}"}, json=scavio_body(body.get("platform"), body.get("query") or body.get("search")))
r.raise_for_status()
return scavio_payload(r.json(), body.get("platform"))
def aggregate(topic: str) -> list[dict]:
with ThreadPoolExecutor(max_workers=3) as ex:
g = ex.submit(fetch, {"query": f"{topic} news", "country_code": "us"})
r = ex.submit(fetch, {"platform": "reddit", "query": topic})
y = ex.submit(fetch, {"platform": "youtube", "query": topic})
items = []
for i in (g.result().get("news_results") or g.result().get("results", []))[:5]:
items.append({"src": "google", "title": i.get("title"), "url": i.get("link")})
for p in r.result().get("data", {}).get("results", [])[:5]:
items.append({"src": "reddit", "title": p.get("title"), "url": p.get("url")})
for v in y.result().get("results", [])[:5]:
items.append({"src": "youtube", "title": v.get("title"), "url": v.get("url")})
return items
if __name__ == "__main__":
for item in aggregate("AI agents 2026"):
print(f"[{item['src']:>7}] {item['title'][:60]}")JavaScript Example
JavaScript
// Scavio has one endpoint per platform - there is no dispatcher endpoint and no
// `platform` request param, so the selector lives in your code.
const SCAVIO = "https://api.scavio.dev";
const SCAVIO_ENDPOINTS = {
google: "/api/v2/google",
reddit: "/api/v1/reddit/search",
youtube: "/api/v1/youtube/search",
amazon: "/api/v1/amazon/search",
walmart: "/api/v1/walmart/search",
};
const SCAVIO_QUERY_KEY = { youtube: "search" };
const SCAVIO_RESULTS_KEY = { google: "organic_results", reddit: "results",
youtube: "results", amazon: "products", walmart: "products" };
const scavioUrl = (platform) => SCAVIO + SCAVIO_ENDPOINTS[platform || "google"];
const scavioBody = (platform, query) =>
({ [SCAVIO_QUERY_KEY[platform || "google"] || "query"]: query });
// Google v2 passes Google's response through as-is; every other endpoint wraps
// its payload in `data`. Item fields differ per platform (see
// https://scavio.dev/docs), so only the result list is normalised here.
function scavioPayload(json, platform = "google") {
const p = platform || "google";
const out = p === "google" ? json : json.data;
return { ...out, results: out[SCAVIO_RESULTS_KEY[p]] || [] };
}
const API_KEY = process.env.SCAVIO_API_KEY || "your_scavio_api_key";
async function call(body) {
const res = await fetch(scavioUrl(body.platform), {
method: "POST",
headers: { "Authorization": `Bearer ${API_KEY}`, "Content-Type": "application/json" },
body: JSON.stringify(scavioBody(body.platform, body.query ?? body.search))
});
return scavioPayload(res.json(), body.platform);
}
async function aggregate(topic) {
const [g, r, y] = await Promise.all([
call({ query: `${topic} news`, country_code: "us" }),
call({ platform: "reddit", query: topic }),
call({ platform: "youtube", query: topic })
]);
const items = [];
(g.news_results || g.organic_results || []).slice(0, 5).forEach(i => items.push({ src: "google", title: i.title }));
(r.data?.results || []).slice(0, 5).forEach(p => items.push({ src: "reddit", title: p.title }));
(y.videos || []).slice(0, 5).forEach(v => items.push({ src: "youtube", title: v.title }));
return items;
}
aggregate("AI agents 2026").then(items => items.forEach(i => console.log(`[${i.src}] ${i.title}`))).catch(console.error);Expected Output
JSON
[ google] OpenAI Launches Agent Building Platform for Enterprise
[ google] Anthropic Expands Claude Agent Capabilities
[ reddit] Has anyone deployed AI agents in production yet?
[ reddit] Best frameworks for building AI agents in 2026
[youtube] I Built an AI Agent That Runs My Business
[youtube] AI Agents Explained - Complete 2026 Guide