"""MCP 服务 (M8) — FastMCP 国际财经 Deep Research 工具。 暴露 5 个 MCP Tool 给 Claude Code / Cherry Studio 调用。 """ import json import logging from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any from mcp.server.fastmcp import FastMCP from crawler.utils import get_news_day from embedding.client import ( EmbeddingConfig, embed_batch, load_embedding_config, make_embedding_client, ) from vectorstore.client import VectorStore, make_qdrant_client from vectorstore.models import SearchFilter logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # # 单例 Backend # --------------------------------------------------------------------------- # @dataclass class _Backend: emb_config: EmbeddingConfig emb_client: Any vector_store: VectorStore _backend: _Backend | None = None def _get_backend() -> _Backend: """延迟初始化 MCP 后端（embedder + vector_store）。""" global _backend if _backend is None: config = load_embedding_config() client = make_embedding_client(config) qdrant = make_qdrant_client() store = VectorStore(qdrant) _backend = _Backend(emb_config=config, emb_client=client, vector_store=store) logger.info("MCP 后端就绪: embed=%s, qdrant=%d 条", config.model, store.count()) return _backend # --------------------------------------------------------------------------- # # 嵌入 + 检索 # --------------------------------------------------------------------------- # def _search( query: str, top_k: int = 10, search_filter: SearchFilter | None = None, score_threshold: float = 0.3, ) -> list[dict[str, Any]]: """嵌入查询 → Qdrant 检索 → 返回 dict 列表。""" be = _get_backend() vectors = embed_batch(be.emb_client, be.emb_config, [query]) if not vectors: return [] results = be.vector_store.query( query_vector=vectors[0], top_k=top_k, search_filter=search_filter, score_threshold=score_threshold, ) return [ { "title": r.title, "title_zh": r.title_zh, "url": r.url, "source": r.source_id, "score": round(r.score, 4), "publish_time": r.publish_time, "events": r.events, "content_zh_preview": r.content_zh_preview, } for r in results ] def _load_today_events(date_str: str | None = None) -> list[dict]: """加载当日高重要度事件。""" day = date_str or get_news_day() ev_dir = Path(f"data/events/{day}") if not ev_dir.is_dir(): return [] events: list[dict] = [] for fp in sorted(ev_dir.glob("*.json")): if fp.name == "index.json": continue try: data = json.loads(fp.read_text(encoding="utf-8")) for ev in data.get("events", []): if ev.get("importance", 0) >= 4: events.append({ **ev, "title": data.get("title", ""), "title_zh": data.get("title_zh", ""), "url": data.get("url", ""), "source_id": data.get("source_id", ""), "publish_time": data.get("publish_time", ""), }) except (json.JSONDecodeError, OSError): pass return sorted(events, key=lambda e: -e.get("importance", 0)) def _fmt_results(hits: list[dict[str, Any]], query: str) -> str: """检索结果 → Markdown 格式化。""" if not hits: return f"未找到与「{query}」相关的结果。" lines = [f"## 🔍 检索结果: {query}", "", f"共 {len(hits)} 条:", ""] for i, h in enumerate(hits, 1): title = h.get("title_zh") or h.get("title", "无标题") sentiment_map = {"positive": "🟢利好", "neutral": "⚪中性", "negative": "🔴利空"} lines.append(f"### {i}. {title}") lines.append(f"- 来源: {h['source']} | 相似度: {h['score']}") lines.append(f"- 🔗 {h.get('url', '')}") for ev in h.get("events", []): s = sentiment_map.get(ev.get("sentiment", ""), "") codes = ",".join(ev.get("stock_codes", [])[:5]) code_str = f" [{codes}]" if codes else "" lines.append( f" - {s} [{ev.get('event_type', '')}] " f"重要度{ev.get('importance', '')} {ev.get('summary_zh', '')}{code_str}" ) lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # # MCP 服务器 & 工具 # --------------------------------------------------------------------------- # mcp = FastMCP( name="国际财经DeepResearch", instructions="国际财经新闻 Deep Research 知识库。语义检索英文财经新闻的翻译与投资事件分析。", ) # ── Tool 1: search_news ── @mcp.tool() def search_news(query: str, top_k: int = 10) -> str: """语义检索国际财经新闻知识库。 参数: query: 自然语言查询（中文或英文，如 "美联储利率决议" "Apple earnings"） top_k: 返回条数（默认 10） 返回: Markdown 格式检索结果，含中英文标题、来源、相似度、事件摘要、URL。 """ logger.info("search_news query=%r top_k=%d", query, top_k) hits = _search(query, top_k=top_k) return _fmt_results(hits, query) # ── Tool 2: search_by_stock ── @mcp.tool() def search_by_stock(stock_code: str, top_k: int = 10) -> str: """检索指定美股代码相关的投资事件。 参数: stock_code: 美股代码（如 AAPL、TSLA、MSFT，1-5 个大写字母） top_k: 返回条数（默认 10） 返回: Markdown 格式检索结果，含与该股票相关的事件、情绪和重要度。 """ code = stock_code.strip().upper() logger.info("search_by_stock code=%r top_k=%d", code, top_k) # 带股票过滤的搜索 hits = _search(code, top_k=top_k, search_filter=SearchFilter(stock_codes=[code])) if not hits: # 降级：纯语义搜索 logger.info("stock 精确命中 0 条，降级为纯语义搜索") hits = _search(code, top_k=top_k) return _fmt_results(hits, f"{code} 相关事件") # ── Tool 3: search_by_sentiment ── @mcp.tool() def search_by_sentiment(query: str, sentiment: str = "all", top_k: int = 20) -> str: """按情绪倾向检索新闻。 参数: query: 自然语言查询 sentiment: positive(利好) / negative(利空) / neutral(中性) / all(全部，默认) top_k: 返回条数（默认 20） 返回: Markdown 检索结果 + 情绪分布统计。 """ logger.info("search_by_sentiment query=%r sentiment=%r", query, sentiment) filt = None if sentiment in ("positive", "negative", "neutral"): filt = SearchFilter(sentiment=sentiment) hits = _search(query, top_k=top_k, search_filter=filt) # 情绪统计 pos = sum(1 for h in hits for ev in h.get("events", []) if ev.get("sentiment") == "positive") neg = sum(1 for h in hits for ev in h.get("events", []) if ev.get("sentiment") == "negative") neu = sum(1 for h in hits for ev in h.get("events", []) if ev.get("sentiment") == "neutral") header = ( f"## 📊 情绪趋势: {query}\n\n" f"共 {len(hits)} 条 | 🟢利好 {pos} | 🔴利空 {neg} | ⚪中性 {neu}\n" ) return header + "\n" + _fmt_results(hits, query) # ── Tool 4: get_today_events ── @mcp.tool() def get_today_events(importance_min: int = 4, limit: int = 15) -> str: """获取当日重要投资事件。 参数: importance_min: 最低重要度（默认 4） limit: 最多返回条数（默认 15） 返回: Markdown 格式的当日重要事件列表。 """ logger.info("get_today_events importance_min=%d limit=%d", importance_min, limit) events = _load_today_events() events = [e for e in events if e.get("importance", 0) >= importance_min] events = events[:limit] if not events: return "当日暂无符合条件的重要事件。" sentiment_map = {"positive": "🟢利好", "neutral": "⚪中性", "negative": "🔴利空"} lines = [ f"## 📅 今日重要事件（重要度 ≥ {importance_min}）", "", f"共 {len(events)} 条:", "", ] for i, ev in enumerate(events, 1): s = sentiment_map.get(ev.get("sentiment", ""), "") codes = ",".join(ev.get("stock_codes", [])[:5]) code_str = f" [{codes}]" if codes else "" lines.append(f"### {i}. [{ev.get('event_type', '')}]{s} 重要度{ev.get('importance', '')}") lines.append(f"- **{ev.get('title_zh', ev.get('title', ''))}**") lines.append(f"- {ev.get('summary_zh', '')}{code_str}") lines.append(f"- 来源: {ev.get('source_id', '')} | {ev.get('publish_time', '')[:10]}") lines.append(f"- 🔗 {ev.get('url', '')}") lines.append("") return "\n".join(lines) # ── Tool 5: get_stats ── @mcp.tool() def get_stats() -> str: """获取系统统计信息（M1-M6 管道数据、Qdrant 库规模等）。 返回: Markdown 格式的系统统计概览。 """ logger.info("get_stats") be = _get_backend() count = be.vector_store.count() info = be.vector_store.info() lines = [ "## 📊 国际财经 Deep Research 系统统计", "", f"**Collection**: {info.name}", f"**向量总数**: {count}", f"**嵌入模型**: {be.emb_config.model} ({be.emb_config.dimension} 维)", f"**Qdrant 模式**: {'文件模式' if not info.exists else '运行中'}", f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "", "### 可用工具", "", "- `search_news` — 语义检索新闻", "- `search_by_stock` — 按美股代码检索", "- `search_by_sentiment` — 按情绪过滤检索", "- `get_today_events` — 获取当日重要事件", "- `get_stats` — 系统统计概览", ] return "\n".join(lines)