AkTools MCP Server

Instructions

Implementation Reference

• mcp_aktools/__init__.py:185-194 (handler)

  The main handler function for the 'stock_indicators_hk' tool. It is decorated with @mcp.tool for registration, takes a stock symbol as input, fetches financial analysis indicators from akshare for HK stocks with caching, and returns the top 15 rows as CSV.

  @mcp.tool(
      title="港股关键指标",
      description="获取港股市场的股票财务报告关键指标",
  )
  def stock_indicators_hk(
      symbol: str = field_symbol,
  ):
      dfs = ak_cache(ak.stock_financial_hk_analysis_indicator_em, symbol=symbol, indicator="报告期")
      keys = dfs.to_csv(index=False, float_format="%.3f").strip().split("\n")
      return "\n".join(keys[0:15])

• mcp_aktools/__init__.py:20-20 (schema)

  Pydantic Field definition for the 'symbol' parameter used in the tool's input schema, shared across multiple tools.

  field_symbol = Field(description="股票代码")

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

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

  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