OpenNews MCP

Instructions

Implementation Reference

• src/opennews_mcp/tools/news.py:13-32 (handler)

  Main handler function get_latest_news that retrieves the most recent crypto news articles. It uses the API client's search_news method, applies limit clamping, and returns results in a serializable format with error handling.

  @mcp.tool()
  async def get_latest_news(ctx: Context, limit: int = 10) -> dict:
      """Get the most recent crypto news articles, newest first.
  
      Returns news with title text, source, link, related coins, AI rating, and tags.
  
      Args:
          limit: Maximum number of articles to return (default 10, max 100).
      """
      api = ctx.request_context.lifespan_context.api
      limit = clamp_limit(limit)
      try:
          result = await api.search_news(limit=limit, page=1)
          data = result.get("data", [])[:limit]
          return make_serializable({
              "success": True, "data": data,
              "count": len(data), "total": result.get("total", 0),
          })
      except Exception as e:
          return {"success": False, "error": str(e) or repr(e)}

• src/opennews_mcp/tools/news.py:13-13 (registration)

  Tool registration via @mcp.tool() decorator on get_latest_news function, which registers it with the FastMCP instance.

  @mcp.tool()

• src/opennews_mcp/app.py:36-40 (registration)

  FastMCP instance creation with app_lifespan context manager and json_response=True configuration. This mcp instance is what the @mcp.tool() decorator registers tools to.

  mcp = FastMCP(
      "opennews-6551",
      lifespan=app_lifespan,
      json_response=True,
  )

• src/opennews_mcp/api_client.py:78-99 (helper)

  API client method search_news that performs POST request to /open/news_search endpoint. Takes optional filters (coins, query, engine_types, has_coin), limit, and page parameters.

  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()

• src/opennews_mcp/config.py:38-57 (helper)

  Helper utilities: clamp_limit() ensures limit is within [1, MAX_ROWS] bounds, and make_serializable() recursively converts non-JSON-serializable types (datetime, Decimal, bytes, etc.) for API responses.

  def clamp_limit(limit: int) -> int:
      """Clamp user-supplied limit to [1, MAX_ROWS]."""
      return min(max(1, limit), MAX_ROWS)
  
  
  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