China Stock MCP

Instructions

Implementation Reference

• src/china_stock_mcp/server.py:497-497 (registration)

  Tool registration decorator defining the name and description for get_macro_data.

  @mcp.tool(name="get_macro_data", description="获取宏观经济指标数据，支持多个指标")

• src/china_stock_mcp/server.py:498-584 (handler)

  Main handler function for get_macro_data tool. Parses indicators, fetches data using akshare functions via cached fetcher, cleans data, concatenates results, and formats output.

  def get_macro_data(    
      indicator: Annotated[
          str | list[
              Literal["money_supply", "gdp", "cpi", "pmi", "stock_summary"]
          ] | None,
          Field(
              description="要获取的宏观经济指标，可以是逗号分隔的字符串（例如: 'gdp,cpi'）或字符串列表（例如: ['gdp', 'cpi']）。支持的指标包括: money_supply, gdp, cpi, pmi, stock_summary。默认: 'gdp'"
          ),
      ] = "gdp",
      output_format: Annotated[
          Literal["json", "csv", "xml", "excel", "markdown", "html"],
          Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"),
      ] = "markdown"
      ) -> str:
          """获取宏观经济数据"""
  
          def _clean_macro_data(df: pd.DataFrame) -> pd.DataFrame:
              """
              通用数据清洗函数，删除全 null 列和全 null 行。
              """
              if df.empty:
                  return df
              # 删除所有列值都为 null 的列
              df = df.dropna(axis=1, how='all')
              # 删除所有行值都为 null 的行
              df = df.dropna(axis=0, how='all')
              return df
          
          macro_indicator_map = {
              "money_supply": ak.macro_china_money_supply,
              "gdp": ak.macro_china_gdp_yearly,
              "cpi": ak.macro_china_cpi_yearly,
              "pmi": ak.macro_china_pmi_yearly,
              "stock_summary": ak.macro_china_stock_market_cap,
          }
  
          @cached_data_fetch()
          def get_macro_data_fetcher(
              indicator_name: str, **kwargs: Any
          ) -> pd.DataFrame:
              fetch_func = macro_indicator_map.get(indicator_name)
              if fetch_func:
                  return fetch_func(**kwargs)
              else:
                  raise ValueError(f"不支持的宏观经济指标: {indicator_name}")
          
          if indicator is None:
              indicator_list = ["gdp"] # 默认指标
          elif isinstance(indicator, str):
              indicator_list = [
                  ind.strip()
                  for ind in indicator.split(",")
                  if ind.strip()
              ]
          else: # 假设是 list[Literal[...]]
              indicator_list = indicator
          
          # 过滤掉无效的指标
          valid_indicators = []
          for ind in indicator_list:
              if ind in macro_indicator_map:
                  valid_indicators.append(ind)
              else:
                  print(f"警告: 宏观经济指标 '{ind}' 不存在，将被忽略。")
          
          if not valid_indicators:
              print("警告: 未找到有效的宏观经济指标，将返回空数据。")
              return _format_dataframe_output(pd.DataFrame(), output_format)
  
          df_list = []
          for ind in valid_indicators:
              try:
                  temp_df = get_macro_data_fetcher(ind)
                  if temp_df is not None and not temp_df.empty:
                      temp_df = _clean_macro_data(temp_df)
                      temp_df['indicator'] = ind # 添加指标名称列
                      df_list.append(temp_df)
              except Exception as e:
                  print(f"获取宏观经济指标 '{ind}' 失败: {e}")
          
          if df_list:
              df = pd.concat(df_list, ignore_index=True)
          else:
              df = pd.DataFrame() # 如果没有获取到任何数据，返回空的DataFrame
              
          return _format_dataframe_output(df, output_format)

• src/china_stock_mcp/server.py:499-511 (schema)

  Input schema definitions using Pydantic Annotated and Field for indicator parameter (supports multiple macro types) and output_format.

  indicator: Annotated[
      str | list[
          Literal["money_supply", "gdp", "cpi", "pmi", "stock_summary"]
      ] | None,
      Field(
          description="要获取的宏观经济指标，可以是逗号分隔的字符串（例如: 'gdp,cpi'）或字符串列表（例如: ['gdp', 'cpi']）。支持的指标包括: money_supply, gdp, cpi, pmi, stock_summary。默认: 'gdp'"
      ),
  ] = "gdp",
  output_format: Annotated[
      Literal["json", "csv", "xml", "excel", "markdown", "html"],
      Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"),
  ] = "markdown"
  ) -> str:

• src/china_stock_mcp/server.py:535-543 (helper)

  Cached helper function to fetch data for a specific macro indicator using the mapped akshare functions.

  def get_macro_data_fetcher(
      indicator_name: str, **kwargs: Any
  ) -> pd.DataFrame:
      fetch_func = macro_indicator_map.get(indicator_name)
      if fetch_func:
          return fetch_func(**kwargs)
      else:
          raise ValueError(f"不支持的宏观经济指标: {indicator_name}")

• src/china_stock_mcp/server.py:514-525 (helper)

  Helper function to clean macro data by dropping all-null rows and columns.

  def _clean_macro_data(df: pd.DataFrame) -> pd.DataFrame:
      """
      通用数据清洗函数，删除全 null 列和全 null 行。
      """
      if df.empty:
          return df
      # 删除所有列值都为 null 的列
      df = df.dropna(axis=1, how='all')
      # 删除所有行值都为 null 的行
      df = df.dropna(axis=0, how='all')
      return df