A-Share MCP Server

Instructions

    Fetches basic information for a given Chinese A-share stock.

    Args:
        code: The stock code in Baostock format (e.g., 'sh.600000', 'sz.000001').
        fields: Optional list to select specific columns from the available basic info
                (e.g., ['code', 'code_name', 'industry', 'listingDate']).
                If None or empty, returns all available basic info columns from Baostock.

    Returns:
        Basic stock information in the requested format.
    

Implementation Reference

• src/tools/stock_market.py:89-109 (handler)

  MCP tool handler decorated with @app.tool(). Logs invocation, delegates to use case fetch_stock_basic_info via run_tool_with_handling for shared error handling and formatting, returns str output.

  @app.tool()
  def get_stock_basic_info(code: str, fields: Optional[List[str]] = None, format: str = "markdown") -> str:
      """
      Fetches basic information for a given Chinese A-share stock.
  
      Args:
          code: The stock code in Baostock format (e.g., 'sh.600000', 'sz.000001').
          fields: Optional list to select specific columns from the available basic info
                  (e.g., ['code', 'code_name', 'industry', 'listingDate']).
                  If None or empty, returns all available basic info columns from Baostock.
  
      Returns:
          Basic stock information in the requested format.
      """
      logger.info(f"Tool 'get_stock_basic_info' called for {code} (fields={fields})")
      return run_tool_with_handling(
          lambda: fetch_stock_basic_info(
              active_data_source, code=code, fields=fields, format=format
          ),
          context=f"get_stock_basic_info:{code}",
      )

• src/data_source_interface.py:67-86 (schema)

  Abstract base class method defining the input/output schema (code: str -> pd.DataFrame) and expected behavior/raises for data source implementations.

  @abstractmethod
  def get_stock_basic_info(self, code: str) -> pd.DataFrame:
      """
      Fetches basic information for a given stock code.
  
      Args:
          code: The stock code (e.g., 'sh.600000', 'sz.000001').
  
      Returns:
          A pandas DataFrame containing the basic stock information.
          The structure and columns depend on the underlying data source.
          Typically contains info like name, industry, listing date, etc.
  
      Raises:
          LoginError: If login to the data source fails.
          NoDataFoundError: If no data is found for the query.
          DataSourceError: For other data source related errors.
          ValueError: If the input code is invalid.
      """
      pass

• mcp_server.py:51-51 (registration)

  Invocation of the registration function that adds the get_stock_basic_info tool (and others) to the FastMCP app instance.

  register_stock_market_tools(app, active_data_source)

• src/use_cases/stock_market.py:51-62 (helper)

  Orchestrates fetching raw DataFrame from data_source, applies output formatting via format_table_output, handles validation.

  def fetch_stock_basic_info(
      data_source: FinancialDataSource,
      *,
      code: str,
      fields: Optional[List[str]] = None,
      format: str = "markdown",
  ) -> str:
      validate_output_format(format)
      df = data_source.get_stock_basic_info(code=code, fields=fields)
      meta = {"code": code}
      return format_table_output(df, format=format, max_rows=df.shape[0] if df is not None else 0, meta=meta)

• src/baostock_data_source.py:262-324 (helper)

  Concrete Baostock implementation: calls bs.query_stock_basic, handles login/context/errors/pagination, builds DataFrame, supports field selection.

  def get_stock_basic_info(self, code: str, fields: Optional[List[str]] = None) -> pd.DataFrame:
      """Fetches basic stock information using Baostock."""
      logger.info(f"Fetching basic info for {code}")
      try:
          # Note: query_stock_basic doesn't seem to have a fields parameter in docs,
          # but we keep the signature consistent. It returns a fixed set.
          # We will use the `fields` argument post-query to select columns if needed.
          logger.debug(
              f"Requesting basic info for {code}. Optional fields requested: {fields}")
  
          with baostock_login_context():
              # Example: Fetch basic info; adjust API call if needed based on baostock docs
              # rs = bs.query_stock_basic(code=code, code_name=code_name) # If supporting name lookup
              rs = bs.query_stock_basic(code=code)
  
              if rs.error_code != '0':
                  logger.error(
                      f"Baostock API error (Basic Info) for {code}: {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 basic info found for {code}. Baostock msg: {rs.error_msg}")
                  else:
                      raise DataSourceError(
                          f"Baostock API error fetching basic info: {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 basic info found for {code} (empty result set from Baostock).")
                  raise NoDataFoundError(
                      f"No basic info found for {code} (empty result set).")
  
              # Crucial: Use rs.fields for column names
              result_df = pd.DataFrame(data_list, columns=rs.fields)
              logger.info(
                  f"Retrieved basic info for {code}. Columns: {result_df.columns.tolist()}")
  
              # Optional: Select subset of columns if `fields` argument was provided
              if fields:
                  available_cols = [
                      col for col in fields if col in result_df.columns]
                  if not available_cols:
                      raise ValueError(
                          f"None of the requested fields {fields} are available in the basic info result.")
                  logger.debug(
                      f"Selecting columns: {available_cols} from basic info for {code}")
                  result_df = result_df[available_cols]
  
              return result_df
  
      except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e:
          logger.warning(
              f"Caught known error fetching basic info for {code}: {type(e).__name__}")
          raise e
      except Exception as e:
          logger.exception(
              f"Unexpected error fetching basic info for {code}: {e}")
          raise DataSourceError(
              f"Unexpected error fetching basic info for {code}: {e}")