diff --git a/CLAUDE.md b/CLAUDE.md index c950bee..debde51 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,7 +4,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## What This Is -World Intelligence MCP Server — 106 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. +World Intelligence MCP Server — 109 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 diff --git a/README.md b/README.md index 0db8fbc..2ede38b 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ [![Python 3.11+](https://img.shields.io/badge/Python-3.11%2B-green)](https://python.org) [![License](https://img.shields.io/badge/License-MIT-yellow)](LICENSE) -Real-time global intelligence across **30+ domains** with **106 MCP tools**, a live ops-center dashboard, a CLI, and a **Qdrant vector store** for enterprise-grade semantic search across accumulated intelligence. All data comes from free, public APIs — no paid subscriptions required. +Real-time global intelligence across **30+ domains** with **109 MCP tools**, a live ops-center dashboard, a CLI, and a **Qdrant vector store** for enterprise-grade semantic search across accumulated intelligence. All data comes from free, public APIs — no paid subscriptions required. Built for AI agents that need world awareness: market conditions, geopolitical risk, military posture, supply chain disruptions, cyber threats, and more — all queryable via the Model Context Protocol. The vector store enables natural language queries like *"military activity near Taiwan"* or *"cyber threats targeting healthcare"* across all historical data. @@ -54,9 +54,10 @@ Built for AI agents that need world awareness: market conditions, geopolitical r | **Cross-Domain** | 2 | Alert digest, weekly trends | | **Vector Search** | 5 | Qdrant semantic search, similarity, timeline, collection | +| **Cross-Domain Analytics** | 3 | Correlation, domain summary, trend detection | | **Data Collection** | 1 | On-demand collector trigger | -**Total: 106 tools** across 30+ intelligence domains. +**Total: 109 tools** across 30+ intelligence domains. --- @@ -333,6 +334,13 @@ collector.py (daemon) ──┘ | `intel_vector_stats` | Vector store collection statistics | | `intel_collect` | Trigger an on-demand collection cycle | +### Cross-Domain Analytics (3) +| Tool | Description | +|------|-------------| +| `intel_cross_correlate` | Find correlated signals across all domains for a given topic | +| `intel_domain_summary` | Per-category summary of stored intelligence (counts, sources, recency) | +| `intel_trend_detection` | Detect activity surges/drops by comparing recent vs baseline periods | + --- ## Vector Store diff --git a/src/world_intel_mcp/server.py b/src/world_intel_mcp/server.py index d008e17..86a4722 100644 --- a/src/world_intel_mcp/server.py +++ b/src/world_intel_mcp/server.py @@ -28,6 +28,8 @@ Phase 15: Business intelligence — forex (3), bonds/yields (2), earnings (2), S Phase 16: Vector intelligence — semantic search, similar events, timeline, vector stats, on-demand collection (+5 = 104 tools). Qdrant vector store auto-populates from all fetches. Collector daemon for 24/7 data accumulation. Enterprise-grade semantic retrieval. +Phase 17: Cross-domain analytics — cross-domain correlation, domain summary, trend detection + (+3 = 109 tools). Historical analysis and early warning from accumulated vector data. """ import asyncio @@ -1636,6 +1638,67 @@ TOOLS: list[Tool] = [ }, }, ), + Tool( + name="intel_cross_correlate", + description="Cross-domain intelligence correlation. Given a topic, finds related signals across ALL intelligence domains (military, financial, cyber, conflict, etc.) and groups them by category. Shows how events ripple across domains — e.g., how a military buildup correlates with market movements and news coverage.", + inputSchema={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Topic to correlate across domains (e.g., 'Taiwan strait tensions', 'oil supply disruption')", + }, + "hours": { + "type": "number", + "description": "Time window in hours (default 24)", + "default": 24, + }, + "limit_per_domain": { + "type": "integer", + "description": "Max signals per domain category (default 5)", + "default": 5, + }, + }, + "required": ["query"], + }, + ), + Tool( + name="intel_domain_summary", + description="Summary of all intelligence data stored in the vector database. Shows per-category data point counts, unique sources, latest/earliest timestamps, and total events tracked. Answers: what intelligence do we have and how recent is it?", + inputSchema={ + "type": "object", + "properties": { + "hours": { + "type": "number", + "description": "Time window in hours (default 24)", + "default": 24, + }, + }, + }, + ), + Tool( + name="intel_trend_detection", + description="Detect activity trends by comparing recent intelligence activity against a baseline. Identifies SURGE (>50% increase), ELEVATED (>20%), DECLINING (<-20%), and DROP (<-50%) patterns. Useful for early warning when a domain suddenly spikes or goes quiet.", + inputSchema={ + "type": "object", + "properties": { + "category": { + "type": "string", + "description": "Focus on one category (e.g., 'Conflict & Security'). Default: all categories.", + }, + "recent_hours": { + "type": "number", + "description": "Recent window to measure (default 6 hours)", + "default": 6, + }, + "baseline_hours": { + "type": "number", + "description": "Baseline window for comparison (default 48 hours)", + "default": 48, + }, + }, + }, + ), # --- System (1 tool) --- Tool( name="intel_status", @@ -2273,6 +2336,29 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any: ), ) + case "intel_cross_correlate": + if _vector_store is None: + return {"error": "Vector store not available (Qdrant not running?)"} + return await _vector_store.cross_domain_correlate( + query=arguments["query"], + hours=arguments.get("hours", 24.0), + limit_per_domain=arguments.get("limit_per_domain", 5), + ) + case "intel_domain_summary": + if _vector_store is None: + return {"error": "Vector store not available (Qdrant not running?)"} + return await _vector_store.domain_summary( + hours=arguments.get("hours", 24.0), + ) + case "intel_trend_detection": + if _vector_store is None: + return {"error": "Vector store not available (Qdrant not running?)"} + return await _vector_store.trend_detection( + category=arguments.get("category"), + recent_hours=arguments.get("recent_hours", 6.0), + baseline_hours=arguments.get("baseline_hours", 48.0), + ) + # System case "intel_status": vs_stats = {} diff --git a/src/world_intel_mcp/vector_store.py b/src/world_intel_mcp/vector_store.py index f61d95d..1083c4c 100644 --- a/src/world_intel_mcp/vector_store.py +++ b/src/world_intel_mcp/vector_store.py @@ -652,3 +652,306 @@ class VectorStore: "embedding_model": EMBEDDING_MODEL, "embedding_dim": EMBEDDING_DIM, } + + # ------------------------------------------------------------------ + # Cross-domain correlation + # ------------------------------------------------------------------ + + async def cross_domain_correlate( + self, + query: str, + hours: float = 24.0, + limit_per_domain: int = 5, + ) -> dict: + """Find correlated signals across multiple intelligence domains. + + Given a topic, searches all domains and groups results by category, + showing how different intelligence streams relate to the same event. + """ + return await asyncio.to_thread( + self._correlate_sync, query, hours, limit_per_domain + ) + + def _correlate_sync( + self, + query: str, + hours: float, + limit_per_domain: int, + ) -> dict: + from qdrant_client.models import FieldCondition, Filter, MatchValue, Range + + client = _get_qdrant() + vector = _embed_text(query) + + conditions = [ + FieldCondition(key="has_error", match=MatchValue(value=False)), + ] + if hours: + cutoff = time.time() - (hours * 3600) + conditions.append(FieldCondition(key="timestamp", range=Range(gte=cutoff))) + + # Fetch more results to ensure coverage across domains + results = client.search( + collection_name=COLLECTION_NAME, + query_vector=vector, + query_filter=Filter(must=conditions), + limit=100, + with_payload=True, + ) + + # Group by category, keep top N per category + by_category: dict[str, list[dict]] = {} + for r in results: + cat = r.payload.get("category", "Other") + if cat not in by_category: + by_category[cat] = [] + if len(by_category[cat]) < limit_per_domain: + by_category[cat].append( + { + "score": round(r.score, 4), + "domain": r.payload.get("domain", ""), + "text": r.payload.get("text", "")[:300], + "datetime": r.payload.get("datetime", ""), + "country": r.payload.get("country"), + } + ) + + # Sort categories by best score + sorted_cats = sorted( + by_category.items(), + key=lambda kv: max(e["score"] for e in kv[1]) if kv[1] else 0, + reverse=True, + ) + + return { + "query": query, + "hours": hours, + "domains_found": len(by_category), + "correlations": [ + { + "category": cat, + "signal_count": len(entries), + "best_score": max(e["score"] for e in entries), + "signals": entries, + } + for cat, entries in sorted_cats + ], + "total_signals": sum(len(v) for v in by_category.values()), + } + + # ------------------------------------------------------------------ + # Domain summary + # ------------------------------------------------------------------ + + async def domain_summary(self, hours: float = 24.0) -> dict: + """Get per-domain summary of stored intelligence.""" + return await asyncio.to_thread(self._domain_summary_sync, hours) + + def _domain_summary_sync(self, hours: float) -> dict: + from qdrant_client.models import FieldCondition, Filter, MatchValue, Range + + client = _get_qdrant() + cutoff = time.time() - (hours * 3600) + + # Scroll all points in the time window (no vector search) + conditions = [ + FieldCondition(key="timestamp", range=Range(gte=cutoff)), + FieldCondition(key="has_error", match=MatchValue(value=False)), + ] + + all_points = [] + offset = None + while True: + points, next_offset = client.scroll( + collection_name=COLLECTION_NAME, + scroll_filter=Filter(must=conditions), + limit=200, + offset=offset, + with_payload=["domain", "category", "timestamp", "event_count"], + ) + all_points.extend(points) + if next_offset is None or len(points) == 0: + break + offset = next_offset + + # Aggregate by category + by_category: dict[str, dict] = {} + for p in all_points: + cat = p.payload.get("category", "Other") + if cat not in by_category: + by_category[cat] = { + "count": 0, + "domains": set(), + "latest": 0.0, + "earliest": float("inf"), + "total_events": 0, + } + entry = by_category[cat] + entry["count"] += 1 + entry["domains"].add(p.payload.get("domain", "")) + ts = p.payload.get("timestamp", 0) + entry["latest"] = max(entry["latest"], ts) + entry["earliest"] = min(entry["earliest"], ts) + ec = p.payload.get("event_count") + if ec and isinstance(ec, (int, float)): + entry["total_events"] += int(ec) + + # Convert sets to lists and format + summaries = [] + for cat, info in sorted( + by_category.items(), key=lambda kv: kv[1]["count"], reverse=True + ): + summaries.append( + { + "category": cat, + "data_points": info["count"], + "unique_sources": len(info["domains"]), + "sources": sorted(info["domains"]), + "total_events_tracked": info["total_events"] or None, + "latest": datetime.fromtimestamp( + info["latest"], tz=timezone.utc + ).isoformat() + if info["latest"] > 0 + else None, + "earliest": datetime.fromtimestamp( + info["earliest"], tz=timezone.utc + ).isoformat() + if info["earliest"] < float("inf") + else None, + } + ) + + return { + "hours": hours, + "total_data_points": len(all_points), + "categories": len(summaries), + "summary": summaries, + } + + # ------------------------------------------------------------------ + # Trend detection (activity anomalies) + # ------------------------------------------------------------------ + + async def trend_detection( + self, + category: str | None = None, + recent_hours: float = 6.0, + baseline_hours: float = 48.0, + ) -> dict: + """Detect activity trends by comparing recent vs baseline periods. + + Compares data point density in the recent window against the baseline + to identify surges or drops in intelligence activity. + """ + return await asyncio.to_thread( + self._trend_sync, category, recent_hours, baseline_hours + ) + + def _trend_sync( + self, + category: str | None, + recent_hours: float, + baseline_hours: float, + ) -> dict: + from qdrant_client.models import FieldCondition, Filter, MatchValue, Range + + client = _get_qdrant() + now = time.time() + recent_cutoff = now - (recent_hours * 3600) + baseline_cutoff = now - (baseline_hours * 3600) + + base_conditions = [ + FieldCondition(key="has_error", match=MatchValue(value=False)), + ] + if category: + base_conditions.append( + FieldCondition(key="category", match=MatchValue(value=category)) + ) + + def _count_in_window(start: float, end: float) -> dict[str, int]: + """Count points per category in a time window.""" + conditions = base_conditions + [ + FieldCondition(key="timestamp", range=Range(gte=start, lte=end)), + ] + points = [] + offset = None + while True: + batch, next_offset = client.scroll( + collection_name=COLLECTION_NAME, + scroll_filter=Filter(must=conditions), + limit=200, + offset=offset, + with_payload=["category"], + ) + points.extend(batch) + if next_offset is None or len(batch) == 0: + break + offset = next_offset + counts: dict[str, int] = {} + for p in points: + cat = p.payload.get("category", "Other") + counts[cat] = counts.get(cat, 0) + 1 + return counts + + recent_counts = _count_in_window(recent_cutoff, now) + baseline_counts = _count_in_window(baseline_cutoff, recent_cutoff) + + # Normalize baseline to per-hour rate + baseline_window = baseline_hours - recent_hours + if baseline_window <= 0: + baseline_window = 1.0 + + all_cats = set(recent_counts.keys()) | set(baseline_counts.keys()) + trends = [] + for cat in sorted(all_cats): + recent = recent_counts.get(cat, 0) + baseline = baseline_counts.get(cat, 0) + + recent_rate = recent / recent_hours if recent_hours > 0 else 0 + baseline_rate = baseline / baseline_window if baseline_window > 0 else 0 + + if baseline_rate > 0: + change_pct = ((recent_rate - baseline_rate) / baseline_rate) * 100 + elif recent_rate > 0: + change_pct = 100.0 # New activity + else: + change_pct = 0.0 + + # Classify + if change_pct > 50: + trend = "SURGE" + elif change_pct > 20: + trend = "ELEVATED" + elif change_pct < -50: + trend = "DROP" + elif change_pct < -20: + trend = "DECLINING" + else: + trend = "NORMAL" + + trends.append( + { + "category": cat, + "recent_count": recent, + "baseline_count": baseline, + "recent_rate_per_hr": round(recent_rate, 2), + "baseline_rate_per_hr": round(baseline_rate, 2), + "change_pct": round(change_pct, 1), + "trend": trend, + } + ) + + # Sort by absolute change + trends.sort(key=lambda t: abs(t["change_pct"]), reverse=True) + + surges = [t for t in trends if t["trend"] == "SURGE"] + drops = [t for t in trends if t["trend"] == "DROP"] + + return { + "recent_window_hours": recent_hours, + "baseline_window_hours": baseline_hours, + "categories_analyzed": len(trends), + "surges": len(surges), + "drops": len(drops), + "trends": trends, + }