A-Share MCP Server

Instructions

Implementation Reference

• src/tools/indices.py:33-39 (handler)

  MCP tool handler and registration for 'get_sz50_stocks'. Delegates to use case 'fetch_index_constituents' via run_tool_with_handling for execution and error handling.

  @app.tool()
  def get_sz50_stocks(date: Optional[str] = None, limit: int = 250, format: str = "markdown") -> str:
      """SZSE 50 constituents."""
      return run_tool_with_handling(
          lambda: fetch_index_constituents(active_data_source, index="sz50", date=date, limit=limit, format=format),
          context="get_sz50_stocks",
      )

• src/use_cases/indices.py:25-35 (helper)

  Use case helper 'fetch_index_constituents' that handles sz50 index by calling data_source.get_sz50_stocks(), validates input, and formats the DataFrame output.

  def fetch_index_constituents(data_source: FinancialDataSource, *, index: str, date: Optional[str], limit: int, format: str) -> str:
      validate_output_format(format)
      key = validate_index_key(index, INDEX_MAP)
      if key == "hs300":
          df = data_source.get_hs300_stocks(date=date)
      elif key == "sz50":
          df = data_source.get_sz50_stocks(date=date)
      else:
          df = data_source.get_zz500_stocks(date=date)
      meta = {"index": key, "as_of": date or "latest"}
      return format_table_output(df, format=format, max_rows=limit, meta=meta)

• src/baostock_data_source.py:570-573 (helper)

  Data source implementation of get_sz50_stocks in BaostockDataSource, which fetches constituents using Baostock's query_sz50_stocks API via shared helper.

  def get_sz50_stocks(self, date: Optional[str] = None) -> pd.DataFrame:
      """Fetches SZSE 50 index constituents using Baostock."""
      return _fetch_index_constituent_data(bs.query_sz50_stocks, "SZSE 50", date)

• src/data_source_interface.py:196-197 (schema)

  Abstract method definition in FinancialDataSource interface specifying the get_sz50_stocks signature.

  def get_sz50_stocks(self, date: Optional[str] = None) -> pd.DataFrame:
      pass

• src/baostock_data_source.py:78-123 (helper)

  Shared helper function _fetch_index_constituent_data used by get_sz50_stocks to query Baostock API, handle errors, and return DataFrame.

  def _fetch_index_constituent_data(
      bs_query_func,
      index_name: str,
      date: Optional[str] = None
  ) -> pd.DataFrame:
      logger.info(
          f"Fetching {index_name} constituents for date={date or 'latest'}")
      try:
          with baostock_login_context():
              # date is optional, defaults to latest
              rs = bs_query_func(date=date)
  
              if rs.error_code != '0':
                  logger.error(
                      f"Baostock API error ({index_name} Constituents) for date {date}: {rs.error_msg} (code: {rs.error_code})")
                  if "no record found" in rs.error_msg.lower() or rs.error_code == '10002':
                      raise NoDataFoundError(
                          f"No {index_name} constituent data found for date {date}. Baostock msg: {rs.error_msg}")
                  else:
                      raise DataSourceError(
                          f"Baostock API error fetching {index_name} constituents: {rs.error_msg} (code: {rs.error_code})")
  
              data_list = []
              while rs.next():
                  data_list.append(rs.get_row_data())
  
              if not data_list:
                  logger.warning(
                      f"No {index_name} constituent data found for date {date} (empty result set).")
                  raise NoDataFoundError(
                      f"No {index_name} constituent data found for date {date} (empty result set).")
  
              result_df = pd.DataFrame(data_list, columns=rs.fields)
              logger.info(
                  f"Retrieved {len(result_df)} {index_name} constituents for date {date or 'latest'}.")
              return result_df
  
      except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e:
          logger.warning(
              f"Caught known error fetching {index_name} constituents for date {date}: {type(e).__name__}")
          raise e
      except Exception as e:
          logger.exception(
              f"Unexpected error fetching {index_name} constituents for date {date}: {e}")
          raise DataSourceError(
              f"Unexpected error fetching {index_name} constituents for date {date}: {e}")