mcp_server_fastmcp.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751
  1. from __future__ import annotations
  2. import asyncio
  3. import logging
  4. from collections import Counter
  5. from datetime import datetime, timezone
  6. from email.utils import parsedate_to_datetime
  7. from fastapi import FastAPI
  8. from mcp.server.fastmcp import FastMCP
  9. from mcp.server.transport_security import TransportSecuritySettings
  10. from news_mcp.config import DEFAULT_LOOKBACK_HOURS, DEFAULT_TOPICS, DB_PATH
  11. from news_mcp.config import (
  12. NEWS_PRUNE_INTERVAL_HOURS,
  13. NEWS_PRUNING_ENABLED,
  14. NEWS_REFRESH_INTERVAL_SECONDS,
  15. NEWS_BACKGROUND_REFRESH_ENABLED,
  16. NEWS_BACKGROUND_REFRESH_ON_START,
  17. NEWS_RETENTION_DAYS,
  18. )
  19. from news_mcp.jobs.poller import refresh_clusters
  20. from news_mcp.storage.sqlite_store import SQLiteClusterStore
  21. from news_mcp.enrichment.llm_enrich import summarize_cluster_llm
  22. from news_mcp.trends_resolution import resolve_entity_via_trends
  23. from news_mcp.llm import active_llm_config
  24. from news_mcp.entity_normalize import normalize_query
  25. from news_mcp.related_entities import related_recent_entities
  26. mcp = FastMCP(
  27. "news-mcp",
  28. transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
  29. )
  30. def _cluster_entity_haystack(cluster: dict) -> list[str]:
  31. """Collect the normalized entity clues attached to a cluster."""
  32. values: list[str] = []
  33. for ent in cluster.get("entities", []) or []:
  34. values.append(str(ent).strip().lower())
  35. for res in cluster.get("entityResolutions", []) or []:
  36. if not isinstance(res, dict):
  37. continue
  38. for key in ("normalized", "canonical_label", "mid"):
  39. val = res.get(key)
  40. if val:
  41. values.append(str(val).strip().lower())
  42. return [v for v in values if v]
  43. def _parse_cluster_timestamp(value) -> datetime:
  44. if not value:
  45. return datetime.min.replace(tzinfo=timezone.utc)
  46. text = str(value).strip()
  47. if not text:
  48. return datetime.min.replace(tzinfo=timezone.utc)
  49. try:
  50. dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
  51. if dt.tzinfo is None:
  52. dt = dt.replace(tzinfo=timezone.utc)
  53. return dt.astimezone(timezone.utc)
  54. except Exception:
  55. pass
  56. try:
  57. dt = parsedate_to_datetime(text)
  58. if dt.tzinfo is None:
  59. dt = dt.replace(tzinfo=timezone.utc)
  60. return dt.astimezone(timezone.utc)
  61. except Exception:
  62. return datetime.min.replace(tzinfo=timezone.utc)
  63. def _sort_clusters_by_recency(clusters: list[dict]) -> list[dict]:
  64. return sorted(
  65. clusters,
  66. key=lambda c: (
  67. _parse_cluster_timestamp(c.get("timestamp")),
  68. float(c.get("importance", 0.0) or 0.0),
  69. ),
  70. reverse=True,
  71. )
  72. def _tool_card(name: str, description: str, inputs: list[dict], outputs: list[str], notes: list[str] | None = None) -> dict:
  73. return {
  74. "name": name,
  75. "description": description,
  76. "inputs": inputs,
  77. "outputs": outputs,
  78. "notes": notes or [],
  79. }
  80. NEWS_TOOL_CARDS = [
  81. _tool_card(
  82. "get_latest_events",
  83. "Get the newest deduplicated clusters for a topic or resolved entity-like query.",
  84. [
  85. {"name": "topic", "type": "string", "default": "crypto", "meaning": "coarse category or entity-like topic"},
  86. {"name": "limit", "type": "integer", "default": 5, "range": "1-20"},
  87. {"name": "include_articles", "type": "boolean", "default": False},
  88. ],
  89. ["headline", "summary", "entities", "sentiment", "importance", "sources", "timestamp", "articles?"],
  90. ["Use when you want the freshest clusters and are willing to let the server decide topic vs entity mode."],
  91. ),
  92. _tool_card(
  93. "get_events_for_entity",
  94. "Search recent clusters for a person, place, company, or theme by entity matching.",
  95. [
  96. {"name": "entity", "type": "string", "meaning": "entity label or phrase"},
  97. {"name": "timeframe", "type": "string", "default": "24h", "examples": ["24h", "72h", "3d"]},
  98. {"name": "limit", "type": "integer", "default": 10, "range": "1-30"},
  99. {"name": "include_articles", "type": "boolean", "default": False},
  100. ],
  101. ["headline", "summary", "entities", "sentiment", "importance", "sources", "timestamp", "articles?"],
  102. ["Normalization is automatic; use this for an entity-centered deep dive."],
  103. ),
  104. _tool_card(
  105. "get_event_summary",
  106. "Produce a concise LLM-written explanation for one cluster and key facts.",
  107. [
  108. {"name": "event_id", "type": "string", "meaning": "cluster_id; do not surface in user-facing prose"},
  109. {"name": "include_articles", "type": "boolean", "default": False},
  110. ],
  111. ["headline", "mergedSummary", "keyFacts", "sources", "articles?"],
  112. ["Prefer this after you have already chosen a specific cluster to explain."],
  113. ),
  114. _tool_card(
  115. "detect_emerging_topics",
  116. "Surface entities and phrases starting to matter in the recent window.",
  117. [{"name": "limit", "type": "integer", "default": 10, "range": "1-20"}],
  118. ["topic", "trend_score", "related_entities", "signal_type", "count", "avg_importance"],
  119. ["Good for 'what is heating up?' style questions."],
  120. ),
  121. _tool_card(
  122. "get_news_sentiment",
  123. "Estimate sentiment around an entity over a lookback window.",
  124. [
  125. {"name": "entity", "type": "string"},
  126. {"name": "timeframe", "type": "string", "default": "24h"},
  127. ],
  128. ["entity", "sentiment", "score", "cluster_count"],
  129. ["Use after locating a cluster set or entity neighborhood."],
  130. ),
  131. _tool_card(
  132. "get_related_recent_entities",
  133. "Blend local co-occurrence with Google Trends related topics, while preserving mids where available.",
  134. [
  135. {"name": "subject", "type": "string", "meaning": "canonical entity or subject phrase"},
  136. {"name": "timeframe", "type": "string", "default": "72h"},
  137. {"name": "limit", "type": "integer", "default": 10, "range": "1-25"},
  138. {"name": "include_trends", "type": "boolean", "default": True},
  139. ],
  140. ["subject", "related[].normalized", "related[].canonical_label", "related[].mid", "related[].sources", "related[].scores"],
  141. ["Use this to drill from a subject into related entities, then feed those into get_events_for_entity."],
  142. ),
  143. ]
  144. NEWS_COMPOSITION_RECIPES = [
  145. {
  146. "name": "fresh-news-tail",
  147. "steps": [
  148. "get_latest_events(topic=...)",
  149. "optionally get_event_summary(event_id=...) for the strongest cluster",
  150. ],
  151. "notes": ["Best for a quick tail of what is happening now."]
  152. },
  153. {
  154. "name": "entity-deep-dive",
  155. "steps": [
  156. "get_events_for_entity(entity=...)",
  157. "get_event_summary(event_id=...)",
  158. "get_news_sentiment(entity=..., timeframe=...)",
  159. ],
  160. "notes": ["Prefer canonical entity labels when you have them; the server normalizes for you."],
  161. },
  162. {
  163. "name": "subject-neighborhood",
  164. "steps": [
  165. "get_related_recent_entities(subject=...)",
  166. "for each strong related entity, call get_events_for_entity(entity=...)",
  167. ],
  168. "notes": ["Use this when you want a graph-like expansion around a subject."]
  169. },
  170. {
  171. "name": "emerging-signal",
  172. "steps": [
  173. "detect_emerging_topics(limit=...)",
  174. "choose a topic/entity",
  175. "get_events_for_entity(entity=...)",
  176. "get_news_sentiment(entity=...)",
  177. ],
  178. "notes": ["Good for trend scouting and risk mapping."],
  179. },
  180. ]
  181. NEWS_AGENT_TIPS = [
  182. "If you need a fast answer, start with get_latest_events, then summarize the strongest cluster with get_event_summary.",
  183. "If a user asks about a person/place/company/theme, use get_events_for_entity before broadening to get_related_recent_entities.",
  184. "Treat cluster_id as an internal cursor, not user-facing output; use it only for follow-up tool calls.",
  185. "When describing clusters, keep sources and timestamps visible so the user can assess recency and provenance.",
  186. "Prefer a short chain of tools over many parallel calls unless you are building a neighborhood map or comparison table.",
  187. "For tricky names, rely on the server’s resolver instead of inventing alias rules in the client.",
  188. ]
  189. NEWS_EXAMPLE_CHAINS = [
  190. {
  191. "task": "What is happening now?",
  192. "chain": [
  193. "get_latest_events(topic=...)",
  194. "get_event_summary(event_id=...) if one cluster looks important",
  195. ],
  196. },
  197. {
  198. "task": "Deep dive on an entity",
  199. "chain": [
  200. "get_events_for_entity(entity=..., timeframe=...)",
  201. "get_news_sentiment(entity=..., timeframe=...)",
  202. "get_event_summary(event_id=...) for the strongest cluster",
  203. ],
  204. },
  205. {
  206. "task": "Broaden from a subject",
  207. "chain": [
  208. "get_related_recent_entities(subject=..., include_trends=true)",
  209. "get_events_for_entity(entity=...) for the strongest related entities",
  210. ],
  211. },
  212. {
  213. "task": "Find what is emerging",
  214. "chain": [
  215. "detect_emerging_topics(limit=...)",
  216. "get_events_for_entity(entity=...) on one or two emerging terms",
  217. ],
  218. },
  219. ]
  220. @mcp.tool(description="Investigate a topic and return the newest deduplicated news clusters, sorted by recency.")
  221. async def get_latest_events(topic: str = "crypto", limit: int = 5, include_articles: bool = False):
  222. limit = max(1, min(int(limit), 20))
  223. # If the caller passes an entity-like value, resolve it and use the canonical
  224. # entity as the query lens. Otherwise keep the original topic path.
  225. topic_norm = normalize_query(topic).lower()
  226. resolved = resolve_entity_via_trends(topic_norm)
  227. allowed = {t.lower() for t in DEFAULT_TOPICS}
  228. is_topic = topic_norm in allowed
  229. query_terms = {
  230. topic_norm,
  231. str(resolved.get("normalized") or "").strip().lower(),
  232. str(resolved.get("canonical_label") or "").strip().lower(),
  233. str(resolved.get("mid") or "").strip().lower(),
  234. }
  235. query_terms = {q for q in query_terms if q}
  236. store = SQLiteClusterStore(DB_PATH)
  237. if is_topic:
  238. # Cache-first: only refresh if we currently have no fresh clusters for this topic.
  239. clusters = store.get_latest_clusters(topic=topic_norm, ttl_hours=DEFAULT_LOOKBACK_HOURS, limit=limit)
  240. if not clusters:
  241. await refresh_clusters(topic=topic_norm, limit=200)
  242. clusters = store.get_latest_clusters(topic=topic_norm, ttl_hours=DEFAULT_LOOKBACK_HOURS, limit=limit)
  243. else:
  244. # Entity-aware mode: search recent clusters across all topics and match by
  245. # raw entity, canonical label, or MID.
  246. clusters = store.get_latest_clusters_all_topics(ttl_hours=DEFAULT_LOOKBACK_HOURS, limit=limit * 8)
  247. filtered = []
  248. for c in clusters:
  249. haystack = _cluster_entity_haystack(c)
  250. if any(any(term in item for item in haystack) for term in query_terms):
  251. filtered.append(c)
  252. if len(filtered) >= limit:
  253. break
  254. clusters = filtered
  255. out = []
  256. for c in _sort_clusters_by_recency(clusters):
  257. item = {
  258. "cluster_id": c.get("cluster_id"),
  259. "headline": c.get("headline"),
  260. "summary": c.get("summary"),
  261. "entities": c.get("entities", []),
  262. "sentiment": c.get("sentiment", "neutral"),
  263. "importance": c.get("importance", 0.0),
  264. "sources": c.get("sources", []),
  265. "timestamp": c.get("timestamp"),
  266. }
  267. if include_articles:
  268. # Return minimal article fields to keep responses compact.
  269. arts = c.get("articles", []) or []
  270. item["articles"] = [
  271. {
  272. "title": a.get("title"),
  273. "url": a.get("url"),
  274. "source": a.get("source"),
  275. "timestamp": a.get("timestamp"),
  276. }
  277. for a in arts
  278. if isinstance(a, dict)
  279. ]
  280. out.append(item)
  281. return out
  282. @mcp.tool(description="Investigate a person, company, place, or theme by matching extracted entities within a time window.")
  283. async def get_events_for_entity(entity: str, limit: int = 10, timeframe: str = "24h", include_articles: bool = False):
  284. limit = max(1, min(int(limit), 30))
  285. query = normalize_query(entity).strip().lower()
  286. if not query:
  287. return []
  288. resolved = resolve_entity_via_trends(query)
  289. query_terms = {
  290. query,
  291. str(resolved.get("normalized") or "").strip().lower(),
  292. str(resolved.get("canonical_label") or "").strip().lower(),
  293. str(resolved.get("mid") or "").strip().lower(),
  294. }
  295. query_terms = {q for q in query_terms if q}
  296. store = SQLiteClusterStore(DB_PATH)
  297. def _match_clusters(clusters: list[dict]) -> list[dict]:
  298. hits: list[dict] = []
  299. for c in _sort_clusters_by_recency(clusters):
  300. haystack = _cluster_entity_haystack(c)
  301. if any(any(term in item for item in haystack) for term in query_terms):
  302. hits.append(c)
  303. if len(hits) >= limit:
  304. break
  305. return hits
  306. hours = _parse_timeframe_to_hours(timeframe)
  307. clusters = store.get_latest_clusters_all_topics(ttl_hours=hours, limit=max(200, limit * 10))
  308. hits = _match_clusters(clusters)
  309. out = []
  310. for c in hits:
  311. item = {
  312. "cluster_id": c.get("cluster_id"),
  313. "headline": c.get("headline"),
  314. "summary": c.get("summary"),
  315. "entities": c.get("entities", []),
  316. "sentiment": c.get("sentiment", "neutral"),
  317. "importance": c.get("importance", 0.0),
  318. "sources": c.get("sources", []),
  319. "timestamp": c.get("timestamp"),
  320. }
  321. if include_articles:
  322. arts = c.get("articles", []) or []
  323. item["articles"] = [
  324. {
  325. "title": a.get("title"),
  326. "url": a.get("url"),
  327. "source": a.get("source"),
  328. "timestamp": a.get("timestamp"),
  329. }
  330. for a in arts
  331. if isinstance(a, dict)
  332. ]
  333. out.append(item)
  334. return out
  335. @mcp.tool(description="Return entities most commonly associated with the subject in recent clusters, optionally blended with Google Trends suggestions.")
  336. async def get_related_recent_entities(subject: str, timeframe: str = "72h", limit: int = 10, include_trends: bool = True):
  337. limit = max(1, min(int(limit), 25))
  338. hours = _parse_timeframe_to_hours(timeframe)
  339. include_trends_bool = str(include_trends).strip().lower() not in {"false", "0", "no"}
  340. store = SQLiteClusterStore(DB_PATH)
  341. result = related_recent_entities(
  342. store=store,
  343. subject=subject,
  344. timeframe_hours=hours,
  345. limit=limit,
  346. include_trends=include_trends_bool,
  347. )
  348. return result
  349. @mcp.tool(description="Investigate one cluster in depth and return a concise LLM-written explanation plus key facts.")
  350. async def get_event_summary(event_id: str, include_articles: bool = False):
  351. store = SQLiteClusterStore(DB_PATH)
  352. # Summary cache: reuse if present within TTL.
  353. cached_summary = store.get_cluster_summary(
  354. cluster_id=event_id,
  355. ttl_hours=DEFAULT_LOOKBACK_HOURS,
  356. )
  357. if cached_summary:
  358. out = {
  359. "event_id": event_id,
  360. "headline": cached_summary.get("headline"),
  361. "mergedSummary": cached_summary.get("mergedSummary"),
  362. "keyFacts": cached_summary.get("keyFacts", []),
  363. "sources": cached_summary.get("sources", []),
  364. }
  365. if include_articles:
  366. cluster = store.get_cluster_by_id(event_id)
  367. arts = (cluster or {}).get("articles", []) or []
  368. out["articles"] = [
  369. {
  370. "title": a.get("title"),
  371. "url": a.get("url"),
  372. "source": a.get("source"),
  373. "timestamp": a.get("timestamp"),
  374. }
  375. for a in arts
  376. if isinstance(a, dict)
  377. ]
  378. return out
  379. cluster = store.get_cluster_by_id(event_id)
  380. if not cluster:
  381. return {
  382. "event_id": event_id,
  383. "error": "NOT_FOUND",
  384. }
  385. articles_out = None
  386. if include_articles:
  387. arts = cluster.get("articles", []) or []
  388. articles_out = [
  389. {
  390. "title": a.get("title"),
  391. "url": a.get("url"),
  392. "source": a.get("source"),
  393. "timestamp": a.get("timestamp"),
  394. }
  395. for a in arts
  396. if isinstance(a, dict)
  397. ]
  398. summary = await summarize_cluster_llm(cluster)
  399. store.upsert_cluster_summary(event_id, summary)
  400. out = {
  401. "event_id": event_id,
  402. "headline": summary.get("headline"),
  403. "mergedSummary": summary.get("mergedSummary"),
  404. "keyFacts": summary.get("keyFacts", []),
  405. "sources": summary.get("sources", []),
  406. }
  407. if include_articles:
  408. out["articles"] = articles_out or []
  409. return out
  410. @mcp.tool(description="Explore what is starting to matter: surface emerging entities and phrases from recent clusters.")
  411. async def detect_emerging_topics(limit: int = 10):
  412. limit = max(1, min(int(limit), 20))
  413. store = SQLiteClusterStore(DB_PATH)
  414. clusters = store.get_latest_clusters_all_topics(ttl_hours=DEFAULT_LOOKBACK_HOURS, limit=200)
  415. import re
  416. entity_counts = Counter()
  417. entity_importance_sum = Counter()
  418. # co-occurrence: ent -> other_ent -> count
  419. entity_cooccur = {}
  420. phrase_counts = Counter()
  421. topic_counts = Counter()
  422. # Very light heuristics to reduce “meta entities” dominating emerging topics.
  423. # Keep it conservative: only skip obvious boilerplate.
  424. def _is_generic_entity(ent: str) -> bool:
  425. e = str(ent).strip().lower()
  426. if not e:
  427. return True
  428. if len(e) < 4:
  429. return True
  430. # common outlet-ish / meta-ish tokens
  431. if e in {"news", "latest", "breaking"}:
  432. return True
  433. return False
  434. for c in clusters:
  435. topic_counts[c.get("topic", "other")] += 1
  436. ents_in_cluster = [e for e in (c.get("entities", []) or []) if not _is_generic_entity(e)]
  437. ents_in_cluster_norm = [str(e).strip().lower() for e in ents_in_cluster if str(e).strip()]
  438. for ent in ents_in_cluster_norm:
  439. if _is_generic_entity(ent):
  440. continue
  441. entity_counts[ent] += 1
  442. try:
  443. entity_importance_sum[ent] += float(c.get("importance", 0.0) or 0.0)
  444. except Exception:
  445. pass
  446. # update co-occurrence counts
  447. for i in range(len(ents_in_cluster_norm)):
  448. a = ents_in_cluster_norm[i]
  449. if not a:
  450. continue
  451. entity_cooccur.setdefault(a, Counter())
  452. for j in range(len(ents_in_cluster_norm)):
  453. if i == j:
  454. continue
  455. b = ents_in_cluster_norm[j]
  456. if not b:
  457. continue
  458. entity_cooccur[a][b] += 1
  459. text = f"{c.get('headline','')} {c.get('summary','')}"
  460. words = [w for w in re.findall(r"[A-Za-z][A-Za-z0-9\-]{2,}", text.lower())]
  461. for i in range(len(words) - 1):
  462. phrase = f"{words[i]} {words[i+1]}"
  463. if len(phrase) > 6:
  464. phrase_counts[phrase] += 1
  465. emerging = []
  466. # Combine frequency with average importance so “big signal” rises over pure repetition.
  467. for ent, count in entity_counts.most_common(limit):
  468. avg_imp = entity_importance_sum[ent] / max(1, count)
  469. # avg_imp is typically 0..~1; keep score bounded.
  470. trend_score = 0.25 + 0.40 * min(1.0, avg_imp) + 0.08 * min(6.0, float(count))
  471. related = []
  472. for other, _cnt in (entity_cooccur.get(ent) or Counter()).most_common(3):
  473. # avoid returning the entity itself (shouldn't happen, but be safe)
  474. if other != ent:
  475. related.append(other)
  476. emerging.append({
  477. "topic": ent,
  478. "trend_score": min(0.99, round(trend_score, 2)),
  479. "related_entities": related if related else [ent],
  480. "signal_type": "entity",
  481. "count": count,
  482. "avg_importance": round(avg_imp, 3),
  483. })
  484. for phrase, count in phrase_counts.most_common(limit * 2):
  485. if any(item["topic"] == phrase for item in emerging):
  486. continue
  487. emerging.append({
  488. "topic": phrase.title(),
  489. "trend_score": min(0.99, round(0.20 + 0.10 * count, 2)),
  490. "related_entities": [],
  491. "signal_type": "phrase",
  492. "count": count,
  493. })
  494. if len(emerging) >= limit:
  495. break
  496. return emerging[:limit]
  497. @mcp.tool(description="Investigate whether sentiment around an entity is positive, negative, or neutral over a chosen lookback window.")
  498. async def get_news_sentiment(entity: str, timeframe: str = "24h"):
  499. store = SQLiteClusterStore(DB_PATH)
  500. ent = normalize_query(entity).strip().lower()
  501. resolved = resolve_entity_via_trends(ent)
  502. query_terms = {
  503. ent,
  504. str(resolved.get("normalized") or "").strip().lower(),
  505. str(resolved.get("canonical_label") or "").strip().lower(),
  506. str(resolved.get("mid") or "").strip().lower(),
  507. }
  508. query_terms = {q for q in query_terms if q}
  509. if not ent:
  510. return {
  511. "entity": entity,
  512. "sentiment": "neutral",
  513. "score": 0.0,
  514. "cluster_count": 0,
  515. }
  516. # timeframe: accept '24h' or '24'
  517. tf = str(timeframe).strip().lower()
  518. try:
  519. hours = int(tf[:-1]) if tf.endswith("h") else int(tf)
  520. except Exception:
  521. hours = 24
  522. hours = max(1, min(int(hours), 168))
  523. clusters = store.get_latest_clusters_all_topics(ttl_hours=hours, limit=500)
  524. matched = []
  525. for c in clusters:
  526. haystack = _cluster_entity_haystack(c)
  527. if any(any(term in item for item in haystack) for term in query_terms):
  528. matched.append(c)
  529. if not matched:
  530. return {
  531. "entity": entity,
  532. "sentiment": "neutral",
  533. "score": 0.0,
  534. "cluster_count": 0,
  535. }
  536. scores = []
  537. for c in matched:
  538. s = c.get("sentimentScore")
  539. if s is not None:
  540. try:
  541. scores.append(float(s))
  542. except Exception:
  543. pass
  544. avg_score = sum(scores) / len(scores) if scores else 0.0
  545. # Keep the label aligned with the numeric score.
  546. # Small magnitudes are treated as neutral to avoid noisy label flips.
  547. if avg_score >= 0.15:
  548. sentiment = "positive"
  549. elif avg_score <= -0.15:
  550. sentiment = "negative"
  551. else:
  552. sentiment = "neutral"
  553. return {
  554. "entity": entity,
  555. "sentiment": sentiment,
  556. "score": round(avg_score, 3),
  557. "cluster_count": len(matched),
  558. }
  559. @mcp.tool(description="Describe the server tool surface, how tools fit together, and output conventions for downstream agents.")
  560. async def get_capabilities():
  561. return {
  562. "server": {
  563. "name": "news-mcp",
  564. "purpose": "Recent news clusters, entity drill-down, sentiment, emerging topics, and related-entity expansion.",
  565. "output_conventions": {
  566. "cluster_ids": "Do not surface cluster_id in user-facing prose unless explicitly requested; treat it as internal navigation metadata.",
  567. "sources": "Always preserve and display sources when summarizing a cluster or entity result.",
  568. "timestamps": "Mention timestamps consistently when comparing multiple clusters or when recency matters.",
  569. },
  570. },
  571. "tools": NEWS_TOOL_CARDS,
  572. "recipes": NEWS_COMPOSITION_RECIPES,
  573. "example_chains": NEWS_EXAMPLE_CHAINS,
  574. "agent_tips": NEWS_AGENT_TIPS,
  575. "guidance": [
  576. "Use get_latest_events for a tail, get_events_for_entity for entity deep dives, and get_related_recent_entities for neighborhood expansion.",
  577. "Prefer normalized/canonical entities when possible, but the server will resolve common aliases and MIDs for you.",
  578. "When presenting results to users, summarize the cluster; avoid exposing internal IDs unless they are needed for follow-up tool calls.",
  579. ],
  580. }
  581. def _parse_timeframe_to_hours(timeframe: str) -> int:
  582. tf = str(timeframe).strip().lower()
  583. try:
  584. if tf.endswith("d"):
  585. days = int(tf[:-1])
  586. return max(1, days * 24)
  587. if tf.endswith("h"):
  588. return max(1, int(tf[:-1]))
  589. return max(1, int(tf))
  590. except Exception:
  591. return 24
  592. app = FastAPI(title="News MCP Server")
  593. logger = logging.getLogger("news_mcp.startup")
  594. app.mount("/mcp", mcp.sse_app())
  595. _background_task_started = False
  596. @app.on_event("startup")
  597. async def _start_background_refresh():
  598. global _background_task_started
  599. if _background_task_started:
  600. return
  601. if not NEWS_BACKGROUND_REFRESH_ENABLED:
  602. return
  603. _background_task_started = True
  604. logger.info("news-mcp llm config: %s", active_llm_config())
  605. store = SQLiteClusterStore(DB_PATH)
  606. prune_result = store.prune_if_due(
  607. pruning_enabled=NEWS_PRUNING_ENABLED,
  608. retention_days=NEWS_RETENTION_DAYS,
  609. interval_hours=NEWS_PRUNE_INTERVAL_HOURS,
  610. )
  611. logger.info("startup prune_result=%s", prune_result)
  612. async def _loop():
  613. if not NEWS_BACKGROUND_REFRESH_ON_START:
  614. await asyncio.sleep(float(NEWS_REFRESH_INTERVAL_SECONDS))
  615. while True:
  616. try:
  617. # Refresh all topics by passing topic=None
  618. await refresh_clusters(topic=None, limit=200)
  619. except Exception:
  620. # Avoid crashing the server on network errors.
  621. pass
  622. await asyncio.sleep(float(NEWS_REFRESH_INTERVAL_SECONDS))
  623. asyncio.create_task(_loop())
  624. @app.get("/")
  625. def root():
  626. return {
  627. "status": "ok",
  628. "transport": "fastmcp+sse",
  629. "mount": "/mcp",
  630. "tools": [
  631. "get_latest_events",
  632. "get_events_for_entity",
  633. "get_event_summary",
  634. "detect_emerging_topics",
  635. "get_news_sentiment",
  636. "get_related_recent_entities",
  637. "get_capabilities",
  638. ],
  639. "refresh": {
  640. "enabled": NEWS_BACKGROUND_REFRESH_ENABLED,
  641. "interval_seconds": NEWS_REFRESH_INTERVAL_SECONDS,
  642. },
  643. "retention": {
  644. "lookback_hours": DEFAULT_LOOKBACK_HOURS,
  645. "retention_days": NEWS_RETENTION_DAYS,
  646. },
  647. "pruning": {
  648. "enabled": NEWS_PRUNING_ENABLED,
  649. "interval_hours": NEWS_PRUNE_INTERVAL_HOURS,
  650. },
  651. }
  652. @app.get("/health")
  653. def health():
  654. store = SQLiteClusterStore(DB_PATH)
  655. return {
  656. "status": "ok",
  657. "lookback_hours": DEFAULT_LOOKBACK_HOURS,
  658. "db": str(DB_PATH),
  659. "refresh": store.get_feed_state("breakingthenews"),
  660. "pruning": store.get_prune_state(
  661. pruning_enabled=NEWS_PRUNING_ENABLED,
  662. retention_days=NEWS_RETENTION_DAYS,
  663. interval_hours=NEWS_PRUNE_INTERVAL_HOURS,
  664. ),
  665. }