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 — 68 tools across 27 domains providing real-time global intelligence from free public APIs. Serves three interfaces: MCP stdio (for Claude Code/Cursor), a live Starlette dashboard with SSE, and a Click CLI with Rich output.

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 report daily               # generate HTML report
intel status                     # cache + circuit breaker health

# Dashboard (requires [dashboard] extra)
intel-dashboard --port 8501

Architecture

Three consumers share the same source modules and infrastructure stack:

server.py (MCP stdio)  ─┐
cli.py (Click CLI)      ├─> sources/*.py ─> Fetcher ─> CircuitBreaker ─> Cache (SQLite)
dashboard/app.py (SSE)  ─┘                                                  │
                                                                   ~/.cache/world-intel-mcp/cache.db

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. 25 modules covering markets, seismology, military, cyber, health, 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, 105 major cities, 28 world leaders, 36 APT groups.

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.
• Dashboard SSE: dashboard/app.py fetches all domains in parallel via asyncio.gather(), streams updates every 30 seconds.
• Tests strip proxy env vars automatically via conftest.py fixture (prevents SOCKS proxy interference).

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.