Build a Breaking News Detector with LangGraph + wpnews (2025)

What you'll build

A LangGraph agent that:

• Polls /api/v1/articles/hot every 30 minutes
• Compares current hot slugs against a previous snapshot
• Triggers an action (Slack message, webhook, email) when a new story enters the top 5
• Stays cost-efficient — skips the heavy context unless a burst is detected

Total code: ~90 lines. Zero custom scraping. Zero polling infrastructure.

Why /articles/hot instead of polling /articles?

Most news APIs return articles sorted by recency. That's noisy — dozens of minor stories per hour. The /articles/hot endpoint scores each article by:

hot_score = (entity_count × 2 + topic_count) / log(1 + hours_since_published)

A 1-hour-old article about OpenAI's new model (many entities, several topics) will outrank a 10-hour-old minor update. Your agent only needs to compare slugs between calls — if a new slug appears in the top 5, something important just broke.

Step 1 — Install dependencies

pip install langgraph langchain-openai httpx python-dotenv

No wpnews SDK required — the agent calls the REST API directly via httpx. (Or install wpnews for the typed client.)

Step 2 — Define the graph state

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

class NewsState(TypedDict):
    # Slugs seen in the previous poll — persist between runs
    previous_slugs: set[str]
    # Freshly fetched hot articles
    hot_articles: list[dict]
    # Stories that just entered the top list
    new_hot: list[dict]
    # Whether any action was triggered
    acted: bool

Step 3 — Build the nodes

import httpx, os

WPNEWS_API_KEY = os.environ.get("WPNEWS_API_KEY", "")
API_BASE = "https://api.wpnews.pro"


def fetch_hot(state: NewsState) -> dict:
    """Fetch the current top 5 hot articles."""
    headers = {"X-API-Key": WPNEWS_API_KEY} if WPNEWS_API_KEY else {}
    r = httpx.get(
        f"{API_BASE}/api/v1/articles/hot",
        params={"hours": 6, "limit": 5, "lang": "en"},
        headers=headers,
        timeout=10,
    )
    r.raise_for_status()
    return {"hot_articles": r.json().get("articles", [])}


def detect_new(state: NewsState) -> dict:
    """Compare current slugs against previous poll."""
    current_slugs = {a["slug"] for a in state["hot_articles"]}
    new_slugs = current_slugs - state["previous_slugs"]
    new_hot = [a for a in state["hot_articles"] if a["slug"] in new_slugs]
    return {
        "new_hot": new_hot,
        "previous_slugs": current_slugs,   # update for next run
    }


def maybe_alert(state: NewsState) -> dict:
    """Trigger an alert if any new story entered the top list."""
    if not state["new_hot"]:
        return {"acted": False}

    for article in state["new_hot"]:
        score = article["hot_score"]
        hrs   = article["hours_ago"]
        title = article["title"]
        print(f"🔥 BREAKING ({score:.1f} hot, {hrs:.1f}h old): {title}")
        # Replace print() with your Slack/webhook/email call here

    return {"acted": True}


def should_alert(state: NewsState) -> str:
    return "alert" if state["new_hot"] else END

Step 4 — Wire the graph

from langgraph.graph import StateGraph, END

builder = StateGraph(NewsState)
builder.add_node("fetch",  fetch_hot)
builder.add_node("detect", detect_new)
builder.add_node("alert",  maybe_alert)

builder.set_entry_point("fetch")
builder.add_edge("fetch", "detect")
builder.add_conditional_edges("detect", should_alert, {"alert": "alert", END: END})
builder.add_edge("alert", END)

graph = builder.compile()

Step 5 — Run the polling loop

import time

state: NewsState = {
    "previous_slugs": set(),
    "hot_articles": [],
    "new_hot": [],
    "acted": False,
}

POLL_INTERVAL_SECONDS = 30 * 60  # 30 minutes

while True:
    result = graph.invoke(state)
    state = {**state, **result}  # carry previous_slugs forward
    print(f"Polled. {len(state['hot_articles'])} hot articles. {len(state['new_hot'])} new.")
    time.sleep(POLL_INTERVAL_SECONDS)

Run this in a background process, a cron job, or a cloud scheduler. The previous_slugs set is the only state you need to persist between restarts (save it to a file or Redis if you want durability).

Production hardening

• Gate on velocity: call /api/v1/news-velocity first; only poll hot articles when velocity_ratio > 1.5 to skip quiet periods.
• Persist state: store previous_slugs in Redis or a file so restarts don't re-alert on existing stories.
• Filter by topic: the hot endpoint accepts ?topic=artificial-intelligence to narrow the radar to your domain.
• Upgrade to Pro: free tier returns top 3; Pro returns up to 20 — more signal for smarter ranking.

Full working example (90 lines)

"""
LangGraph breaking news detector — wpnews.pro
Run: WPNEWS_API_KEY=your_key python breaking_news.py
"""
import os, time, httpx
from typing import TypedDict
from langgraph.graph import StateGraph, END

WPNEWS_API_KEY = os.environ.get("WPNEWS_API_KEY", "")
API_BASE = "https://api.wpnews.pro"

class NewsState(TypedDict):
    previous_slugs: set
    hot_articles:   list
    new_hot:        list
    acted:          bool

def fetch_hot(state):
    headers = {"X-API-Key": WPNEWS_API_KEY} if WPNEWS_API_KEY else {}
    r = httpx.get(f"{API_BASE}/api/v1/articles/hot",
                  params={"hours": 6, "limit": 5}, headers=headers, timeout=10)
    r.raise_for_status()
    return {"hot_articles": r.json().get("articles", [])}

def detect_new(state):
    current = {a["slug"] for a in state["hot_articles"]}
    new_slugs = current - state["previous_slugs"]
    return {"new_hot": [a for a in state["hot_articles"] if a["slug"] in new_slugs],
            "previous_slugs": current}

def alert(state):
    for a in state["new_hot"]:
        print(f"🔥 {a['hot_score']:.1f}pt | {a['hours_ago']:.1f}h | {a['title']}")
    return {"acted": bool(state["new_hot"])}

builder = StateGraph(NewsState)
builder.add_node("fetch",  fetch_hot)
builder.add_node("detect", detect_new)
builder.add_node("alert",  alert)
builder.set_entry_point("fetch")
builder.add_edge("fetch", "detect")
builder.add_conditional_edges("detect",
    lambda s: "alert" if s["new_hot"] else END,
    {"alert": "alert", END: END})
builder.add_edge("alert", END)
graph = builder.compile()

state = {"previous_slugs": set(), "hot_articles": [], "new_hot": [], "acted": False}
while True:
    state = {**state, **graph.invoke(state)}
    print(f"Checked. {len(state['hot_articles'])} hot, {len(state['new_hot'])} new.")
    time.sleep(1800)