China Stock MCP

Instructions

Implementation Reference

• src/china_stock_mcp/server.py:663-690 (handler)

  The main handler function for the 'get_product_info' tool. It is decorated with @mcp.tool for registration. Fetches company product/business info using a fallback mechanism between 'ths' (ak.stock_zyjs_ths) and 'eastmoney' (ak.stock_zygc_em) data sources, handles empty DataFrames, and formats the output using _format_dataframe_output.

  @mcp.tool(name="get_product_info", description="获取公司主要产品/业务构成")
  def get_product_info(
      symbol: Annotated[str, Field(description="股票代码 (例如: '000001')")], 
      output_format: Annotated[
          Literal["json", "csv", "xml", "excel", "markdown", "html"],
          Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"),
      ] = "markdown"  
  ) -> str:
      """获取产品情况"""
      def get_product_info_fetcher(
          source:str,**kwargs: Any
      ) -> pd.DataFrame:
          if source == "ths":
              df = ak.stock_zyjs_ths(symbol)
          # elif data_source == "eastmoney":
          else:
              df = ak.stock_zygc_em(symbol)
          return df
    
      df = _fetch_data_with_fallback(
          fetch_func=get_product_info_fetcher,
          primary_source="ths",
          fallback_sources=["eastmoney"],
          symbol=symbol,
      )     
      if df.empty:
          df = pd.DataFrame()
      return _format_dataframe_output(df, output_format)

• src/china_stock_mcp/server.py:672-681 (helper)

  Inner helper function that fetches product data based on the source: 'ths' uses ak.stock_zyjs_ths, default (eastmoney) uses ak.stock_zygc_em.

  def get_product_info_fetcher(
      source:str,**kwargs: Any
  ) -> pd.DataFrame:
      if source == "ths":
          df = ak.stock_zyjs_ths(symbol)
      # elif data_source == "eastmoney":
      else:
          df = ak.stock_zygc_em(symbol)
      return df

• src/china_stock_mcp/server.py:15-71 (helper)

  Shared utility helper for fetching data with primary source and fallbacks, used by get_product_info.

  def _fetch_data_with_fallback(
      fetch_func: Callable[..., pd.DataFrame],
      primary_source: str,
      fallback_sources: List[str],
      **kwargs: Any,
  ) -> pd.DataFrame:
      """
      通用的数据源故障切换辅助函数。
      按优先级尝试数据源，直到获取到有效数据或所有数据源都失败。
  
      Args:
          fetch_func: 实际调用 akshare 或 akshare_one 获取数据的函数。
                      这个函数应该接受 'source' 参数，或者在内部处理 source 的映射。
          primary_source: 用户指定的首选数据源。
          fallback_sources: 备用数据源列表，按优先级排序。
          **kwargs: 传递给 fetch_func 的其他参数。
  
      Returns:
          pd.DataFrame: 获取到的数据。
  
      Raises:
          RuntimeError: 如果所有数据源都未能获取到有效数据。
      """
      if primary_source is None:
          return fetch_func(**kwargs)
      data_source_priority = [primary_source] + fallback_sources
      # 移除重复项并保持顺序
      seen = set()
      unique_data_source_priority = []
      for x in data_source_priority:
          if x not in seen:
              unique_data_source_priority.append(x)
              seen.add(x)
  
      df = None
      errors = []
  
      for current_source in unique_data_source_priority:
          try:
              # 假设 fetch_func 能够接受 source 参数
              # 或者 fetch_func 内部根据 kwargs 中的 source 参数进行逻辑判断
              temp_df = fetch_func(source=current_source, **kwargs)
              if temp_df is not None and not temp_df.empty:
                  print(f"成功从数据源 '{current_source}' 获取数据。")
                  df = temp_df
                  break
              else:
                  errors.append(f"数据源 '{current_source}' 返回空数据。")
          except Exception as e:
              errors.append(f"从数据源 '{current_source}' 获取数据失败: {str(e)}")
  
      if df is None or df.empty:
          raise RuntimeError(
              f"所有数据源都未能获取到有效数据。详细错误: {'; '.join(errors)}"
          )
  
      return df

• src/china_stock_mcp/server.py:89-118 (helper)

  Shared utility to format DataFrame output in various formats (json, csv, etc.), used by the handler.

  def _format_dataframe_output(
      df: pd.DataFrame,
      output_format: Literal["json", "csv", "xml", "excel", "markdown", "html"],
  ) -> str:
      """
      根据指定的格式格式化 DataFrame 输出。
      """
      if df.empty:
          return json.dumps([])
  
      if output_format == "json":
          return df.to_json(orient="records", force_ascii=False)
      elif output_format == "csv":
          return df.to_csv(index=False)
      elif output_format == "xml":
          return df.to_xml(index=False)
      elif output_format == "excel":
          # 使用 BytesIO 将 Excel 写入内存
          output = io.BytesIO()
          df.to_excel(output, index=False, engine='openpyxl')
          # 返回 base64 编码的二进制数据，或者直接返回字节流
          # 为了兼容性，这里尝试返回 utf-8 编码的字符串，但对于二进制文件，通常直接传输字节流更合适
          return output.getvalue().decode("utf-8", errors="ignore")
      elif output_format == "markdown":
          return df.to_markdown(index=False)
      elif output_format == "html":
          return df.to_html(index=False)
      else:
          return df.to_json(orient="records", force_ascii=False)