get_high_score_news
Retrieve cryptocurrency news articles filtered by AI-generated quality scores to identify high-impact market information for trading decisions.
Instructions
Get highly-rated news articles (by AI score), sorted by score descending.
Args: min_score: Minimum score threshold (default 70). limit: Maximum results to return (default 10, max 100).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| min_score | No | ||
| limit | No |
Implementation Reference
- src/opennews_mcp/tools/news.py:173-200 (handler)Main implementation of get_high_score_news tool. Fetches news articles, filters them by AI score threshold (min_score), sorts by score descending, and returns the top results. Uses @mcp.tool() decorator for registration.
@mcp.tool() async def get_high_score_news(ctx: Context, min_score: int = 70, limit: int = 10) -> dict: """Get highly-rated news articles (by AI score), sorted by score descending. Args: min_score: Minimum score threshold (default 70). limit: Maximum results to return (default 10, max 100). """ api = ctx.request_context.lifespan_context.api limit = clamp_limit(limit) try: fetch_limit = min(limit * 3, MAX_ROWS) result = await api.search_news(limit=fetch_limit, page=1) raw = result.get("data", []) filtered = [it for it in raw if (it.get("aiRating") or {}).get("score", 0) >= min_score] filtered.sort( key=lambda x: (x.get("aiRating") or {}).get("score", 0), reverse=True, ) data = filtered[:limit] return make_serializable({ "success": True, "min_score": min_score, "data": data, "count": len(data), }) except Exception as e: return {"success": False, "error": str(e) or repr(e)} - src/opennews_mcp/tools/news.py:173-173 (registration)Tool registration via @mcp.tool() decorator, which registers get_high_score_news as an available MCP tool.
@mcp.tool() - src/opennews_mcp/config.py:38-40 (helper)clamp_limit helper function used by get_high_score_news to validate and constrain the user-provided limit parameter to [1, MAX_ROWS].
def clamp_limit(limit: int) -> int: """Clamp user-supplied limit to [1, MAX_ROWS].""" return min(max(1, limit), MAX_ROWS) - src/opennews_mcp/config.py:43-57 (helper)make_serializable helper function used by get_high_score_news to recursively convert non-JSON-serializable types (datetime, Decimal, bytes) into JSON-compatible formats.
def make_serializable(obj): """Recursively convert non-JSON-serializable types.""" if obj is None: return None if isinstance(obj, dict): return {k: make_serializable(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [make_serializable(item) for item in obj] if isinstance(obj, (datetime, date)): return obj.isoformat() if isinstance(obj, Decimal): return float(obj) if isinstance(obj, bytes): return obj.decode("utf-8", errors="replace") return obj - src/opennews_mcp/api_client.py:78-99 (helper)search_news API client method called by get_high_score_news to fetch news articles from the REST API endpoint POST /open/news_search.
async def search_news( self, coins: Optional[list[str]] = None, query: Optional[str] = None, engine_types: Optional[dict[str, list[str]]] = None, has_coin: bool = False, limit: int = 20, page: int = 1, ) -> dict: """POST /open/news_search — 搜索新闻文章""" body: dict[str, Any] = {"limit": limit, "page": page} if coins: body["coins"] = coins if query: body["q"] = query if engine_types: body["engineTypes"] = engine_types if has_coin: body["hasCoin"] = has_coin resp = await self._request("POST", f"{self.base_url}/open/news_search", json=body) return resp.json()