gavindiaz/world-intel-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: IODA fallback, UCDP timeout fix, PDF export, 18 tests

Browse Source 
- infrastructure.py: Add IODA (Georgia Tech) as fallback when Cloudflare
  Radar returns 403 (no API token). Internet outage data now available
  without CLOUDFLARE_API_TOKEN.
- conflict.py: Increase UCDP GED API timeout from 15s to 30s to handle
  slow responses from ucdpapi.pcr.uu.se.
- dashboard/app.py: Add /api/report/pdf endpoint for PDF daily brief
  export via weasyprint (optional dependency).
- pyproject.toml: Add dashboard + pdf optional deps, intel-dashboard
  script entry point, update description for 55 tools.
- tests: Add 11 new tests covering health, sanctions, elections, shipping,
  social, nuclear, infrastructure (IODA fallback + cable health), and
  UCDP conflict. Fix cross-loop asyncio lock issue in conftest.
- Total: 50 tests passing, 55 tools verified.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Marc Shade
2026-02-24 07:05:58 -05:00
parent 6fee3feae5
commit 2d33ea136a
6 changed files with 548 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+3 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "world-intel-mcp"
version = "0.1.0"
description = "World Intelligence MCP Server - real-time global intelligence across 17 domains"
description = "World Intelligence MCP Server - real-time global intelligence across 23 domains with 55 MCP tools"
readme = "README.md"
requires-python = ">=3.11"
license = {text = "MIT"}
@@ -21,6 +21,7 @@ dependencies = [
]
[project.optional-dependencies]
dashboard = ["starlette>=0.37.0", "uvicorn>=0.29.0"]
pdf = ["weasyprint>=62.0"]
dev = [
"pytest>=8.0.0",
@@ -32,6 +33,7 @@ dev = [
[project.scripts]
world-intel-mcp = "world_intel_mcp.server:run"
intel = "world_intel_mcp.cli:main"
intel-dashboard = "world_intel_mcp.dashboard.app:run"
[build-system]
requires = ["hatchling"]
src/world_intel_mcp/dashboard/app.py
+51 -1
View File
@@ -11,7 +11,7 @@ from datetime import datetime, timezone
from pathlib import Path
from starlette.applications import Starlette
from starlette.responses import HTMLResponse, JSONResponse, StreamingResponse
from starlette.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
from starlette.routing import Route
from world_intel_mcp.cache import Cache
@@ -205,6 +205,55 @@ async def api_health(request):
return JSONResponse({"status": "ok"})
async def api_report_pdf(request):
"""Generate a PDF daily brief report.
Renders the daily_brief.html template with live data, then converts
to PDF via weasyprint. Requires ``pip install world-intel-mcp[pdf]``.
"""
try:
from weasyprint import HTML as WeasyHTML
except ImportError:
return JSONResponse(
{"error": "weasyprint not installed — run: pip install world-intel-mcp[pdf]"},
status_code=501,
)
from world_intel_mcp.reports.html_report import render_template
data = await _fetch_overview()
context = {
"title": "Daily Intelligence Brief",
"generated_at": data.get("timestamp", ""),
"market_quotes": data.get("market_quotes", {}),
"crypto_quotes": data.get("crypto_quotes", {}),
"macro_signals": data.get("macro_signals", {}),
"earthquakes": data.get("earthquakes", {}),
"cyber_threats": data.get("cyber_threats", {}),
"news_feed": data.get("news_feed", {}),
"military_flights": data.get("military_flights", {}),
"internet_outages": data.get("internet_outages", {}),
"climate_anomalies": data.get("climate_anomalies", {}),
"displacement": data.get("displacement", {}),
"risk_scores": data.get("risk_scores", {}),
"alert_digest": data.get("alert_digest", {}),
}
html_str = render_template("daily_brief.html", context)
pdf_bytes = WeasyHTML(string=html_str).write_pdf()
now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
return Response(
content=pdf_bytes,
media_type="application/pdf",
headers={
"Content-Disposition": f'attachment; filename="intel-brief-{now_str}.pdf"',
"Access-Control-Allow-Origin": "*",
},
)
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
@@ -215,6 +264,7 @@ app = Starlette(
Route("/api/overview", api_overview),
Route("/api/stream", api_stream),
Route("/api/health", api_health),
Route("/api/report/pdf", api_report_pdf),
],
)
src/world_intel_mcp/sources/conflict.py
+3 -1
View File
@@ -155,13 +155,14 @@ async def fetch_ucdp_events(
now = datetime.now(timezone.utc)
cutoff = now - timedelta(days=days)
# Fetch page 1 to discover total pages
# Fetch page 1 to discover total pages (UCDP is slow — 30s timeout)
page1_data = await fetcher.get_json(
_UCDP_GED_URL,
source="ucdp",
cache_key=f"conflict:ucdp:{days}:page1",
cache_ttl=21600,
params={"pagesize": min(limit, 1000), "page": 0},
timeout=30.0,
)
if page1_data is None:
@@ -187,6 +188,7 @@ async def fetch_ucdp_events(
cache_key=f"conflict:ucdp:{days}:page{p}",
cache_ttl=21600,
params={"pagesize": min(limit, 1000), "page": p},
timeout=30.0,
)
for p in range(1, total_pages)
]
src/world_intel_mcp/sources/infrastructure.py
+108 -24
View File
@@ -148,6 +148,85 @@ def _point_in_corridor(
return lon_lo <= lon <= lon_hi
# ---------------------------------------------------------------------------
# IODA fallback (Georgia Tech Internet Intelligence — public, no auth)
# ---------------------------------------------------------------------------
_IODA_OUTAGES_URL = "https://api.ioda.inetintel.cc.gatech.edu/v2/outages/overall"
async def _fetch_ioda_outages(fetcher: Fetcher) -> dict | None:
"""Fallback: fetch recent internet outages from IODA public API.
Returns a dict matching the fetch_internet_outages output shape,
or None if IODA is also unavailable.
"""
from datetime import timedelta
now = datetime.now(timezone.utc)
since = now - timedelta(days=7)
data = await fetcher.get_json(
url=_IODA_OUTAGES_URL,
source="ioda",
cache_key="infra:outages:ioda",
cache_ttl=300,
params={
"from": int(since.timestamp()),
"until": int(now.timestamp()),
"limit": 20,
},
timeout=15.0,
)
if data is None:
return None
# IODA returns {"data": [{"entity": {...}, "events": [...]}]}
raw_items = data if isinstance(data, list) else data.get("data", [])
if not isinstance(raw_items, list):
return None
outages: list[dict] = []
ongoing_count = 0
for item in raw_items:
if not isinstance(item, dict):
continue
entity = item.get("entity", {}) if isinstance(item.get("entity"), dict) else {}
events = item.get("events", []) if isinstance(item.get("events"), list) else [item]
for ev in events:
if not isinstance(ev, dict):
continue
start = ev.get("from") or ev.get("start")
end = ev.get("until") or ev.get("end")
is_ongoing = end is None
if is_ongoing:
ongoing_count += 1
outages.append({
"id": ev.get("id") or entity.get("code"),
"start": start,
"end": end,
"description": ev.get("summary", entity.get("name", "")),
"scope": ev.get("level", "unknown"),
"countries": [entity.get("code", "")] if entity.get("code") else [],
"asns": [],
"is_ongoing": is_ongoing,
})
return {
"outages": outages,
"ongoing_count": ongoing_count,
"total_7d": len(outages),
"source": "ioda-gatech",
"timestamp": _utc_now_iso(),
}
# ---------------------------------------------------------------------------
# Internet Outages (Cloudflare Radar)
# ---------------------------------------------------------------------------
@@ -155,44 +234,49 @@ def _point_in_corridor(
async def fetch_internet_outages(fetcher: Fetcher) -> dict:
"""Fetch recent internet outage annotations from Cloudflare Radar.
Uses the public Radar annotations API for the last 7 days. If the
``CLOUDFLARE_API_TOKEN`` environment variable is set, authenticated
requests are made which may yield higher rate limits.
Tries authenticated Cloudflare Radar API first (requires
``CLOUDFLARE_API_TOKEN``). Falls back to IODA (Georgia Tech
Internet Intelligence) public API for internet outage signals.
Returns:
Dict with outages list, ongoing/total counts, source, and timestamp.
"""
token = os.environ.get("CLOUDFLARE_API_TOKEN")
headers: dict[str, str] | None = None
if token:
headers = {"Authorization": f"Bearer {token}"}
params = {
"limit": 20,
"dateRange": "7d",
}
data = await fetcher.get_json(
url=_CF_RADAR_OUTAGES_URL,
source="cloudflare-radar",
cache_key="infra:outages",
cache_ttl=300,
headers=headers,
params=params,
)
now_iso = _utc_now_iso()
data = None
# --- Attempt 1: Cloudflare Radar (requires token) ---
if token:
headers = {"Authorization": f"Bearer {token}"}
data = await fetcher.get_json(
url=_CF_RADAR_OUTAGES_URL,
source="cloudflare-radar",
cache_key="infra:outages:cf",
cache_ttl=300,
headers=headers,
params={"limit": 20, "dateRange": "7d"},
)
# --- Attempt 2: IODA public API (no auth needed) ---
if data is None:
logger.warning("Cloudflare Radar outages API returned no data")
return {
data = await _fetch_ioda_outages(fetcher)
if data is not None:
return data # IODA already returns our output shape
if data is None:
note = "Set CLOUDFLARE_API_TOKEN for detailed outage data" if not token else None
logger.warning("Internet outages: no data from Cloudflare or IODA")
result: dict = {
"outages": [],
"ongoing_count": 0,
"total_7d": 0,
"source": "cloudflare-radar",
"timestamp": now_iso,
}
if note:
result["note"] = note
return result
outages: list[dict] = []
src/world_intel_mcp/tests/conftest.py
+12
View File
@@ -1,5 +1,6 @@
"""Test configuration — strips proxy env vars so httpx doesn't try SOCKS."""
import asyncio
import os
import pytest
@@ -16,3 +17,14 @@ def _strip_proxy_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Remove system proxy env vars so httpx creates clean connections."""
for var in _PROXY_VARS:
monkeypatch.delenv(var, raising=False)
@pytest.fixture(autouse=True)
def _reset_fetcher_locks() -> None:
"""Reset global asyncio locks between tests to avoid cross-loop binding."""
import world_intel_mcp.fetcher as fetcher_mod
fetcher_mod._yahoo_lock = asyncio.Lock()
fetcher_mod._yahoo_last_call = 0.0
fetcher_mod._source_locks.clear()
fetcher_mod._source_last_call.clear()
src/world_intel_mcp/tests/test_sources.py
+371
View File
@@ -216,3 +216,374 @@ async def test_fetch_world_bank_indicators(fetcher: Fetcher) -> None:
assert len(result["indicators"]) == 1
assert result["indicators"][0]["id"] == "NY.GDP.MKTP.CD"
assert result["source"] == "world-bank"
# ---------------------------------------------------------------------------
# Health (disease outbreaks)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_disease_outbreaks(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.health import fetch_disease_outbreaks
rss_xml = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0"><channel><title>WHO</title>
<item>
<title>Ebola outbreak in DRC - Update 5</title>
<link>https://who.int/ebola-update</link>
<pubDate>Mon, 10 Feb 2026 12:00:00 GMT</pubDate>
<description>Ebola virus disease outbreak continues in North Kivu.</description>
</item>
<item>
<title>Seasonal influenza update</title>
<link>https://who.int/flu</link>
<pubDate>Sun, 09 Feb 2026 08:00:00 GMT</pubDate>
<description>Northern hemisphere flu season report.</description>
</item>
</channel></rss>"""
# Mock all 3 health feeds returning same XML
respx.get(url__regex=r".*who\.int.*").mock(
return_value=httpx.Response(200, text=rss_xml)
)
respx.get(url__regex=r".*cdc\.gov.*").mock(
return_value=httpx.Response(200, text=rss_xml)
)
respx.get(url__regex=r".*outbreaknewstoday.*").mock(
return_value=httpx.Response(200, text=rss_xml)
)
result = await fetch_disease_outbreaks(fetcher)
assert result["source"] == "health-outbreak-monitor"
assert result["count"] > 0
assert result["high_concern_count"] > 0 # "ebola" in title
assert any(item["is_high_concern"] for item in result["items"])
# ---------------------------------------------------------------------------
# Sanctions (OFAC SDN)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_sanctions_search(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.sanctions import fetch_sanctions_search
csv_data = (
'100,"DOE, John",individual,SDGT,"","","","","","","","nationality Iran; DOB 01 Jan 1970"\n'
'101,"ACME CORP",entity,CUBA,"","","","","","","",""\n'
'102,"SMITH, Jane",individual,SYRIA,"","","","","","","","nationality Syria"\n'
)
respx.get(url__regex=r".*treasury\.gov.*sdn\.csv.*").mock(
return_value=httpx.Response(200, text=csv_data)
)
result = await fetch_sanctions_search(fetcher, query="DOE")
assert result["source"] == "ofac-sdn"
assert result["count"] == 1
assert result["matches"][0]["name"] == "DOE, John"
assert result["total_entities"] >= 1
@respx.mock
@pytest.mark.asyncio
async def test_fetch_sanctions_search_country_filter(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.sanctions import fetch_sanctions_search
csv_data = (
'100,"DOE, John",individual,SDGT,"","","","","","","","nationality Iran; DOB 01 Jan 1970"\n'
'101,"ACME CORP",entity,CUBA,"","","","","","","",""\n'
)
respx.get(url__regex=r".*treasury\.gov.*sdn\.csv.*").mock(
return_value=httpx.Response(200, text=csv_data)
)
result = await fetch_sanctions_search(fetcher, country="iran")
assert result["count"] == 1
assert result["matches"][0]["name"] == "DOE, John"
# ---------------------------------------------------------------------------
# Elections (pure data — no HTTP mocking needed)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_fetch_election_calendar(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.elections import fetch_election_calendar
result = await fetch_election_calendar(fetcher)
assert result["source"] == "election-calendar"
assert result["count"] > 0
assert "elections" in result
# Each election should have risk_score
for election in result["elections"]:
assert "risk_score" in election
assert "days_until" in election
assert election["status"] in ("past", "upcoming")
@pytest.mark.asyncio
async def test_fetch_election_calendar_country_filter(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.elections import fetch_election_calendar
result = await fetch_election_calendar(fetcher, country="USA")
# May or may not match depending on config data
assert result["source"] == "election-calendar"
assert isinstance(result["elections"], list)
# ---------------------------------------------------------------------------
# Shipping (Yahoo Finance quotes)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_shipping_index(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.shipping import fetch_shipping_index
chart_response = {
"chart": {
"result": [{
"meta": {
"symbol": "BDRY",
"regularMarketPrice": 15.50,
"regularMarketChangePercent": 4.2,
"currency": "USD",
}
}]
}
}
# Mock all 4 shipping symbols
respx.get(url__regex=r".*finance\.yahoo\.com.*").mock(
return_value=httpx.Response(200, json=chart_response)
)
result = await fetch_shipping_index(fetcher)
assert result["source"] == "yahoo-finance"
assert len(result["quotes"]) > 0
assert isinstance(result["stress_score"], (int, float))
assert result["assessment"] in ("low", "moderate", "elevated", "high", "extreme")
# ---------------------------------------------------------------------------
# Social (Reddit)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_social_signals(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.social import fetch_social_signals
reddit_response = {
"data": {
"children": [
{
"data": {
"title": "Ukraine conflict escalation analysis",
"score": 5000,
"num_comments": 300,
"upvote_ratio": 0.95,
"created_utc": 1708700000,
"permalink": "/r/worldnews/comments/abc123/",
"is_self": False,
}
},
{
"data": {
"title": "US-China trade tensions rise",
"score": 2000,
"num_comments": 150,
"upvote_ratio": 0.88,
"created_utc": 1708690000,
"permalink": "/r/worldnews/comments/def456/",
"is_self": True,
}
},
]
}
}
respx.get(url__regex=r".*reddit\.com.*hot\.json.*").mock(
return_value=httpx.Response(200, json=reddit_response)
)
result = await fetch_social_signals(fetcher)
assert result["source"] == "reddit-public"
assert result["velocity_metrics"]["total_posts"] > 0
assert result["velocity_metrics"]["high_engagement_count"] > 0
assert result["subreddits_queried"] == ["worldnews", "geopolitics"]
# ---------------------------------------------------------------------------
# Nuclear (USGS near test sites)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_nuclear_monitor(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.nuclear import fetch_nuclear_monitor
geojson = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"id": "nn00900001",
"properties": {
"mag": 2.8,
"place": "50km N of Test Site",
"time": 1708700000000,
"tsunami": 0,
},
"geometry": {
"type": "Point",
"coordinates": [129.08, 41.30, 3.0], # Near Punggye-ri
},
}
],
}
# Mock USGS for all nuclear sites
respx.get("https://earthquake.usgs.gov/fdsnws/event/1/query").mock(
return_value=httpx.Response(200, json=geojson)
)
result = await fetch_nuclear_monitor(fetcher, hours=72)
assert result["source"] == "usgs-nuclear-monitor"
assert len(result["sites"]) == 5 # 5 nuclear test sites
assert isinstance(result["total_flagged_events"], int)
assert isinstance(result["critical_flags"], int)
# ---------------------------------------------------------------------------
# Infrastructure (Cloudflare + IODA fallback)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_internet_outages_ioda_fallback(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.infrastructure import fetch_internet_outages
# IODA response shape
ioda_response = {
"data": [
{
"entity": {"code": "US", "name": "United States"},
"events": [
{
"id": "out-123",
"from": "2026-02-20T00:00:00Z",
"until": None,
"summary": "BGP outage detected",
"level": "country",
}
],
}
]
}
# Cloudflare returns 403 (no token)
respx.get(url__regex=r".*cloudflare\.com.*").mock(
return_value=httpx.Response(403, json={"error": "unauthorized"})
)
# IODA responds
respx.get(url__regex=r".*ioda\.inetintel.*").mock(
return_value=httpx.Response(200, json=ioda_response)
)
import os
os.environ.pop("CLOUDFLARE_API_TOKEN", None)
result = await fetch_internet_outages(fetcher)
assert result["source"] == "ioda-gatech"
assert result["total_7d"] == 1
assert result["ongoing_count"] == 1
@respx.mock
@pytest.mark.asyncio
async def test_fetch_cable_health(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.infrastructure import fetch_cable_health
warnings = [
{
"msgYear": 2026,
"msgNumber": 42,
"navArea": "XII",
"subregion": "31",
"status": "in force",
"issueDate": "2026-02-20",
"text": "SUBMARINE CABLE OPERATIONS 40-30.5N/030-15.2E VESSELS ADVISED",
}
]
respx.get(url__regex=r".*nga\.mil.*broadcast-warn.*").mock(
return_value=httpx.Response(200, json=warnings)
)
result = await fetch_cable_health(fetcher)
assert result["source"] == "nga-msi"
assert "corridors" in result
assert len(result["corridors"]) == 6
assert result["cable_related_warnings"] >= 1 # "cable" keyword in text
# ---------------------------------------------------------------------------
# Conflict (UCDP + ACLED)
# ---------------------------------------------------------------------------
@respx.mock
@pytest.mark.asyncio
async def test_fetch_ucdp_events(fetcher: Fetcher) -> None:
from world_intel_mcp.sources.conflict import fetch_ucdp_events
from datetime import datetime, timezone
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
ucdp_response = {
"TotalPages": 1,
"Result": [
{
"id": 12345,
"relid": "11-1",
"year": 2026,
"date_start": today,
"date_end": today,
"country": "Ukraine",
"region": "Europe",
"type_of_violence": 1,
"side_a": "Government of Ukraine",
"side_b": "DPR",
"best": 5,
"high": 10,
"low": 2,
"latitude": 48.0,
"longitude": 37.5,
"source_article": "Reuters",
"source_headline": "Fighting continues",
}
],
}
respx.get(url__regex=r".*ucdpapi\.pcr\.uu\.se.*").mock(
return_value=httpx.Response(200, json=ucdp_response)
)
result = await fetch_ucdp_events(fetcher, days=30)
assert result["source"] == "ucdp"
assert result["count"] == 1
assert result["events"][0]["country"] == "Ukraine"
assert result["total_fatalities_best"] == 5