feat: add cross-domain analytics — correlation, domain summary, trend detection (109 tools)
- intel_cross_correlate: finds related signals across all domains for a topic, groups by category, shows how events ripple across intelligence streams - intel_domain_summary: per-category aggregate of stored intelligence with data point counts, unique sources, latest/earliest timestamps - intel_trend_detection: compares recent vs baseline activity rates per category, identifies SURGE/ELEVATED/DECLINING/DROP patterns for early warning - All three leverage accumulated Qdrant vector store data - 109 tools total across 30+ domains
This commit is contained in:
Marc Shade
@@ -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
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
[](https://python.org)
|
||||
[](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
|
||||
|
||||
@@ -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 = {}
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user