AkTools MCP Server

Instructions

Implementation Reference

• mcp_aktools/__init__.py:75-108 (handler)

  Core implementation of the 'stock_prices' tool. Handles input parameters, determines start date, selects appropriate akshare function based on market, fetches data with caching, adds technical indicators, and returns formatted recent historical prices.

  def stock_prices(
      symbol: str = field_symbol,
      market: str = field_market,
      period: str = Field("daily", description="周期，如: daily(日线), weekly(周线，不支持美股)"),
      limit: int = Field(30, description="返回数量(int)", strict=False),
  ):
      if period == "weekly":
          delta = {"weeks": limit + 62}
      else:
          delta = {"days": limit + 62}
      start_date = (datetime.now() - timedelta(**delta)).strftime("%Y%m%d")
      markets = [
          ["sh", ak.stock_zh_a_hist, {}],
          ["sz", ak.stock_zh_a_hist, {}],
          ["hk", ak.stock_hk_hist, {}],
          ["us", stock_us_daily, {}],
          ["sh", fund_etf_hist_sina, {"market": "sh"}],
          ["sz", fund_etf_hist_sina, {"market": "sz"}],
      ]
      for m in markets:
          if m[0] != market:
              continue
          kws = {"period": period, "start_date": start_date, **m[2]}
          dfs = ak_cache(m[1], symbol=symbol, ttl=3600, **kws)
          if dfs is None or dfs.empty:
              continue
          add_technical_indicators(dfs, dfs["收盘"], dfs["最低"], dfs["最高"])
          columns = [
              "日期", "开盘", "收盘", "最高", "最低", "成交量", "换手率",
              "MACD", "DIF", "DEA", "KDJ.K", "KDJ.D", "KDJ.J", "RSI", "BOLL.U", "BOLL.M", "BOLL.L",
          ]
          all = dfs.to_csv(columns=columns, index=False, float_format="%.2f").strip().split("\n")
          return "\n".join([all[0], *all[-limit:]])
      return f"Not Found for {symbol}.{market}"

• mcp_aktools/__init__.py:71-74 (registration)

  Registers the stock_prices function as an MCP tool with title and description.

  @mcp.tool(
      title="获取股票历史价格",
      description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币",
  )

• mcp_aktools/__init__.py:76-80 (schema)

  Pydantic schema definitions for the tool inputs using Field for validation and descriptions.

      symbol: str = field_symbol,
      market: str = field_market,
      period: str = Field("daily", description="周期，如: daily(日线), weekly(周线，不支持美股)"),
      limit: int = Field(30, description="返回数量(int)", strict=False),
  ):

• mcp_aktools/__init__.py:581-611 (helper)

  Supporting function called by stock_prices to add technical indicators (MACD, KDJ, RSI, Bollinger Bands) to the dataframe.

  def add_technical_indicators(df, clos, lows, high):
      # 计算MACD指标
      ema12 = clos.ewm(span=12, adjust=False).mean()
      ema26 = clos.ewm(span=26, adjust=False).mean()
      df["DIF"] = ema12 - ema26
      df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean()
      df["MACD"] = (df["DIF"] - df["DEA"]) * 2
  
      # 计算KDJ指标
      low_min  = lows.rolling(window=9, min_periods=1).min()
      high_max = high.rolling(window=9, min_periods=1).max()
      rsv = (clos - low_min) / (high_max - low_min) * 100
      df["KDJ.K"] = rsv.ewm(com=2, adjust=False).mean()
      df["KDJ.D"] = df["KDJ.K"].ewm(com=2, adjust=False).mean()
      df["KDJ.J"] = 3 * df["KDJ.K"] - 2 * df["KDJ.D"]
  
      # 计算RSI指标
      delta = clos.diff()
      gain = delta.where(delta > 0, 0)
      loss = -delta.where(delta < 0, 0)
      avg_gain = gain.rolling(window=14).mean()
      avg_loss = loss.rolling(window=14).mean()
      rs = avg_gain / avg_loss
      df["RSI"] = 100 - (100 / (1 + rs))
  
      # 计算布林带指标
      df["BOLL.M"] = clos.rolling(window=20).mean()
      std = clos.rolling(window=20).std()
      df["BOLL.U"] = df["BOLL.M"] + 2 * std
      df["BOLL.L"] = df["BOLL.M"] - 2 * std

• mcp_aktools/__init__.py:111-119 (helper)

  Helper function for US stock historical data, used by stock_prices for 'us' market.

  def stock_us_daily(symbol, start_date="2025-01-01", period="daily"):
      dfs = ak.stock_us_daily(symbol=symbol)
      if dfs is None or dfs.empty:
          return None
      dfs.rename(columns={"date": "日期", "open": "开盘", "close": "收盘", "high": "最高", "low": "最低", "volume": "成交量"}, inplace=True)
      dfs["换手率"] = None
      dfs.index = pd.to_datetime(dfs["日期"], errors="coerce")
      return dfs[start_date:"2222-01-01"]