MCP Server for stock and crypto

Instructions

Implementation Reference

• mcp_aktools/__init__.py:130-145 (handler)

  The primary handler for the 'stock_news' tool. Registered via @mcp.tool decorator with title and description. Defines input schema for symbol and optional limit. Executes by caching results from stock_news_em helper, extracting unique news content, and returning formatted list.

  @mcp.tool(
      title="获取股票/加密货币相关新闻",
      description="根据股票代码或加密货币符号获取近期相关新闻",
  )
  def stock_news(
      symbol: str = Field(description="股票代码/加密货币符号"),
      limit: int = Field(15, description="返回数量(int)", strict=False),
  ):
      news = list(dict.fromkeys([
          v["新闻内容"]
          for v in ak_cache(stock_news_em, symbol=symbol, ttl=3600).to_dict(orient="records")
          if isinstance(v, dict)
      ]))
      if news:
          return "\n".join(news[0:limit])
      return f"Not Found for {symbol}"

• mcp_aktools/__init__.py:147-170 (helper)

  Supporting helper function stock_news_em that retrieves news data from East Money search API using a JSONP request with specific parameters for the symbol. Parses the response, creates and sorts a DataFrame by date, limits results, cleans HTML tags from content, and returns the processed DataFrame.

  def stock_news_em(symbol, limit=20):
      cbk = "jQuery351013927587392975826_1763361926020"
      resp = requests.get(
          "http://search-api-web.eastmoney.com/search/jsonp",
          headers={
              "User-Agent": USER_AGENT,
              "Referer": f"https://so.eastmoney.com/news/s?keyword={symbol}",
          },
          params={
              "cb": cbk,
              "param": '{"uid":"",'
                       f'"keyword":"{symbol}",'
                       '"type":["cmsArticleWebOld"],"client":"web","clientType":"web","clientVersion":"curr",'
                       '"param":{"cmsArticleWebOld":{"searchScope":"default","sort":"default","pageIndex":1,"pageSize":10,'
                       '"preTag":"<em>","postTag":"</em>"}}}',
          },
      )
      text = resp.text.replace(cbk, "").strip().strip("()")
      data = json.loads(text) or {}
      dfs = pd.DataFrame(data.get("result", {}).get("cmsArticleWebOld") or [])
      dfs.sort_values("date", ascending=False, inplace=True)
      dfs = dfs.head(limit)
      dfs["新闻内容"] = dfs["content"].str.replace(r"</?em>", "", regex=True)
      return dfs