MCP Server for stock and crypto

Instructions

Implementation Reference

• mcp_aktools/__init__.py:241-260 (handler)

  The main handler function for the 'stock_zt_pool_em' MCP tool, including registration decorator, input schema via Pydantic Fields, and logic to fetch, cache, process, and format A-share daily limit-up stocks data from akshare.

  @mcp.tool(
      title="A股涨停股池",
      description="获取中国A股市场(上证、深证)的所有涨停股票",
  )
  def stock_zt_pool_em(
      date: str = Field("", description="交易日日期(可选)，默认为最近的交易日，格式: 20251231"),
      limit: int = Field(50, description="返回数量(int,30-100)", strict=False),
  ):
      if not date:
          date = recent_trade_date().strftime("%Y%m%d")
      dfs = ak_cache(ak.stock_zt_pool_em, date=date, ttl=1200)
      cnt = len(dfs)
      try:
          dfs.drop(columns=["序号", "流通市值", "总市值"], inplace=True)
      except Exception:
          pass
      dfs.sort_values("成交额", ascending=False, inplace=True)
      dfs = dfs.head(int(limit))
      desc = f"共{cnt}只涨停股\n"
      return desc + dfs.to_csv(index=False, float_format="%.2f").strip()

• mcp_aktools/__init__.py:229-238 (helper)

  Helper function 'recent_trade_date' used by the tool to determine the default date (most recent trading day).

  def recent_trade_date():
      now = datetime.now().date()
      dfs = ak_cache(ak.tool_trade_date_hist_sina, ttl=43200)
      if dfs is None:
          return now
      dfs.sort_values("trade_date", ascending=False, inplace=True)
      for d in dfs["trade_date"]:
          if d <= now:
              return d
      return now

• mcp_aktools/__init__.py:564-579 (helper)

  Helper function 'ak_cache' used by the tool to cache akshare API calls with TTL.

  def ak_cache(fun, *args, **kwargs) -> pd.DataFrame | None:
      key = kwargs.pop("key", None)
      if not key:
          key = f"{fun.__name__}-{args}-{kwargs}"
      ttl1 = kwargs.pop("ttl", 86400)
      ttl2 = kwargs.pop("ttl2", None)
      cache = CacheKey.init(key, ttl1, ttl2)
      all = cache.get()
      if all is None:
          try:
              _LOGGER.info("Request akshare: %s", [key, args, kwargs])
              all = fun(*args, **kwargs)
              cache.set(all)
          except Exception as exc:
              _LOGGER.exception(str(exc))
      return all