A-Share MCP Server

Instructions

Implementation Reference

• src/tools/financial_reports.py:79-87 (handler)

  The MCP tool handler function, decorated with @app.tool() for registration. It wraps the use case execution with standardized error handling, logging, and context.
  @app.tool() def get_performance_express_report(code: str, start_date: str, end_date: str, limit: int = 250, format: str = "markdown") -> str: """Performance express report within date range.""" return run_tool_with_handling( lambda: fetch_performance_express_report( active_data_source, code=code, start_date=start_date, end_date=end_date, limit=limit, format=format ), context=f"get_performance_express_report:{code}:{start_date}-{end_date}", )
• mcp_server.py:52-52 (registration)

  Invocation of the registration function that defines and registers the financial report tools, including get_performance_express_report, to the FastMCP app instance.
  register_financial_report_tools(app, active_data_source)
• src/use_cases/financial_reports.py:70-74 (helper)

  Use case function that fetches raw data from the data source interface and formats it into markdown or other output for the tool.
  def fetch_performance_express_report(data_source: FinancialDataSource, *, code: str, start_date: str, end_date: str, limit: int, format: str) -> str: validate_output_format(format) df = data_source.get_performance_express_report(code=code, start_date=start_date, end_date=end_date) meta = {"code": code, "start_date": start_date, "end_date": end_date, "dataset": "Performance Express"} return format_table_output(df, format=format, max_rows=limit, meta=meta)
• src/baostock_data_source.py:437-469 (helper)

  Concrete implementation in BaostockDataSource that queries the Baostock API for performance express reports and returns a pandas DataFrame.
  def get_performance_express_report(self, code: str, start_date: str, end_date: str) -> pd.DataFrame: """Fetches performance express reports (业绩快报) using Baostock.""" logger.info( f"Fetching Performance Express Report for {code} ({start_date} to {end_date})") try: with baostock_login_context(): rs = bs.query_performance_express_report( code=code, start_date=start_date, end_date=end_date) if rs.error_code != '0': logger.error( f"Baostock API error (Perf Express) 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 performance express report found for {code} in range {start_date}-{end_date}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching performance express report: {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 performance express report found for {code} in range {start_date}-{end_date} (empty result set).") raise NoDataFoundError( f"No performance express report found for {code} in range {start_date}-{end_date} (empty result set).") result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info( f"Retrieved {len(result_df)} performance express report records for {code}.") return result_df
• src/data_source_interface.py:158-160 (schema)

  Abstract method in the FinancialDataSource interface defining the expected signature for fetching performance express report data.
  @abstractmethod def get_performance_express_report(self, code: str, start_date: str, end_date: str) -> pd.DataFrame: pass