YouTube 30-Agent Team: The Data Layer
n8n user built 30 YouTube agents. Where each gets its intelligence: trending topics from search, competitor analysis from YouTube, audience signals from Reddit.
An n8n user built 30 specialized YouTube agents, each handling a different task: topic research, thumbnail analysis, title optimization, comment moderation, competitor tracking, and more. The critical question they face is the data layer: where does each agent get its intelligence? The answer is a combination of search APIs for trending topics and competitor data, YouTube API for channel metrics, and Reddit monitoring for audience signals.
The 30-agent architecture
Each agent in the team has a narrow responsibility. Topic Research Agent finds trending subjects. SEO Agent optimizes titles and descriptions. Thumbnail Agent analyzes click-through patterns. Competitor Agent tracks rival channels. Comment Agent moderates and identifies engagement opportunities. Scheduling Agent picks optimal publish times. The orchestrator routes tasks between them. But every agent needs data to work with.
Data layer breakdown by agent type
- Topic Research: search API for trending queries, People Also Ask data, related searches
- Competitor Tracking: search API for competitor rankings, SERP features, AI Overview citations
- SEO Optimization: search API for keyword data, featured snippets, PAA patterns
- Audience Research: Reddit monitoring for what viewers discuss, complain about, request
- Trend Detection: search API for rising queries in niche, news mentions
Topic research agent data pipeline
import requests, os
H = {"x-api-key": os.environ["SCAVIO_API_KEY"]}
BASE = "https://api.scavio.dev/api/v1/search"
def topic_research_agent(niche: str, existing_topics: list[str]) -> dict:
"""Find new video topics by analyzing search data."""
# Get People Also Ask for the niche
resp = requests.post(BASE, headers=H,
json={"platform": "google", "query": f"{niche} tutorial 2026"},
timeout=10)
data = resp.json()
paa_questions = [q.get("question") for q in data.get("paa", [])]
related = data.get("related_searches", [])
# Filter out topics already covered
new_topics = [q for q in paa_questions if not any(
existing.lower() in q.lower() for existing in existing_topics
)]
# Check which topics have AI Overviews (opportunity to be cited)
topics_with_aio = []
for topic in new_topics[:5]:
check = requests.post(BASE, headers=H,
json={"platform": "google", "query": topic}, timeout=10)
if check.json().get("ai_overview"):
topics_with_aio.append(topic)
return {
"new_topics": new_topics,
"related_searches": related,
"aio_opportunities": topics_with_aio
}Competitor tracking agent
def competitor_tracking_agent(competitors: list[str], keywords: list[str]) -> dict:
"""Monitor competitor visibility across search."""
report = {"competitors": {}}
for comp in competitors:
report["competitors"][comp] = {"rankings": [], "aio_citations": 0}
for kw in keywords:
resp = requests.post(BASE, headers=H,
json={"platform": "google", "query": kw}, timeout=10)
data = resp.json()
# Check organic rankings
organic = data.get("organic", [])
for i, r in enumerate(organic):
if comp.lower() in r.get("link", "").lower():
report["competitors"][comp]["rankings"].append({
"keyword": kw, "position": i + 1
})
# Check AI Overview citations
aio_sources = data.get("ai_overview", {}).get("sources", [])
if any(comp.lower() in s.get("link", "").lower() for s in aio_sources):
report["competitors"][comp]["aio_citations"] += 1
return reportAudience signal agent (Reddit)
def audience_signal_agent(niche: str, subreddits: list[str]) -> dict:
"""Find what your audience is asking for on Reddit."""
signals = {"requests": [], "complaints": [], "trends": []}
for sub in subreddits:
query = f"site:reddit.com/r/{sub} {niche} wish OR need OR want OR looking for"
resp = requests.post(BASE, headers=H,
json={"platform": "google", "query": query}, timeout=10)
results = resp.json().get("organic", [])
for r in results:
snippet = r.get("snippet", "")
title = r.get("title", "")
signals["requests"].append({
"subreddit": sub,
"title": title,
"snippet": snippet[:200],
"url": r.get("link")
})
return signalsOrchestrating the data flow
In n8n, each agent is a workflow node. The data flow is: Topic Research runs daily, outputs candidate topics. SEO Agent takes each topic, runs keyword analysis, scores by difficulty and opportunity. Competitor Agent checks if competitors already covered the topic. Audience Agent validates demand from Reddit signals. The orchestrator picks the top 3 topics that pass all gates. Total data cost: roughly 50-100 search queries per daily run = $0.25-$0.50/day.
Cost of the data layer
Running 30 agents daily with search data: approximately 100-200 queries/day depending on how many keywords and competitors you track. Monthly cost: $15-30 on the Scavio plan ($30/mo for 7K credits). Compare to running each agent manually (hours of research per topic) or using multiple tools (Ahrefs at $129/mo + social listening at $200/mo). The API-first data layer is both cheaper and automatable.
Scaling the team
Adding a new agent to the team means: define its data needs, wire it to the search API with the right queries, add it to the n8n workflow. No new tool subscriptions, no new logins. The search API is the universal data provider. Whether the agent needs keyword data, competitor rankings, trending topics, or audience signals, it all comes from the same endpoint with different queries.