Threat Intel MCP

Phoenix Intelligence Dashboard

世界情报MCP服务器

  

实时全球情报 30+域名 随着 109个MCP工具、实时运营中心仪表板、CLI和 Qdrant矢量存储 用于跨积累的智能进行企业级语义搜索。所有数据都来自免费的公共API，不需要付费订阅。

为需要世界意识的人工智能代理而构建：市场状况、地缘政治风险、军事态势、供应链中断、网络威胁等等——所有这些都可以通过模型上下文协议进行查询。矢量存储支持自然语言查询，如 *“台湾附近的军事活动”* 或 *“针对医疗保健的网络威胁”* 在所有历史数据中。

______________________________________________________________________

所得

总计：110个工具 跨越30多个情报领域。

______________________________________________________________________

快速开始

安装

git clone https://github.com/marc-shade/world-intel-mcp.git
cd world-intel-mcp
pip install -e .

# Optional extras
pip install -e ".[dashboard]"  # Live ops-center dashboard
pip install -e ".[vector]"     # Qdrant vector store + FastEmbed
pip install -e ".[dev]"        # pytest, respx, coverage

作为MCP服务器运行

world-intel-mcp  # stdio mode for Claude Code, Cursor, etc.

Claude代码配置

添加 ~/.claude.json:

{
  "mcpServers": {
    "world-intel-mcp": {
      "command": "world-intel-mcp"
    }
  }
}

仪表盘

intel-dashboard              # http://localhost:8501
intel-dashboard --port 9000  # custom port

PDF/HTML报告

pip install -e ".[pdf]"      # requires: brew install pango (macOS)
intel report                 # full PDF report → ~/.cache/world-intel-mcp/
intel report --format html   # HTML (no native deps needed)
intel report -o brief.pdf    # custom output path
intel report -s markets,cyber,earthquakes  # select sections

地图初步行动中心：带有可切换图层（地震、军事、冲突、火灾、融合、核、基础设施）的传单地图，35+实时SSE馈送，HUD条，玻璃形态面板，每个源断路器健康状况。

命令行界面

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

______________________________________________________________________

建筑

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

• 抓取器：集中式异步HTTP客户端（httpx）。按源速率限制检索过时数据回退。自动存储会在新鲜提取时产生矢量存储。
• 断路器：按来源跟踪。连续3次故障跳闸5分钟。每个RSS提要都有自己的断路器。
• 缓存：SQLite WAL模式TTL缓存。 get() 返回实时数据， get_stale() 返回过期数据以备回退。
• 向量存储：Qdrant+FastEmbed（BAAI/bge-small-en-v1.5，384暗）。用于非阻塞存储的异步后台工作队列。启用所有累积情报的语义搜索。
• 收集者：并行获取所有43个源并填充向量存储的独立守护进程。运行一次或作为守护进程运行（默认间隔：5分钟）。
• 来源 (sources/*.py)：30+个模块，每个导出 async def fetch_*(fetcher, **kwargs) -> dict.
• 分析 (analysis/*.py)：跨域合成——信号聚合、不稳定性索引、NLP、公司富集、宏观合成。
• 配置 (config/*.py)：整理数据集——22个热点、70多个基地、40个港口、24条管道、24个核设施、34条电缆、48个数据中心、27个太空港、82个交换站。

______________________________________________________________________

MCP工具参考

金融市场（7）

外汇与货币（3）

债券与收益率（2）

收益（2）

美国证券交易委员会文件（3）

公司丰富（1）

宏观综合（1）

经济（3）

中央银行（1）

比特币技术（1）

自然灾害（2）

环境（2）

冲突与安全（4）

军事与国防（6）

基础设施（4）

海事（2）

地理空间数据集（10）

新闻与媒体（3）

情报分析（8）

NLP智能（4）

战略综合（4）

气候（1）

预测市场（1）

选举（1）

位移（1）

航空（2）

网络威胁（1）

空间天气（1）

人工智能与技术（4）

健康（1）

社会与制裁（3）

航运与贸易（1）

政府（1）

国家情报（3）

扩展地理空间（1）

交通（2）

跨域警报（2）

监测（2）

矢量搜索（5）

跨域分析（3）

报告（1）

______________________________________________________________________

向量存储

可选的Qdrant向量存储随着时间的推移积累了语义检索的智能。通过Fetcher获取的所有数据都会自动嵌入和存储。

设置

# Install Qdrant (Docker)
docker run -p 6333:6333 qdrant/qdrant

# Install vector dependencies
pip install -e ".[vector]"

# Run the collector daemon (populates vector store 24/7)
intel-collector --daemon              # every 5 minutes
intel-collector --daemon --interval 120  # every 2 minutes
intel-collector --sources markets,cyber  # specific domains only
intel-collector                        # single collection cycle

语义搜索示例

一旦数据积累，AI代理就可以跨所有域进行查询：

• *“台湾海峡附近的军事活动”* --查找军事飞行、海军预警、战区态势数据
• *“针对医疗保健的网络威胁”* --查找与医疗保健相关的URLhaus、CISA KEV条目
• *“经济指标显示经济衰退”* --查找收益率曲线反转、宏观信号、FRED数据

矢量存储使用FastEmbed（基于ONNX，BAAI/bge-small-en-v1.5）进行嵌入——不需要GPU，冷启动约3秒。

______________________________________________________________________

环境变量

其他一切都使用免费的、未经身份验证的公共API。

______________________________________________________________________

发展

pip install -e ".[dev]"
pytest                       # 186 tests
pytest --cov=world_intel_mcp # with coverage
pytest tests/test_forex.py -v # single module

添加新源

1. 创建 sources/your_source.py 随着 async def fetch_your_data(fetcher: Fetcher, **kwargs) -> dict
2. 使用 fetcher.get_json(url, source="your-source", cache_key=..., cache_ttl=300) --自动缓存、重试、断路、速率限制
3. 在 server.py：添加 Tool(...) 到 TOOLS，添加 case 到 _dispatch() （使用内联导入）
4. 使用添加测试 respx 模拟HTTP（请参见 tests/test_forex.py 图案）
5. 可选择添加到 dashboard/app.py （苏格兰和南方能源公司）和 cli.py （点击）

______________________________________________________________________

许可证

麻省理工学院