| name | content-intelligence |
| description | Use when scanning the data ecosystem for content opportunities, generating a daily content brief, or populating the Content Pipeline Notion database with ranked content ideas tied to dbt Labs content pillars |
Content Intelligence Agent
Scans RSS feeds, Hacker News, Reddit, and news sources for trending data ecosystem conversations. Scores signals by strategic relevance, freshness, and engagement. Writes a ranked daily brief to Notion and creates Content Pipeline database entries.
When to Use
- Morning content brief generation ("run today's content scan")
- Ad-hoc signal scanning ("what's trending in analytics engineering?")
- Populating the Content Pipeline with new ideas
- Researching a specific topic across data ecosystem sources
Mode 1: MCP Playbook (Cowork / Claude Code)
Follow these 6 steps using WebFetch, WebSearch, and Notion MCP tools. No Python or API keys needed.
Step 1 — Gather Signals
Fetch from ALL sources. If one fails, skip it and continue.
RSS Feeds (WebFetch each):
| Source | URL |
|---|
| Snowflake Blog | https://www.snowflake.com/feed/ |
| Databricks Blog | https://www.databricks.com/feed |
| dbt Discourse | https://discourse.getdbt.com/latest.rss |
| r/dataengineering | https://www.reddit.com/r/dataengineering/hot.json?limit=25 |
Hacker News (WebFetch each separately — Algolia doesn't support OR):
https://hn.algolia.com/api/v1/search_by_date?query=dbt&tags=story&hitsPerPage=10https://hn.algolia.com/api/v1/search_by_date?query=%22analytics+engineering%22&tags=story&hitsPerPage=10https://hn.algolia.com/api/v1/search_by_date?query=%22data+transformation%22&tags=story&hitsPerPage=10
Web Search (WebSearch each):
"analytics engineering" trending news"dbt data" news this week"agentic data engineering" OR "AI data pipelines"
Extract from each: title, URL, summary, publish date, engagement metrics.
Step 2 — Filter by Content Pillars
Keep only items matching at least one pillar:
| Pillar | Keywords |
|---|
| Agentic Eng | agentic, ai agent, llm pipeline, autonomous data, vibe coding, copilot data |
| Analytics Eng | analytics engineering, dbt, sql transformation, data testing, data modeling, semantic layer |
| Modern Data Stack | modern data stack, data warehouse, lakehouse, snowflake, databricks, bigquery, orchestration, data observability |
| dbt Product | dbt cloud, dbt core, dbt mesh, dbt explorer, dbt semantic layer |
| Data Culture | data governance, data contracts, data mesh org, data team, analytics team, data literacy |
Discard pure ML/deep learning or unrelated product announcements.
Step 3 — Score (1-10)
For each signal, compute: score = (strategic x 0.5) + (freshness x 0.3) + (engagement x 0.2)
Strategic fit (50%): 10 = directly about dbt/AE, 7 = adjacent with clear angle, 4 = tangential, 1 = barely relevant
Freshness (30%): 10 = today, 7 = yesterday, 4 = 2 days, 1 = 3+ days
Engagement (20%): 10 = viral (100+ points), 7 = strong (50+), 4 = moderate (10+), 1 = low
For each, also generate: a 1-2 sentence content angle, suggested format(s), and engagement level (High/Medium/Low).
Step 4 — Write Notion Brief
Create a page under Content Intelligence Briefs (page ID: 32ebb38e-bda7-8120-8362-dbbba6349d53) using mcp__claude_ai_Notion__notion-create-pages.
Properties: title: "Content Brief -- [YYYY-MM-DD]", icon: "📊"
Page content (Notion-flavored Markdown):
## Top 3 Opportunities
### 1. [Title]
**Pillar:** [name] · **Score:** [X.X]/10 · **Source:** [platform]
[2-3 sentence content angle]
**Suggested format:** [format(s)]
---
[repeat for #2, #3]
## Full Ranked List
<table header-row="true" fit-page-width="true">
<tr>
<td>**#**</td>
<td>**Opportunity**</td>
<td>**Pillar**</td>
<td>**Score**</td>
<td>**Source**</td>
<td>**Why Now**</td>
</tr>
[one row per opportunity, sorted by score desc]
</table>
## Raw Signals
- [Title](URL) -- 1-line context
[for each source signal]
Step 5 — Create Content Pipeline DB Entries
For the top 10, create entries in the Content Pipeline database using mcp__claude_ai_Notion__notion-create-pages with data_source_id: 3e757691-8564-403f-bab4-4ad466f98683.
Properties per entry:
| Property | Value |
|---|
"Title" | opportunity headline |
"Stage" | "Backlog" |
"Pillar" | JSON array from ["Agentic Eng", "Analytics Eng", "Modern Data Stack", "dbt Product", "Data Culture"] |
"Content Type" | JSON array from ["Blog post", "Social post", "Newsletter", "Video", "Webinar", "Short-form", "Podcast", "Case Study"] |
"Content Angle" | 1-2 sentence angle |
"Score" | float (composite score) |
"Engagement Signal" | "High", "Medium", or "Low" |
"Source URL" | link to original |
"Source Platform" | one of "Reddit", "HN", "Snowflake Blog", "Databricks Blog", "BigQuery Blog", "Discourse", "News" |
"date:Brief Date:start" | YYYY-MM-DD |
"date:Brief Date:is_datetime" | 0 |
"Quarter" | JSON array with current FQ (dbt FY starts Feb 1; e.g. Mar 2026 = ["FY26 Q3"]) |
Batch all 10 in one mcp__claude_ai_Notion__notion-create-pages call.
Step 6 — Summary
Report: signals gathered, signals after filtering, DB entries created, top opportunity title and score.
Mode 2: Python Agent (Local / Cron)
For running as a standalone Python script with API-based scoring and persistent deduplication. Requires ANTHROPIC_API_KEY and NOTION_TOKEN env vars.
Setup:
uv init --name contented-but-not-satisfied --python 3.12
mkdir -p src/collectors src/scoring src/outputs tests
touch src/__init__.py src/collectors/__init__.py src/scoring/__init__.py src/outputs/__init__.py tests/__init__.py
Add to pyproject.toml dependencies:
anthropic>=0.50.0
feedparser>=6.0
httpx>=0.27
notion-client>=2.0
src/config.py
"""Configuration for the Content Intelligence Agent."""
from dataclasses import dataclass, field
PILLARS: dict[str, list[str]] = {
"Agentic Eng": [
"agentic", "ai agent", "llm pipeline", "autonomous data",
"vibe coding", "ai-assisted", "copilot data", "genie code",
],
"Analytics Eng": [
"analytics engineering", "dbt", "sql transformation", "data testing",
"data documentation", "data modeling", "semantic layer", "metrics layer",
],
"Modern Data Stack": [
"modern data stack", "data warehouse", "lakehouse", "snowflake",
"databricks", "bigquery", "orchestration", "data observability",
"data catalog", "data lineage",
],
"dbt Product": [
"dbt cloud", "dbt core", "dbt mesh", "dbt explorer",
"dbt semantic layer", "dbt build", "dbt run", "dbt test",
],
"Data Culture": [
"data governance", "data contracts", "data mesh org",
"data team", "analytics team", "data culture", "data literacy",
],
}
RSS_FEEDS: dict[str, str] = {
"Snowflake Blog": "https://www.snowflake.com/feed/",
"Databricks Blog": "https://www.databricks.com/feed",
"Discourse": "https://discourse.getdbt.com/latest.rss",
}
REDDIT_JSON_URL = "https://www.reddit.com/r/dataengineering/hot.json"
REDDIT_USER_AGENT = "ContentAgent/1.0"
HN_API_URL = "https://hn.algolia.com/api/v1/search_by_date"
HN_QUERIES = ["dbt", '"analytics engineering"', '"data transformation"']
WEB_SEARCH_QUERIES = [
'"analytics engineering" trending news',
'"dbt data" news this week',
'"agentic data engineering" OR "AI data pipelines"',
]
NOTION_BRIEFS_PAGE_ID = "32ebb38e-bda7-8120-8362-dbbba6349d53"
NOTION_PIPELINE_DATASOURCE_ID = "3e757691-8564-403f-bab4-4ad466f98683"
WEIGHT_STRATEGIC = 0.5
WEIGHT_FRESHNESS = 0.3
WEIGHT_ENGAGEMENT = 0.2
MAX_DB_ENTRIES = 10
TOP_BRIEF_COUNT = 3
@dataclass
class Signal:
title: str
url: str
summary: str
source_platform: str
published: str
engagement: int = 0
pillars: list[str] = field(default_factory=list)
score: float = 0.0
content_angle: str = ""
engagement_level: str = "Low"
content_types: list[str] = field(default_factory=list)
src/collectors/rss.py
import feedparser
from src.config import RSS_FEEDS, Signal
def collect_rss() -> list[Signal]:
signals: list[Signal] = []
for source_name, feed_url in RSS_FEEDS.items():
try:
feed = feedparser.parse(feed_url)
for entry in feed.entries[:10]:
published = getattr(entry, "published", "") or getattr(entry, "updated", "")
summary = (getattr(entry, "summary", "") or getattr(entry, "description", ""))[:300]
signals.append(Signal(
title=entry.get("title", "Untitled"), url=entry.get("link", ""),
summary=summary, source_platform=source_name, published=published,
))
except Exception as e:
print(f"[WARN] Failed to fetch {source_name}: {e}")
return signals
src/collectors/hn.py
import time
import httpx
from src.config import HN_API_URL, HN_QUERIES, Signal
def collect_hn() -> list[Signal]:
signals: list[Signal] = []
cutoff = int(time.time()) - 86400
for query in HN_QUERIES:
try:
resp = httpx.get(HN_API_URL, params={
"query": query, "tags": "story",
"numericFilters": f"created_at_i>{cutoff}", "hitsPerPage": 10,
}, timeout=15)
resp.raise_for_status()
for hit in resp.json().get("hits", []):
url = hit.get("url") or f"https://news.ycombinator.com/item?id={hit['objectID']}"
signals.append(Signal(
title=hit.get("title", "Untitled"), url=url,
summary=f"HN: {hit.get('points', 0)} pts, {hit.get('num_comments', 0)} comments",
source_platform="HN", published=hit.get("created_at", ""),
engagement=hit.get("points", 0),
))
except Exception as e:
print(f"[WARN] HN query '{query}' failed: {e}")
seen: set[str] = set()
return [s for s in signals if not (s.url in seen or seen.add(s.url))]
src/collectors/reddit.py
import httpx
from src.config import REDDIT_JSON_URL, REDDIT_USER_AGENT, Signal
def collect_reddit() -> list[Signal]:
signals: list[Signal] = []
try:
resp = httpx.get(REDDIT_JSON_URL, params={"limit": 25},
headers={"User-Agent": REDDIT_USER_AGENT}, timeout=15, follow_redirects=True)
resp.raise_for_status()
for post in resp.json().get("data", {}).get("children", []):
d = post.get("data", {})
if d.get("stickied"):
continue
signals.append(Signal(
title=d.get("title", "Untitled"),
url=f"https://www.reddit.com{d.get('permalink', '')}",
summary=(d.get("selftext", "")[:300] or d.get("title", "")),
source_platform="Reddit", published=str(d.get("created_utc", "")),
engagement=d.get("ups", 0),
))
except Exception as e:
print(f"[WARN] Reddit fetch failed: {e}")
return signals
src/scoring/pillars.py
from src.config import PILLARS, Signal
def match_pillars(signal: Signal) -> list[str]:
text = f"{signal.title} {signal.summary}".lower()
return [name for name, kws in PILLARS.items() if any(kw in text for kw in kws)]
def filter_by_pillars(signals: list[Signal]) -> list[Signal]:
filtered = []
for s in signals:
pillars = match_pillars(s)
if pillars:
s.pillars = pillars
filtered.append(s)
return filtered
src/scoring/relevance.py
from datetime import datetime, timezone
import anthropic
from src.config import WEIGHT_ENGAGEMENT, WEIGHT_FRESHNESS, WEIGHT_STRATEGIC, Signal
SCORING_PROMPT = """You are a content strategist for dbt Labs. Score each signal's strategic fit
on a scale of 1-10 based on alignment with dbt Labs' content pillars and ICP
(data teams at mid-to-large companies).
10 = directly about dbt or analytics engineering
7 = adjacent topic with clear content angle
4 = tangentially related, 1 = barely relevant
Also suggest: a 1-2 sentence content angle, and recommended content types from:
Blog post, Social post, Newsletter, Video, Webinar, Short-form, Podcast, Case Study.
Respond as JSON array: [{{"title": str, "strategic_score": int, "content_angle": str, "content_types": [str]}}]
Signals:
{signals_json}"""
def _freshness_score(published: str) -> float:
if not published:
return 4.0
try:
pub_date = datetime.fromisoformat(published.replace("Z", "+00:00"))
except ValueError:
try:
from email.utils import parsedate_to_datetime
pub_date = parsedate_to_datetime(published)
except Exception:
return 4.0
age_hours = (datetime.now(timezone.utc) - pub_date).total_seconds() / 3600
if age_hours < 24: return 10.0
elif age_hours < 48: return 7.0
elif age_hours < 72: return 4.0
else: return 1.0
def _engagement_score(engagement: int) -> float:
if engagement >= 100: return 10.0
elif engagement >= 50: return 7.0
elif engagement >= 10: return 4.0
else: return 1.0
def _engagement_level(engagement: int) -> str:
if engagement >= 50: return "High"
elif engagement >= 10: return "Medium"
return "Low"
def score_signals_with_api(signals: list[Signal]) -> list[Signal]:
import json
signals_json = json.dumps(
[{{"title": s.title, "summary": s.summary, "pillars": s.pillars}} for s in signals], indent=2)
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-sonnet-4-20250514", max_tokens=4096,
messages=[{{"role": "user", "content": SCORING_PROMPT.format(signals_json=signals_json)}}])
response_text = response.content[0].text
if "```" in response_text:
response_text = response_text.split("```")[1]
if response_text.startswith("json"): response_text = response_text[4:]
scored_by_title = {{s["title"]: s for s in json.loads(response_text)}}
for signal in signals:
api = scored_by_title.get(signal.title, {{}})
signal.score = round(
api.get("strategic_score", 5) * WEIGHT_STRATEGIC
+ _freshness_score(signal.published) * WEIGHT_FRESHNESS
+ _engagement_score(signal.engagement) * WEIGHT_ENGAGEMENT, 1)
signal.content_angle = api.get("content_angle", "")
signal.content_types = api.get("content_types", ["Blog post"])
signal.engagement_level = _engagement_level(signal.engagement)
return sorted(signals, key=lambda s: s.score, reverse=True)
def score_signals_keyword_only(signals: list[Signal]) -> list[Signal]:
for signal in signals:
strategic = min(10, len(signal.pillars) * 4 + 2)
signal.score = round(
strategic * WEIGHT_STRATEGIC
+ _freshness_score(signal.published) * WEIGHT_FRESHNESS
+ _engagement_score(signal.engagement) * WEIGHT_ENGAGEMENT, 1)
signal.engagement_level = _engagement_level(signal.engagement)
signal.content_types = ["Blog post"]
return sorted(signals, key=lambda s: s.score, reverse=True)
src/outputs/notion.py
from datetime import date
from notion_client import Client
from src.config import (MAX_DB_ENTRIES, NOTION_BRIEFS_PAGE_ID,
NOTION_PIPELINE_DATASOURCE_ID, TOP_BRIEF_COUNT, Signal)
def _current_quarter() -> str:
today = date.today()
fy_month = (today.month - 2) % 12 + 1
fy_year = today.year if today.month >= 2 else today.year - 1
quarter = (fy_month - 1) // 3 + 1
return f"FY{fy_year - 2000 + 1} Q{quarter}"
def publish_brief(notion: Client, signals: list[Signal]) -> str:
today = date.today().isoformat()
top = signals[:TOP_BRIEF_COUNT]
sections = ["## Top 3 Opportunities\n"]
for i, s in enumerate(top, 1):
sections.append(
f"### {i}. {s.title}\n"
f"**Pillar:** {', '.join(s.pillars)} | **Score:** {s.score}/10 | **Source:** {s.source_platform}\n\n"
f"{s.content_angle}\n\n**Suggested format:** {', '.join(s.content_types)}\n\n---\n")
page = notion.pages.create(
parent={{"page_id": NOTION_BRIEFS_PAGE_ID}},
icon={{"type": "emoji", "emoji": "📊"}},
properties={{"title": {{"title": [{{"text": {{"content": f"Content Brief -- {{today}}"}}}]}}}}},
children=[])
notion.blocks.children.append(block_id=page["id"], children=[
{{"object": "block", "type": "paragraph",
"paragraph": {{"rich_text": [{{"type": "text", "text": {{"content": "\n".join(sections)[:2000]}}}}]}}}}])
return page["url"]
def create_pipeline_entries(notion: Client, signals: list[Signal]) -> int:
today, quarter, count = date.today().isoformat(), _current_quarter(), 0
for signal in signals[:MAX_DB_ENTRIES]:
try:
notion.pages.create(
parent={{"database_id": NOTION_PIPELINE_DATASOURCE_ID}},
properties={{
"Title": {{"title": [{{"text": {{"content": signal.title}}}}]}},
"Stage": {{"select": {{"name": "Backlog"}}}},
"Pillar": {{"multi_select": [{{"name": p}} for p in signal.pillars]}},
"Content Type": {{"multi_select": [{{"name": ct}} for ct in signal.content_types]}},
"Content Angle": {{"rich_text": [{{"text": {{"content": signal.content_angle[:2000]}}}}]}},
"Score": {{"number": signal.score}},
"Engagement Signal": {{"select": {{"name": signal.engagement_level}}}},
"Source URL": {{"url": signal.url}},
"Source Platform": {{"select": {{"name": signal.source_platform}}}},
"Brief Date": {{"date": {{"start": today}}}},
"Quarter": {{"multi_select": [{{"name": quarter}}]}},
}})
count += 1
except Exception as e:
print(f"[WARN] Failed to create entry for '{signal.title}': {{e}}")
return count
src/agent.py
import hashlib, json, os
from pathlib import Path
from src.collectors.hn import collect_hn
from src.collectors.reddit import collect_reddit
from src.collectors.rss import collect_rss
from src.config import Signal
CACHE_DIR = Path.home() / ".content-agent"
SEEN_FILE = CACHE_DIR / "seen.json"
def _load_seen() -> set[str]:
if not SEEN_FILE.exists(): return set()
return set(json.loads(SEEN_FILE.read_text()).get("hashes", []))
def _save_seen(hashes: set[str]) -> None:
CACHE_DIR.mkdir(exist_ok=True)
SEEN_FILE.write_text(json.dumps({{"hashes": list(hashes)}}))
def _signal_hash(s: Signal) -> str:
return hashlib.sha256(f"{{s.title}}|{{s.url}}".encode()).hexdigest()[:16]
def main(use_api: bool = True) -> None:
from src.scoring.pillars import filter_by_pillars
signals = collect_rss() + collect_hn() + collect_reddit()
print(f"[1/6] Gathered {{len(signals)}} signals")
seen = _load_seen()
signals = [s for s in signals if _signal_hash(s) not in seen]
print(f"[2/6] {{len(signals)}} new signals")
signals = filter_by_pillars(signals)
print(f"[3/6] {{len(signals)}} match pillars")
if not signals: return
if use_api:
from src.scoring.relevance import score_signals_with_api
signals = score_signals_with_api(signals)
else:
from src.scoring.relevance import score_signals_keyword_only
signals = score_signals_keyword_only(signals)
notion_token = os.environ.get("NOTION_TOKEN")
if notion_token:
from notion_client import Client
from src.outputs.notion import create_pipeline_entries, publish_brief
notion = Client(auth=notion_token)
url = publish_brief(notion, signals)
count = create_pipeline_entries(notion, signals)
print(f"[4-5] Brief: {{url}}, {{count}} entries")
for s in signals: seen.add(_signal_hash(s))
_save_seen(seen)
print(f"Top: {{signals[0].title}} ({{signals[0].score}}/10)")
if __name__ == "__main__":
import sys
main(use_api="--no-api" not in sys.argv)
Run: uv sync && uv run python -m src.agent --no-api (keyword scoring) or without --no-api (Claude API scoring, needs ANTHROPIC_API_KEY).
Common Issues
| Issue | Fix |
|---|
| Snowflake RSS 404 | Use https://www.snowflake.com/feed/ (redirects to Adobe AEM) |
| Reddit RSS blocked | Use .json endpoint with User-Agent header |
| HN returns 0 results | Make 3 separate API calls; Algolia doesn't support OR with quoted phrases |
| Notion create fails | Check that data_source_id 3e757691-8564-403f-bab4-4ad466f98683 still exists |
| Quarter calculation | dbt Labs FY starts Feb 1. March 2026 = FY26 Q3 |
