A Company Monitoring Agent That Doesn't Cry Wolf

The hard part of a news monitoring agent isn't finding coverage. It's deciding which of forty articles is worth interrupting someone for.

Profile picture of Serply
Serply
News coverage of a company scored and filtered down to a few alerts

The first version of every monitoring agent sends too many alerts. It finds the news correctly, summarises it competently, and gets muted within two weeks because forty percent of what it sends is a syndicated rewrite of something you already saw.

Finding coverage is the easy part. The value is in what you throw away.

Pulling coverage

import os
import requests
from urllib.parse import quote_plus
from concurrent.futures import ThreadPoolExecutor

API_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"


def news(query: str) -> list[dict]:
    resp = requests.get(
        f"{BASE}/news/q={quote_plus(query)}",
        headers={"X-Api-Key": API_KEY},
        timeout=30,
    )
    resp.raise_for_status()
    entries = resp.json().get("feed", {}).get("entries", [])
    return [
        {"title": e.get("title", ""), "url": e.get("link", ""),
         "summary": e.get("summary", ""), "published": e.get("published"),
         "source": e.get("source")}
        for e in entries
    ]


def coverage(company: str) -> list[dict]:
    queries = [
        f'"{company}"',
        f"{company} earnings results",
        f"{company} announcement",
        f"{company} acquisition OR funding OR layoffs",
    ]
    with ThreadPoolExecutor(max_workers=4) as pool:
        batches = list(pool.map(news, queries))
    return [a for batch in batches for a in batch]

Articles come back under feed.entries — not results like the other search endpoints. The published and source fields are what the rest of this pipeline runs on.

Several targeted queries beat one broad one. A bare company name query drowns in passing mentions; the event-specific queries surface the things you’d actually want to know.

Deduplicating syndication

A single wire story appears under a dozen bylines with lightly edited headlines. Exact-match deduplication catches none of them:

import re
from difflib import SequenceMatcher
from urllib.parse import urlparse

STOP = {"the", "a", "an", "to", "of", "in", "for", "on", "as", "at", "by",
        "says", "said", "report", "reports", "reportedly"}


def title_key(title: str) -> str:
    words = re.findall(r"[a-z0-9]+", title.lower())
    return " ".join(w for w in words if w not in STOP and len(w) > 2)


def domain(url: str) -> str:
    host = urlparse(url or "").netloc.lower()
    return host[4:] if host.startswith("www.") else host


def dedupe(articles: list[dict], threshold: float = 0.72) -> list[dict]:
    clusters: list[dict] = []

    for a in articles:
        key = title_key(a["title"])
        if not key:
            continue

        placed = False
        for c in clusters:
            if SequenceMatcher(None, key, c["key"]).ratio() >= threshold:
                c["articles"].append(a)
                c["domains"].add(domain(a["url"]))
                placed = True
                break

        if not placed:
            clusters.append({
                "key": key,
                "articles": [a],
                "domains": {domain(a["url"])},
            })

    out = []
    for c in clusters:
        rep = min(c["articles"], key=lambda a: a.get("published") or "9999")
        out.append({
            **rep,
            "outlet_count": len(c["domains"]),
            "outlets": sorted(c["domains"]),
            "duplicates": len(c["articles"]) - 1,
        })
    return out

Two things fall out of clustering rather than just filtering.

outlet_count is a materiality signal you get for free. A story picked up by eleven outlets is bigger than one covered by a single trade blog, and that’s a better proxy for importance than any keyword list.

Choosing the earliest article as the cluster representative points at the original reporting rather than the fourth rewrite, which is what someone following the link actually wants.

Only alert on new clusters

State is what keeps a daily job from re-alerting on the same story every morning:

import json
import hashlib
from pathlib import Path

STATE = Path("monitor_state.json")


def load_state() -> dict:
    return json.loads(STATE.read_text()) if STATE.exists() else {"seen": {}}


def cluster_id(article: dict) -> str:
    return hashlib.sha1(title_key(article["title"]).encode()).hexdigest()[:16]


def new_only(clusters: list[dict], state: dict) -> list[dict]:
    seen = state.setdefault("seen", {})
    fresh = []

    for c in clusters:
        cid = cluster_id(c)
        prev = seen.get(cid)

        if prev is None:
            fresh.append({**c, "escalation": None})
        elif c["outlet_count"] >= prev["outlet_count"] * 3 and c["outlet_count"] >= 6:
            # Story materially escalated since we last saw it
            fresh.append({**c, "escalation":
                          f"{prev['outlet_count']} → {c['outlet_count']} outlets"})

        seen[cid] = {"outlet_count": c["outlet_count"]}

    return fresh

The escalation rule is worth the extra branch. A story that went from two outlets to twenty overnight is a different event than it was yesterday, and suppressing it purely because you’ve seen the headline before is how monitoring systems miss the thing they exist to catch.

Scoring materiality

import json
from anthropic import Anthropic

client = Anthropic()

SCORE = """You triage company news for a busy executive.

Score each story 0-10 for how much it warrants their attention today:

9-10  Existential or immediate: bankruptcy, criminal charges, CEO departure,
      major recall, acquisition of or by the company
6-8   Material: earnings surprise, significant funding, large layoffs, major
      product launch, regulatory action, notable executive change
3-5   Worth knowing: partnerships, minor products, awards, routine hires,
      analyst rating changes
0-2   Noise: passing mentions, sponsored content, listicles, stock-movement
      recaps with no underlying news, republished press releases

Be strict. Most stories are 0-4. Reserve 6+ for things that would change a
decision. Do not inflate scores for stories that sound dramatic but describe
routine business.

For each, return: {"index": int, "score": int, "category": str,
                   "one_line": str, "why_material": str|null}

Return JSON: {"scored": [...]}"""


def score(clusters: list[dict], company: str) -> list[dict]:
    payload = [
        {"index": i, "title": c["title"], "summary": c["summary"],
         "source": c["source"], "published": c["published"],
         "outlets": c["outlet_count"]}
        for i, c in enumerate(clusters)
    ]
    msg = client.messages.create(
        model="claude-sonnet-4-5", max_tokens=3000, system=SCORE,
        messages=[{"role": "user", "content":
                   f"COMPANY: {company}\n\n{json.dumps(payload, indent=2)}"}],
    )
    scored = json.loads(msg.content[0].text)["scored"]

    out = []
    for s in scored:
        c = clusters[s["index"]]
        out.append({**c, **s})
    return sorted(out, key=lambda x: -x["score"])

“Be strict. Most stories are 0-4” is the line that makes this usable. Without an explicit expectation about the distribution, models cluster everything at 6-7 — every story sounds important when you read it in isolation, and the scale collapses.

Verifying the big ones before you send

A 9 or 10 should not go out on one article’s say-so:

def corroborate(cluster: dict, company: str) -> dict:
    if cluster["score"] < 8:
        return {**cluster, "corroborated": None}

    if cluster["outlet_count"] >= 3:
        return {**cluster, "corroborated": True,
                "basis": f"{cluster['outlet_count']} independent outlets"}

    resp = requests.get(
        f"{BASE}/search/q={quote_plus(company + ' ' + cluster['title'][:60])}&num=10",
        headers={"X-Api-Key": API_KEY, "X-Proxy-Location": "US"},
        timeout=30,
    )
    resp.raise_for_status()
    domains = {domain(r.get("link", "")) for r in resp.json().get("results", [])}

    return {**cluster,
            "corroborated": len(domains - {domain(cluster["url"])}) >= 2,
            "basis": f"{len(domains)} domains in web search"}

Falling back to a web search when the news endpoint only shows one outlet catches the case that matters most: a dramatic headline from a single low-quality source. Sending that as a high-priority alert once costs you more credibility than a week of good alerts earns.

The digest

def run(company: str, threshold: int = 6) -> dict:
    state = load_state()
    clusters = dedupe(coverage(company))
    fresh = new_only(clusters, state)

    if not fresh:
        STATE.write_text(json.dumps(state))
        return {"company": company, "alerts": [], "checked": len(clusters)}

    scored = score(fresh, company)
    alerts = [corroborate(c, company) for c in scored if c["score"] >= threshold]

    STATE.write_text(json.dumps(state))
    return {
        "company": company,
        "checked": len(clusters),
        "new": len(fresh),
        "alerts": [
            {"score": a["score"], "category": a["category"],
             "headline": a["one_line"], "why": a.get("why_material"),
             "url": a["url"], "outlets": a["outlet_count"],
             "corroborated": a.get("corroborated"),
             "escalation": a.get("escalation")}
            for a in alerts
        ],
    }

Reporting checked and new alongside the alerts is a small thing that buys a lot of trust. A digest that says “checked 43 stories, 2 worth your attention” reads as a filter doing its job. The same two alerts with no denominator read as everything the system found — and the first time someone discovers a story it missed, they stop believing it.