A Share MCP

Instructions

Implementation Reference

• src/tools/financial_reports.py:31-37 (handler)

  MCP tool handler for 'get_profit_data'. Decorated with @app.tool(), it invokes the use case via run_tool_with_handling for caching/error handling.

  @app.tool()
  def get_profit_data(code: str, year: str, quarter: int, limit: int = 250, format: str = "markdown") -> str:
      """Quarterly profitability data."""
      return run_tool_with_handling(
          lambda: fetch_profit_data(active_data_source, code=code, year=year, quarter=quarter, limit=limit, format=format),
          context=f"get_profit_data:{code}:{year}Q{quarter}",
      )

• mcp_server.py:51-58 (registration)

  Main server file where register_financial_report_tools is called to register the financial reports tools, including 'get_profit_data'.

  register_stock_market_tools(app, active_data_source)
  register_financial_report_tools(app, active_data_source)
  register_index_tools(app, active_data_source)
  register_market_overview_tools(app, active_data_source)
  register_macroeconomic_tools(app, active_data_source)
  register_date_utils_tools(app, active_data_source)
  register_analysis_tools(app, active_data_source)
  register_helpers_tools(app)

• src/use_cases/financial_reports.py:22-27 (helper)

  Use case implementation: input validation, data fetching from datasource, and output formatting.

  def fetch_profit_data(data_source: FinancialDataSource, *, code: str, year: str, quarter: int, limit: int, format: str) -> str:
      validate_year(year)
      validate_quarter(quarter)
      validate_output_format(format)
      df = data_source.get_profit_data(code=code, year=year, quarter=quarter)
      return _format_financial_df(df, code=code, year=year, quarter=quarter, dataset="Profitability", format=format, limit=limit)

• src/baostock_data_source.py:413-415 (helper)

  Concrete data source method implementing get_profit_data by calling Baostock API via shared helper.

  def get_profit_data(self, code: str, year: str, quarter: int) -> pd.DataFrame:
      """Fetches quarterly profitability data using Baostock."""
      return _fetch_financial_data(bs.query_profit_data, "Profitability", code, year, quarter)

• src/baostock_data_source.py:26-64 (helper)

  Shared helper function in data source for fetching financial data from Baostock, handling login, errors, and DataFrame construction.

  def _fetch_financial_data(
      bs_query_func,
      data_type_name: str,
      code: str,
      year: str,
      quarter: int
  ) -> pd.DataFrame:
      logger.info(
          f"Fetching {data_type_name} data for {code}, year={year}, quarter={quarter}")
      try:
          with baostock_login_context():
              # Assuming all these functions take code, year, quarter
              rs = bs_query_func(code=code, year=year, quarter=quarter)
  
              if rs.error_code != '0':
                  logger.error(
                      f"Baostock API error ({data_type_name}) 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 {data_type_name} data found for {code}, {year}Q{quarter}. Baostock msg: {rs.error_msg}")
                  else:
                      raise DataSourceError(
                          f"Baostock API error fetching {data_type_name} 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 {data_type_name} data found for {code}, {year}Q{quarter} (empty result set from Baostock).")
                  raise NoDataFoundError(
                      f"No {data_type_name} data found for {code}, {year}Q{quarter} (empty result set).")
  
              result_df = pd.DataFrame(data_list, columns=rs.fields)
              logger.info(
                  f"Retrieved {len(result_df)} {data_type_name} records for {code}, {year}Q{quarter}.")
              return result_df