MCP Server for stock and crypto

Instructions

Implementation Reference

• mcp_aktools/__init__.py:48-69 (handler)

  The handler function that implements the core logic of the 'stock_info' tool. It fetches basic stock information from akshare APIs for specified market and symbol, with caching.
  def stock_info( symbol: str = field_symbol, market: str = field_market, ): markets = [ ["sh", ak.stock_individual_info_em], ["sz", ak.stock_individual_info_em], ["hk", ak.stock_hk_security_profile_em], ] for m in markets: if m[0] != market: continue all = ak_cache(m[1], symbol=symbol, ttl=43200) if all is None or all.empty: continue return all.to_string() info = ak_search(symbol, market) if info is not None: return info.to_string() return f"Not Found for {symbol}.{market}"
• mcp_aktools/__init__.py:44-47 (registration)

  The @mcp.tool decorator that registers the 'stock_info' tool with FastMCP, including title and description.
  @mcp.tool( title="获取股票信息", description="根据股票代码和市场获取股票基本信息, 不支持加密货币", )
• mcp_aktools/__init__.py:20-22 (schema)

  Pydantic Field definitions providing input schema for symbol and market parameters used in the stock_info tool.
  field_symbol = Field(description="股票代码") field_market = Field("sh", description="股票市场，仅支持: sh(上证), sz(深证), hk(港股), us(美股), 不支持加密货币")
• mcp_aktools/__init__.py:527-562 (helper)

  Helper function ak_search used by stock_info as fallback to search for stock info if direct fetch fails. It uses various ak.stock_info_* functions.
  def ak_search(symbol=None, keyword=None, market=None): markets = [ ["sh", ak.stock_info_a_code_name, "code", "name"], ["sh", ak.stock_info_sh_name_code, "证券代码", "证券简称"], ["sz", ak.stock_info_sz_name_code, "A股代码", "A股简称"], ["hk", ak.stock_hk_spot, "代码", "中文名称"], ["hk", ak.stock_hk_spot_em, "代码", "名称"], ["us", ak.get_us_stock_name, "symbol", "cname"], ["us", ak.get_us_stock_name, "symbol", "name"], ["sh", ak.fund_etf_spot_ths, "基金代码", "基金名称"], ["sz", ak.fund_etf_spot_ths, "基金代码", "基金名称"], ["sh", ak.fund_info_index_em, "基金代码", "基金名称"], ["sz", ak.fund_info_index_em, "基金代码", "基金名称"], ["sh", ak.fund_etf_spot_em, "代码", "名称"], ["sz", ak.fund_etf_spot_em, "代码", "名称"], ] for m in markets: if market and market != m[0]: continue all = ak_cache(m[1], ttl=86400, ttl2=86400*7) if all is None or all.empty: continue for _, v in all.iterrows(): code, name = str(v[m[2]]).upper(), str(v[m[3]]).upper() if symbol and symbol.upper() == code: return v if keyword and keyword.upper() in [code, name]: return v for _, v in all.iterrows() if keyword else []: name = str(v[m[3]]) if len(keyword) >= 4 and keyword in name: return v if name.startswith(keyword): return v return None
• mcp_aktools/__init__.py:564-580 (helper)

  Caching helper ak_cache used extensively in stock_info to cache akshare API responses.
  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