A-Share MCP Server

Instructions

    Fetches dividend information for a given stock code and year.

    Args:
        code: The stock code in Baostock format (e.g., 'sh.600000', 'sz.000001').
        year: The year to query (e.g., '2023').
        year_type: Type of year. Valid options (from Baostock):
                     'report': Announcement year (预案公告年份)
                     'operate': Ex-dividend year (除权除息年份)
                   Defaults to 'report'.

    Returns:
        Dividend records table.
    

Implementation Reference

• src/tools/stock_market.py:111-138 (handler)

  The primary MCP tool handler for 'get_dividend_data'. Decorated with @app.tool(), it logs the call, invokes the fetch_dividend_data use case wrapped in error handling, and returns formatted results.

  @app.tool()
  def get_dividend_data(code: str, year: str, year_type: str = "report", limit: int = 250, format: str = "markdown") -> str:
      """
      Fetches dividend information for a given stock code and year.
  
      Args:
          code: The stock code in Baostock format (e.g., 'sh.600000', 'sz.000001').
          year: The year to query (e.g., '2023').
          year_type: Type of year. Valid options (from Baostock):
                       'report': Announcement year (预案公告年份)
                       'operate': Ex-dividend year (除权除息年份)
                     Defaults to 'report'.
  
      Returns:
          Dividend records table.
      """
      logger.info(f"Tool 'get_dividend_data' called for {code}, year={year}, year_type={year_type}")
      return run_tool_with_handling(
          lambda: fetch_dividend_data(
              active_data_source,
              code=code,
              year=year,
              year_type=year_type,
              limit=limit,
              format=format,
          ),
          context=f"get_dividend_data:{code}:{year}",
      )

• mcp_server.py:51-51 (registration)

  The registration call in the main MCP server that registers all stock market tools, including 'get_dividend_data', via the register_stock_market_tools function.

  register_stock_market_tools(app, active_data_source)

• src/use_cases/stock_market.py:64-79 (helper)

  Helper use case function that performs input validation and formats the dividend DataFrame returned from the data source into the requested output format.

  def fetch_dividend_data(
      data_source: FinancialDataSource,
      *,
      code: str,
      year: str,
      year_type: str = "report",
      limit: int = 250,
      format: str = "markdown",
  ) -> str:
      validate_year(year)
      validate_year_type(year_type)
      validate_output_format(format)
  
      df = data_source.get_dividend_data(code=code, year=year, year_type=year_type)
      meta = {"code": code, "year": year, "year_type": year_type}
      return format_table_output(df, format=format, max_rows=limit, meta=meta)

• src/baostock_data_source.py:325-368 (helper)

  The core data fetching implementation in BaostockDataSource class, which queries the Baostock API (bs.query_dividend_data) to retrieve raw dividend data as a pandas DataFrame.

  def get_dividend_data(self, code: str, year: str, year_type: str = "report") -> pd.DataFrame:
      """Fetches dividend information using Baostock."""
      logger.info(
          f"Fetching dividend data for {code}, year={year}, year_type={year_type}")
      try:
          with baostock_login_context():
              rs = bs.query_dividend_data(
                  code=code, year=year, yearType=year_type)
  
              if rs.error_code != '0':
                  logger.error(
                      f"Baostock API error (Dividend) 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 dividend data found for {code} and year {year}. Baostock msg: {rs.error_msg}")
                  else:
                      raise DataSourceError(
                          f"Baostock API error fetching dividend data: {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 dividend data found for {code}, year {year} (empty result set from Baostock).")
                  raise NoDataFoundError(
                      f"No dividend data found for {code}, year {year} (empty result set).")
  
              result_df = pd.DataFrame(data_list, columns=rs.fields)
              logger.info(
                  f"Retrieved {len(result_df)} dividend records for {code}, year {year}.")
              return result_df
  
      except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e:
          logger.warning(
              f"Caught known error fetching dividend data for {code}: {type(e).__name__}")
          raise e
      except Exception as e:
          logger.exception(
              f"Unexpected error fetching dividend data for {code}: {e}")
          raise DataSourceError(
              f"Unexpected error fetching dividend data for {code}: {e}")