Add multi-provider search/fetch failover and block empty newsletters from being sent

.env.example

@@ -5,8 +5,14 @@ OPENAI_API_KEY=your-openrouter-or-openai-key-here
 # Default model for every agent (override per role at the bottom of this file).
 OPENAI_MODEL=google/gemini-3-flash-preview
 
-# Web search via serper.dev. Required: the agents research every section through it.
+# Web search and page fetch run through a deterministic round-robin with failover across whichever
+# providers have a key set. Serper is the baseline (search + fetch); Exa and Tavily add search +
+# fetch, Firecrawl and Diffbot add fetch only. Set at least one search provider so research works.
 SERPER_API_KEY=your-serper-api-key-here
+EXA_API_KEY=
+TAVILY_API_KEY=
+FIRECRAWL_API_KEY=
+DIFFBOT_API_KEY=
 
 # Email delivery via Resend. Only needed when actually sending (python src/app.py run --send).
 RESEND_API_KEY=

CLAUDE.md

@@ -4,7 +4,7 @@ Guidance for working in this repository.
 
 ## What this is
 
-`agentic-mediapulse` is an agentic newsletter generator. Given a subject (a stock ticker, company name, or industry theme), a newsroom of focused agents researches, writes, and edits a locale-aware briefing across five editorial sections, with every claim traced to a real source. It is built on the Microsoft Agent Framework (`agent-framework`) plus Serper for web search.
+`agentic-mediapulse` is an agentic newsletter generator. Given a subject (a stock ticker, company name, or industry theme), a newsroom of focused agents researches, writes, and edits a locale-aware briefing across five editorial sections, with every claim traced to a real source. It is built on the Microsoft Agent Framework (`agent-framework`) plus a round-robin web toolbelt (Serper, Exa, Tavily, Firecrawl, Diffbot) for search and page fetch.
 
 ## Layout
 
@@ -13,7 +13,7 @@ All application code lives under `src/`. Packages keep their top-level names (`a
 - `src/api.py` — FastAPI service. `POST /run` (full campaign) and `POST /test` (one user). Both run in the background, return `202`, and default to dry-run. Auth via the `X-API-Key` header matched against `SECRET_KEY`.
 - `src/app.py` — local CLI mirroring the API (`run`, `test`), for testing without HTTP.
 - `src/agents/orchestrator.py` — the pipeline: analyst → 5 parallel beat desks (researcher → writer → editor) → managing-editor gap roundtable → masthead → reviewer → deterministic clean/assemble/dedupe. Most non-agent logic (citation gating, URL/article validation, dedupe, subject-name canonicalization, prose humanizing) lives here.
-- `src/agents/` — one module per agent (`analyst`, `researcher`, `writer`, `editor`, `managing_editor`, `reviewer`), plus `beats.py` (beat desks), `campaign.py` (top-level run over subscriptions), `sections.py` (the five editorial beats), `providers/` (subject-memory and ticker-profile context providers), and `tools/` (Serper search, web fetch).
+- `src/agents/` — one module per agent (`analyst`, `researcher`, `writer`, `editor`, `managing_editor`, `reviewer`), plus `beats.py` (beat desks), `campaign.py` (top-level run over subscriptions, which skips and marks `failed` any newsletter that comes out with zero sections rather than sending it), `sections.py` (the five editorial beats), `providers/` (subject-memory and ticker-profile context providers), and `tools/` (the `web_search` and `web_fetch` tools over a round-robin + failover provider package in `tools/providers/`: Serper, Exa, Tavily, Firecrawl, Diffbot, selected deterministically and hidden from the LLM).
 - `src/agents/runtime/` — agent plumbing shared by every agent: `chat_client.py` (per-role chat client plus the `SKILLS` provider), `make_agent.py` (the factory that wires generic activity tracking into every agent), `guardrails.py` (guardrail/citation middleware), and `tracking.py` (the generic `ActivityTracker`/`ToolTracker` middleware, `newsletter_scope`, and run context vars). New agents are built via `make_agent(...)` so tracking is automatic.
 - `src/agents/skills/` — `SKILL.md` files that control agent behavior (`subject-profile`, `section-research`, `newsletter-format`). Prefer editing these over code when changing how agents research or write.
 - `src/db/` — all database access. `mediapulse.py` reads subscriptions and ticker profiles from the upstream MediaPulse Postgres (`MEDIAPULSE_DATABASE_URL`, read-only, raw psycopg). The app's own Postgres (`DATABASE_URL`, SQLModel) uses `engine.py` for the shared engine, with each table's model alongside its operations: `newsletters.py` (archives each newsletter as markdown plus JSONB metadata, with a `pending`/`complete`/`failed` lifecycle via `create_newsletter`/`finalize_newsletter`), `memory.py` (subject-brief agent memory), and `agent_activity.py` (one row per agent run and tool call, tied to its `newsletter_id`, recording status, duration, model, and token usage). The schema is owned by Alembic migrations, not `create_all` (see Migrations).
@@ -101,7 +101,8 @@ The suite covers the deterministic logic, not the LLM agents: orchestrator text/
 
 ## External services
 
-- `SERPER_API_KEY` — web search.
+- `SERPER_API_KEY` — web search and page fetch (the baseline provider).
+- `EXA_API_KEY`, `TAVILY_API_KEY`, `FIRECRAWL_API_KEY`, `DIFFBOT_API_KEY` — optional extra providers; when their key is set they join a deterministic round-robin with failover (Exa and Tavily also search, Firecrawl and Diffbot fetch only). `web_search`/`web_fetch` keep an identical signature, so the LLM never sees which provider served a call.
 - `MEDIAPULSE_DATABASE_URL` — read-only Postgres for subscriptions and ticker data; schema is defined in the upstream [MediaPulse](https://github.com/hyperjumptech/mediapulse) repo.
 - `DATABASE_URL` — the app's own read-write Postgres for archived newsletters and agent memory. Tables are auto-created on first use, separate from `MEDIAPULSE_DATABASE_URL`.
 - `SECRET_KEY` — API auth.

src/agents/analyst.py

@@ -3,7 +3,7 @@
 from agents.runtime.chat_client import SKILLS, chat_client
 from agents.runtime.guardrails import SubjectGuardrail
 from agents.runtime.make_agent import make_agent
-from agents.tools import search, web_fetch
+from agents.tools import web_fetch, web_search
 
 analyst = make_agent(
     name="analyst",
@@ -13,7 +13,7 @@
         "You are a research analyst. The subject may be a ticker, a company, or an industry/theme. "
         "Use the subject-profile skill to turn the subject into a brief."
     ),
-    tools=[search, web_fetch],
+    tools=[web_search, web_fetch],
     context_providers=[SKILLS, TickerProfileProvider(), SubjectMemoryProvider()],
     middleware=[SubjectGuardrail()],
 )

src/agents/campaign.py

@@ -6,7 +6,7 @@
 from db.mediapulse import fetch_subscriptions
 from db.newsletters import create_newsletter, finalize_newsletter
 from emails.mailer import send_email
-from emails.templates.newsletter import newsletter_sources, render_newsletter_email
+from emails.templates.newsletter import has_sections, newsletter_sources, render_newsletter_email
 
 CONCURRENCY = 3
 SEND_INTERVAL = 1.0  # seconds between emails
@@ -32,6 +32,15 @@ async def run_campaign(*, subscriptions: list[dict] | None = None, send: bool =
 
     async def deliver(ticker: str, recipients: list[dict], markdown: str, newsletter_id: int | None) -> None:
         nonlocal last_send
+
+        if not has_sections(markdown):
+            finalize_newsletter(
+                newsletter_id, content=markdown, metadata={"ticker": ticker, "sources": []}, status="failed"
+            )
+            log(f"skipped {ticker}: newsletter has no sections, not sending")
+
+            return
+
         email = render_newsletter_email(markdown, ticker=ticker)
         finalize_newsletter(
             newsletter_id, content=markdown, metadata={"ticker": ticker, "sources": newsletter_sources(markdown)}

src/agents/researcher.py

@@ -4,7 +4,7 @@
 from agents.runtime.guardrails import RecordSources, SourceRegistry
 from agents.runtime.make_agent import make_agent
 from agents.sections import Section
-from agents.tools import search
+from agents.tools import web_search
 
 
 def make_researcher(section: Section, registry: SourceRegistry) -> Agent:
@@ -20,7 +20,7 @@ def make_researcher(section: Section, registry: SourceRegistry) -> Agent:
             "matters, written in English. The writer needs at least 2 and up to 5 strong, distinct stories. "
             "You may query in the subject's local language. Never invent a URL. Output only the list, no preamble."
         ),
-        tools=[search],
+        tools=[web_search],
         context_providers=[SKILLS],
         middleware=[RecordSources(registry)],
     )

src/agents/skills/section-research/SKILL.md

@@ -15,7 +15,7 @@ Research one newsletter beat, find the strongest recent articles, and output the
 ## Search instructions
 
 1. Build simple queries: the subject's short common name on its own first, then add at most one topic word to narrow if needed. Never use long multi-keyword queries, `site:` operators, or mixed-language queries — they return nothing.
-2. Call `search` with kind="news" first; use kind="web" only for background. Always pass the `gl` and `hl` codes from the brief's Locale line. Searches default to the past week; widen to recency="month" if results are thin, narrow to recency="day" for breaking news.
+2. Call `web_search` with kind="news" first; use kind="web" only for background. Always pass the `gl` and `hl` codes from the brief's Locale line. Searches default to the past week; widen to recency="month" if results are thin, narrow to recency="day" for breaking news.
 3. Keep the subject the lead: most items should be about it. You may cover named competitors and industry context secondarily, but only from the subject's home market.
 4. Select the strongest, most relevant recent items — at least 2, at most 5. Prefer substantive developments: products, strategy, expansion, deals, regulation, technology, operations, leadership. Skip stock-index roundups (IHSG levels), "top gainers/losers", analyst price targets, and technical analysis unless a price move is itself the news. Never invent an article or URL.
 

src/agents/skills/subject-profile/SKILL.md

@@ -13,7 +13,7 @@ The subject string the user provided. If verified exchange listing details are a
 
 ## Instructions
 
-1. If the subject is not obvious and no verified listing details are provided, call `search` (kind="web" or kind="news", recency="") to confirm what it is.
+1. If the subject is not obvious and no verified listing details are provided, call `web_search` (kind="web" or kind="news", recency="") to confirm what it is.
 2. Output a compact brief using exactly the labelled lines in the Output section below — nothing else.
 
 ## Output

src/agents/tools/__init__.py

@@ -1,4 +1,4 @@
-from agents.tools.search import search
 from agents.tools.web_fetch import web_fetch
+from agents.tools.web_search import web_search
 
-__all__ = ["search", "web_fetch"]
+__all__ = ["web_fetch", "web_search"]

src/agents/tools/providers/__init__.py

@@ -0,0 +1,14 @@
+from agents.tools.providers import diffbot, exa, firecrawl, serper, tavily
+from agents.tools.providers.dispatch import AllProvidersFailed, Provider, dispatch, reset_cursor
+
+__all__ = [
+    "AllProvidersFailed",
+    "Provider",
+    "diffbot",
+    "dispatch",
+    "exa",
+    "firecrawl",
+    "reset_cursor",
+    "serper",
+    "tavily",
+]

src/agents/tools/providers/diffbot.py

@@ -0,0 +1,16 @@
+import os
+
+import httpx
+
+
+def fetch(url: str) -> str:
+    """Adapter: Diffbot article extraction for `url`, raising on HTTP error."""
+    response = httpx.get(
+        "https://api.diffbot.com/v3/article",
+        params={"token": os.environ["DIFFBOT_API_KEY"], "url": url},
+        timeout=30.0,
+    )
+    response.raise_for_status()
+    objects = response.json().get("objects", [])
+
+    return objects[0].get("text", "") if objects else ""

src/agents/tools/providers/dispatch.py

@@ -0,0 +1,72 @@
+import os
+from collections.abc import Callable
+
+
+class AllProvidersFailed(Exception):
+    """Raised when every configured provider for a capability errored on one call."""
+
+    def __init__(self, capability: str, failures: list[tuple[str, str]]) -> None:
+        self.capability = capability
+        self.failures = failures
+        detail = "; ".join(f"{name}={reason}" for name, reason in failures) or "no providers configured"
+
+        super().__init__(f"all {capability} providers failed: {detail}")
+
+
+class Provider:
+    """A named web provider, active only when its API-key env var is set."""
+
+    def __init__(self, name: str, env: str, fn: Callable) -> None:
+        self.name = name
+        self.env = env
+        self.fn = fn
+
+    def available(self) -> bool:
+        return bool(os.getenv(self.env))
+
+
+# Round-robin cursor per capability ("search"/"fetch"). Lock-free: a benign race only skews load.
+_cursor: dict[str, int] = {}
+
+
+def reset_cursor() -> None:
+    """Clear the round-robin cursor so tests start from a known position."""
+    _cursor.clear()
+
+
+def _rotate(capability: str, active: list[Provider]):
+    start = _cursor.get(capability, 0)
+    _cursor[capability] = start + 1
+
+    for offset in range(len(active)):
+        yield active[(start + offset) % len(active)]
+
+
+def dispatch(capability: str, providers: list[Provider], call: Callable, accept: Callable) -> object:
+    """Round-robin across the available providers, failing over on error or unusable result.
+
+    Returns the first result accepted by `accept`. If providers respond but none is usable (for
+    example all empty), returns the last such result. If every configured provider errors, raises
+    AllProvidersFailed naming each one.
+    """
+    active = [provider for provider in providers if provider.available()]
+    errors: list[tuple[str, str]] = []
+    empty_result, saw_empty = None, False
+
+    for provider in _rotate(capability, active):
+        try:
+            result = call(provider.fn)
+        except Exception as error:
+            errors.append((provider.name, f"{type(error).__name__}: {error}"))
+
+            continue
+
+        if accept(result):
+            return result
+
+        empty_result, saw_empty = result, True
+
+    if saw_empty:
+        return empty_result
+
+    raise AllProvidersFailed(capability, errors)

src/agents/tools/providers/exa.py

@@ -0,0 +1,61 @@
+import os
+from datetime import datetime, timedelta, timezone
+
+import httpx
+
+_RECENCY_DAYS = {"day": 1, "week": 7, "month": 30}
+
+
+def _start_date(recency: str) -> str | None:
+    days = _RECENCY_DAYS.get(recency)
+
+    if not days:
+        return None
+
+    return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
+
+
+def search(query: str, kind: str, gl: str, hl: str, recency: str) -> list[dict]:
+    """Adapter: Exa neural/keyword search, mapped to the Serper result shape."""
+    body = {"query": query, "numResults": 10, "type": "auto", "contents": {"text": {"maxCharacters": 500}}}
+
+    if kind == "news":
+        body["category"] = "news"
+
+    start_date = _start_date(recency)
+
+    if start_date:
+        body["startPublishedDate"] = start_date
+
+    response = httpx.post(
+        "https://api.exa.ai/search",
+        headers={"x-api-key": os.environ["EXA_API_KEY"]},
+        json=body,
+        timeout=30.0,
+    )
+    response.raise_for_status()
+    results = response.json().get("results", [])
+
+    return [
+        {
+            "title": item.get("title", ""),
+            "link": item.get("url", ""),
+            "snippet": " ".join((item.get("text") or "").split())[:300],
+            "date": item.get("publishedDate", "") or "",
+        }
+        for item in results
+    ]
+
+
+def fetch(url: str) -> str:
+    """Adapter: Exa page contents for `url`, raising on HTTP error."""
+    response = httpx.post(
+        "https://api.exa.ai/contents",
+        headers={"x-api-key": os.environ["EXA_API_KEY"]},
+        json={"urls": [url], "text": True},
+        timeout=30.0,
+    )
+    response.raise_for_status()
+    results = response.json().get("results", [])
+
+    return results[0].get("text", "") if results else ""

src/agents/tools/providers/firecrawl.py

@@ -0,0 +1,17 @@
+import os
+
+import httpx
+
+
+def fetch(url: str) -> str:
+    """Adapter: Firecrawl scrape for `url` as markdown, raising on HTTP error."""
+    response = httpx.post(
+        "https://api.firecrawl.dev/v1/scrape",
+        headers={"Authorization": f"Bearer {os.environ['FIRECRAWL_API_KEY']}"},
+        json={"url": url, "formats": ["markdown"]},
+        timeout=30.0,
+    )
+    response.raise_for_status()
+    data = response.json().get("data", {})
+
+    return data.get("markdown", "") or ""

src/agents/tools/providers/serper.py

@@ -0,0 +1,49 @@
+import os
+
+import httpx
+
+_RECENCY = {"day": "qdr:d", "week": "qdr:w", "month": "qdr:m"}
+
+
+def serper(endpoint: str, query: str, gl: str, hl: str, tbs: str = "") -> list[dict]:
+    body = {"q": query}
+
+    if gl:
+        body["gl"] = gl
+
+    if hl:
+        body["hl"] = hl
+
+    if tbs:
+        body["tbs"] = tbs
+
+    response = httpx.post(
+        f"https://google.serper.dev/{endpoint}",
+        headers={"X-API-KEY": os.environ["SERPER_API_KEY"]},
+        json=body,
+        timeout=30.0,
+    )
+    response.raise_for_status()
+
+    return response.json().get("news" if endpoint == "news" else "organic", [])
+
+
+def search(query: str, kind: str, gl: str, hl: str, recency: str) -> list[dict]:
+    """Adapter: map the unified search call onto Serper's news/web endpoints."""
+    endpoint = "news" if kind == "news" else "search"
+
+    return serper(endpoint, query, gl, hl, _RECENCY.get(recency, ""))
+
+
+def fetch(url: str) -> str:
+    """Adapter: scrape readable text for `url` via Serper, raising on HTTP error."""
+    response = httpx.post(
+        "https://scrape.serper.dev",
+        headers={"X-API-KEY": os.environ["SERPER_API_KEY"]},
+        json={"url": url},
+        timeout=30.0,
+    )
+    response.raise_for_status()
+    data = response.json()
+
+    return data.get("text") or data.get("markdown") or ""