world-intel-mcp/CLAUDE.md

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

What This Is

World Intelligence MCP Server — 113 tools across 30+ domains providing real-time global intelligence from free public APIs. Serves four interfaces: MCP stdio (for Claude Code/Cursor), a live Starlette dashboard with SSE, a Click CLI with Rich output, and a collector daemon for 24/7 vector store population. Python 3.11+, built with hatchling.

Commands

# Install
pip install -e ".[dev,dashboard]"

# Run MCP server (stdio mode)
world-intel-mcp

# Run tests (pytest-asyncio, auto mode)
pytest
pytest --cov=world_intel_mcp
pytest src/world_intel_mcp/tests/test_sources.py::test_fetch_market_quotes -v  # single test

# CLI
intel markets                    # stock indices
intel earthquakes --min-mag 5.0  # USGS quakes
intel status                     # cache + circuit breaker health

# Dashboard (requires [dashboard] extra)
intel-dashboard --port 8501
# Or with .env auto-loaded:
./run-dashboard.sh

Architecture

Three consumers share the same source modules and infrastructure stack:

server.py (MCP stdio)  ─┐                                              ┌─ VectorStore (Qdrant)
cli.py (Click CLI)      ├─> sources/*.py ─> Fetcher ─> CircuitBreaker ─┤
dashboard/app.py (SSE)  ─┘    analysis/*.py                            └─ Cache (SQLite)
collector.py (daemon)  ──┘

Infrastructure layer (fetcher.py, cache.py, circuit_breaker.py):

• Fetcher: Centralized async HTTP client (httpx). All external calls go through get_json(), get_text(), or get_xml(). Handles retries (2 max), per-source rate limiting, and stale-data fallback (never returns blank if old data exists).
• CircuitBreaker: Per-source tracking. 3 consecutive failures trips the breaker for 5 minutes. Each RSS feed gets its own breaker (rss:bbc_world).
• Cache: SQLite WAL-mode TTL cache. get() returns live data, get_stale() returns expired data for fallback.

Source modules (sources/*.py): Each module exports async def fetch_*(fetcher: Fetcher, **kwargs) -> dict. Pure data fetching — no MCP awareness. 30 modules covering markets, seismology, military, cyber, health, tech, environmental, etc.

Analysis modules (analysis/*.py): Cross-domain intelligence that consumes outputs from multiple sources. Includes signal aggregation, instability indexing, NLP (entity extraction, classification, clustering, spike detection via Welford's algorithm), and strategic synthesis.

Static config (config/*.py): Curated datasets — 22 intel hotspots, 70+ military bases, 40 ports, 24 pipelines, 24 nuclear facilities, 34 undersea cables, 48 AI datacenters, 27 spaceports, 27 mineral deposits, 82 stock exchanges, 105 major cities, 28 world leaders, 36 APT groups.

Vector store (vector_store.py): Optional Qdrant + FastEmbed (BAAI/bge-small-en-v1.5, 384-dim ONNX). Async background worker queue (asyncio.Queue + asyncio.to_thread()) for non-blocking storage. Fetcher auto-stores on fresh HTTP fetches. 65+ source name → category mappings for filtered search.

Collector (collector.py): Standalone daemon for 24/7 vector store population. 43 sources organized by domain, with --daemon, --interval, --sources CLI args. Dynamic import via _import_fetch_fn(). Entry point: intel-collector.

Reports (reports.py): PDF/HTML intelligence report generator. Collects 18 domains in parallel, renders styled HTML, converts to PDF via WeasyPrint. Optional dependency: pip install -e ".[pdf]".

Dashboard (dashboard/): Self-contained Starlette app with a single index.html template (no frontend build step). SSE endpoint streams all domains in parallel via asyncio.gather(), refreshes every 30 seconds. Loads .env from project root on startup.

Adding a New Tool

1. Create sources/your_source.py with async def fetch_your_data(fetcher: Fetcher, **kwargs) -> dict
2. Use fetcher.get_json(url, source="your-source", cache_key=..., cache_ttl=300) — this gives you caching, retries, circuit breaking, and rate limiting automatically
3. In server.py: import the module, add a Tool(...) to the TOOLS list, add a case to _dispatch()
4. Optionally add to dashboard/app.py (SSE endpoint) and cli.py (Click command)
5. Add tests using respx to mock HTTP (see tests/test_sources.py for pattern)

Key Patterns

• Source name string: The source parameter in fetcher.get_json() identifies the API for circuit breaking and rate limiting. Must match entries in _SOURCE_RATE_LIMITS if rate-limited (e.g., "yahoo-finance", "coingecko", "adsblol").
• Tool dispatch: server.py uses Python match/case to route tool names to source functions. Tool names follow intel_* convention.
• All source functions take fetcher as first arg — never construct your own httpx client.
• Tests strip proxy env vars automatically via conftest.py fixture (prevents SOCKS proxy interference). The conftest also resets global fetcher rate-limit locks between tests to avoid cross-event-loop binding.

Environment Variables

Only these unlock additional data sources (everything else works unauthenticated):

• ACLED_ACCESS_TOKEN — conflict events
• NASA_FIRMS_API_KEY — satellite wildfire data
• EIA_API_KEY — energy prices
• CLOUDFLARE_API_TOKEN — internet outage data
• FRED_API_KEY — macro economic data
• OPENSKY_CLIENT_ID / OPENSKY_CLIENT_SECRET — military flight fallback

Testing

Tests use respx to mock httpx responses. Fixtures in conftest.py provide cache (tmp_path SQLite) and fetcher (with clean breaker). Tests are async (pytest-asyncio in auto mode). Proxy env vars are stripped automatically.